options refactor (#1088)

* refactor config loading

* wip

* move autocert to its own config source

* refactor options updaters

* fix stuttering

* fix autocert validate check

internal/autocert/manager.go

@@ -0,0 +1,198 @@
+// Package autocert implements automatic management of TLS certificates.
+package autocert
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sort"
+	"sync"
+
+	"github.com/caddyserver/certmagic"
+
+	"github.com/pomerium/pomerium/config"
+	"github.com/pomerium/pomerium/internal/httputil"
+	"github.com/pomerium/pomerium/internal/log"
+)
+
+// Manager manages TLS certificates.
+type Manager struct {
+	src config.Source
+
+	mu        sync.RWMutex
+	config    *config.Config
+	certmagic *certmagic.Config
+	acmeMgr   *certmagic.ACMEManager
+	srv       *http.Server
+
+	config.ChangeDispatcher
+}
+
+// New creates a new autocert manager.
+func New(src config.Source) (*Manager, error) {
+	mgr := &Manager{
+		src:       src,
+		certmagic: certmagic.NewDefault(),
+	}
+	err := mgr.update(src.GetConfig())
+	if err != nil {
+		return nil, err
+	}
+	mgr.src.OnConfigChange(func(cfg *config.Config) {
+		err := mgr.update(cfg)
+		if err != nil {
+			log.Error().Err(err).Msg("autocert: error updating config")
+			return
+		}
+
+		mgr.Trigger(mgr.GetConfig())
+	})
+	return mgr, nil
+}
+
+func (mgr *Manager) getCertMagicConfig(options *config.Options) (*certmagic.Config, error) {
+	mgr.certmagic.MustStaple = options.AutocertOptions.MustStaple
+	mgr.certmagic.OnDemand = nil // disable on-demand
+	mgr.certmagic.Storage = &certmagic.FileStorage{Path: options.AutocertOptions.Folder}
+	// add existing certs to the cache, and staple OCSP
+	for _, cert := range options.Certificates {
+		if err := mgr.certmagic.CacheUnmanagedTLSCertificate(cert, nil); err != nil {
+			return nil, fmt.Errorf("config: failed caching cert: %w", err)
+		}
+	}
+	acmeMgr := certmagic.NewACMEManager(mgr.certmagic, certmagic.DefaultACME)
+	acmeMgr.Agreed = true
+	if options.AutocertOptions.UseStaging {
+		acmeMgr.CA = certmagic.LetsEncryptStagingCA
+	}
+	acmeMgr.DisableTLSALPNChallenge = true
+	mgr.certmagic.Issuer = acmeMgr
+	mgr.acmeMgr = acmeMgr
+
+	return mgr.certmagic, nil
+}
+
+func (mgr *Manager) update(cfg *config.Config) error {
+	cfg = cfg.Clone()
+
+	mgr.mu.Lock()
+	defer mgr.mu.Unlock()
+	defer func() { mgr.config = cfg }()
+
+	mgr.updateServer(cfg)
+	return mgr.updateAutocert(cfg)
+}
+
+func (mgr *Manager) updateAutocert(cfg *config.Config) error {
+	if !cfg.Options.AutocertOptions.Enable {
+		return nil
+	}
+
+	cm, err := mgr.getCertMagicConfig(cfg.Options)
+	if err != nil {
+		return err
+	}
+
+	for _, domain := range sourceHostnames(cfg) {
+		cert, err := cm.CacheManagedCertificate(domain)
+		if err != nil {
+			log.Info().Str("domain", domain).Msg("obtaining certificate")
+			err = cm.ObtainCert(context.Background(), domain, false)
+			if err != nil {
+				return fmt.Errorf("autocert: failed to obtain client certificate: %w", err)
+			}
+			cert, err = cm.CacheManagedCertificate(domain)
+		}
+		if err == nil && cert.NeedsRenewal(cm) {
+			log.Info().Str("domain", domain).Msg("renewing certificate")
+			err = cm.RenewCert(context.Background(), domain, false)
+			if err != nil {
+				return fmt.Errorf("autocert: failed to renew client certificate: %w", err)
+			}
+			cert, err = cm.CacheManagedCertificate(domain)
+		}
+		if err == nil {
+			cfg.Options.Certificates = append(cfg.Options.Certificates, cert.Certificate)
+		} else {
+			log.Error().Err(err).Msg("autocert: failed to obtain client certificate")
+		}
+	}
+
+	return nil
+}
+
+func (mgr *Manager) updateServer(cfg *config.Config) {
+	if mgr.srv != nil {
+		// nothing to do if the address hasn't changed
+		if mgr.srv.Addr == cfg.Options.HTTPRedirectAddr {
+			return
+		}
+		// close immediately, don't care about the error
+		_ = mgr.srv.Close()
+		mgr.srv = nil
+	}
+
+	if cfg.Options.HTTPRedirectAddr == "" {
+		return
+	}
+
+	redirect := httputil.RedirectHandler()
+
+	hsrv := &http.Server{
+		Addr: cfg.Options.HTTPRedirectAddr,
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if mgr.handleHTTPChallenge(w, r) {
+				return
+			}
+			redirect.ServeHTTP(w, r)
+		}),
+	}
+	go func() {
+		log.Info().Str("addr", hsrv.Addr).Msg("starting http redirect server")
+		err := hsrv.ListenAndServe()
+		if err != nil {
+			log.Error().Err(err).Msg("failed to run http redirect server")
+		}
+	}()
+	mgr.srv = hsrv
+}
+
+func (mgr *Manager) handleHTTPChallenge(w http.ResponseWriter, r *http.Request) bool {
+	mgr.mu.RLock()
+	acmeMgr := mgr.acmeMgr
+	mgr.mu.RUnlock()
+	if acmeMgr == nil {
+		return false
+	}
+	return acmeMgr.HandleHTTPChallenge(w, r)
+}
+
+// GetConfig gets the config.
+func (mgr *Manager) GetConfig() *config.Config {
+	mgr.mu.RLock()
+	defer mgr.mu.RUnlock()
+
+	return mgr.config
+}
+
+func sourceHostnames(cfg *config.Config) []string {
+	if len(cfg.Options.Policies) == 0 {
+		return nil
+	}
+
+	dedupe := map[string]struct{}{}
+	for _, p := range cfg.Options.Policies {
+		dedupe[p.Source.Hostname()] = struct{}{}
+	}
+	if cfg.Options.AuthenticateURL != nil {
+		dedupe[cfg.Options.AuthenticateURL.Hostname()] = struct{}{}
+	}
+
+	var h []string
+	for k := range dedupe {
+		h = append(h, k)
+	}
+	sort.Strings(h)
+
+	return h
+}

internal/cmd/pomerium/pomerium.go

@@ -11,8 +11,6 @@ import (
 	"sync"
 	"syscall"
 
-	"github.com/pomerium/pomerium/internal/telemetry"
-
 	envoy_service_auth_v2 "github.com/envoyproxy/go-control-plane/envoy/service/auth/v2"
 	"golang.org/x/sync/errgroup"
 
@@ -20,10 +18,12 @@ import (
 	"github.com/pomerium/pomerium/authorize"
 	"github.com/pomerium/pomerium/cache"
 	"github.com/pomerium/pomerium/config"
+	"github.com/pomerium/pomerium/internal/autocert"
 	"github.com/pomerium/pomerium/internal/controlplane"
 	"github.com/pomerium/pomerium/internal/envoy"
 	"github.com/pomerium/pomerium/internal/httputil"
 	"github.com/pomerium/pomerium/internal/log"
+	"github.com/pomerium/pomerium/internal/telemetry"
 	"github.com/pomerium/pomerium/internal/telemetry/metrics"
 	"github.com/pomerium/pomerium/internal/telemetry/trace"
 	"github.com/pomerium/pomerium/internal/urlutil"
@@ -33,31 +33,36 @@ import (
 
 // Run runs the main pomerium application.
 func Run(ctx context.Context, configFile string) error {
-	opt, err := config.NewOptionsFromConfig(configFile)
+	var src config.Source
+
+	src, err := config.NewFileOrEnvironmentSource(configFile)
 	if err != nil {
 		return err
 	}
-	var optionsUpdaters []config.OptionsUpdater
+
+	src, err = autocert.New(src)
+	if err != nil {
+		return err
+	}
+
+	cfg := src.GetConfig()
 
 	log.Info().Str("version", version.FullVersion()).Msg("cmd/pomerium")
 
-	if err := setupMetrics(ctx, opt); err != nil {
+	if err := setupMetrics(ctx, cfg.Options); err != nil {
 		return err
 	}
-	if err := setupTracing(ctx, opt); err != nil {
+	if err := setupTracing(ctx, cfg.Options); err != nil {
 		return err
 	}
 
 	// setup the control plane
-	controlPlane, err := controlplane.NewServer(opt.Services)
+	controlPlane, err := controlplane.NewServer(cfg.Options.Services)
 	if err != nil {
 		return fmt.Errorf("error creating control plane: %w", err)
 	}
-	optionsUpdaters = append(optionsUpdaters, controlPlane)
-	err = controlPlane.UpdateOptions(*opt)
-	if err != nil {
-		return fmt.Errorf("error updating control plane options: %w", err)
-	}
+	src.OnConfigChange(controlPlane.OnConfigChange)
+	controlPlane.OnConfigChange(cfg)
 
 	_, grpcPort, _ := net.SplitHostPort(controlPlane.GRPCListener.Addr().String())
 	_, httpPort, _ := net.SplitHostPort(controlPlane.HTTPListener.Addr().String())
@@ -66,36 +71,33 @@ func Run(ctx context.Context, configFile string) error {
 	log.Info().Str("port", httpPort).Msg("HTTP server started")
 
 	// create envoy server
-	envoyServer, err := envoy.NewServer(opt, grpcPort, httpPort)
+	envoyServer, err := envoy.NewServer(cfg.Options, grpcPort, httpPort)
 	if err != nil {
 		return fmt.Errorf("error creating envoy server: %w", err)
 	}
 
 	// add services
-	if err := setupAuthenticate(opt, controlPlane, &optionsUpdaters); err != nil {
+	if err := setupAuthenticate(src, cfg, controlPlane); err != nil {
 		return err
 	}
 	var authorizeServer *authorize.Authorize
-	if config.IsAuthorize(opt.Services) {
-		authorizeServer, err = setupAuthorize(opt, controlPlane, &optionsUpdaters)
+	if config.IsAuthorize(cfg.Options.Services) {
+		authorizeServer, err = setupAuthorize(src, cfg, controlPlane)
 		if err != nil {
 			return err
 		}
 	}
 	var cacheServer *cache.Cache
-	if config.IsCache(opt.Services) {
-		cacheServer, err = setupCache(opt, controlPlane)
+	if config.IsCache(cfg.Options.Services) {
+		cacheServer, err = setupCache(cfg.Options, controlPlane)
 		if err != nil {
 			return err
 		}
 	}
-	if err := setupProxy(opt, controlPlane); err != nil {
+	if err := setupProxy(cfg.Options, controlPlane); err != nil {
 		return err
 	}
 
-	// start the config change listener
-	go config.WatchChanges(configFile, opt, optionsUpdaters)
-
 	ctx, cancel := context.WithCancel(ctx)
 	go func(ctx context.Context) {
 		ch := make(chan os.Signal, 2)
@@ -132,21 +134,21 @@ func Run(ctx context.Context, configFile string) error {
 	return eg.Wait()
 }
 
-func setupAuthenticate(opt *config.Options, controlPlane *controlplane.Server, optionsUpdaters *[]config.OptionsUpdater) error {
-	if !config.IsAuthenticate(opt.Services) {
+func setupAuthenticate(src config.Source, cfg *config.Config, controlPlane *controlplane.Server) error {
+	if !config.IsAuthenticate(cfg.Options.Services) {
 		return nil
 	}
 
-	svc, err := authenticate.New(*opt)
+	svc, err := authenticate.New(cfg.Options)
 	if err != nil {
 		return fmt.Errorf("error creating authenticate service: %w", err)
 	}
-	*optionsUpdaters = append(*optionsUpdaters, svc)
-	err = svc.UpdateOptions(*opt)
+	src.OnConfigChange(svc.OnConfigChange)
+	svc.OnConfigChange(cfg)
 	if err != nil {
 		return fmt.Errorf("error updating authenticate options: %w", err)
 	}
-	host := urlutil.StripPort(opt.GetAuthenticateURL().Host)
+	host := urlutil.StripPort(cfg.Options.GetAuthenticateURL().Host)
 	sr := controlPlane.HTTPRouter.Host(host).Subrouter()
 	svc.Mount(sr)
 	log.Info().Str("host", host).Msg("enabled authenticate service")
@@ -154,20 +156,16 @@ func setupAuthenticate(opt *config.Options, controlPlane *controlplane.Server, o
 	return nil
 }
 
-func setupAuthorize(opt *config.Options, controlPlane *controlplane.Server, optionsUpdaters *[]config.OptionsUpdater) (*authorize.Authorize, error) {
-	svc, err := authorize.New(*opt)
+func setupAuthorize(src config.Source, cfg *config.Config, controlPlane *controlplane.Server) (*authorize.Authorize, error) {
+	svc, err := authorize.New(cfg.Options)
 	if err != nil {
 		return nil, fmt.Errorf("error creating authorize service: %w", err)
 	}
 	envoy_service_auth_v2.RegisterAuthorizationServer(controlPlane.GRPCServer, svc)
 
 	log.Info().Msg("enabled authorize service")
-
-	*optionsUpdaters = append(*optionsUpdaters, svc)
-	err = svc.UpdateOptions(*opt)
-	if err != nil {
-		return nil, fmt.Errorf("error updating authorize options: %w", err)
-	}
+	src.OnConfigChange(svc.OnConfigChange)
+	svc.OnConfigChange(cfg)
 	return svc, nil
 }
 

internal/controlplane/server.go

@@ -141,17 +141,16 @@ func (srv *Server) Run(ctx context.Context) error {
 	return eg.Wait()
 }
 
-// UpdateOptions updates the pomerium config options.
-func (srv *Server) UpdateOptions(options config.Options) error {
+// OnConfigChange updates the pomerium config options.
+func (srv *Server) OnConfigChange(cfg *config.Config) {
 	select {
 	case <-srv.configUpdated:
 	default:
 	}
 	prev := srv.currentConfig.Load()
 	srv.currentConfig.Store(versionedOptions{
-		Options: options,
+		Options: *cfg.Options,
 		version: prev.version + 1,
 	})
 	srv.configUpdated <- struct{}{}
-	return nil
 }