// Command updater-master serves the package-updater UI and accepts agent // connections. package main import ( "context" "crypto/tls" "errors" "fmt" "net" "net/http" "os" "os/signal" "path/filepath" "strings" "syscall" "time" "github.com/shcizo/package-updater/internal/logging" "github.com/shcizo/package-updater/internal/master/config" "github.com/shcizo/package-updater/internal/master/events" "github.com/shcizo/package-updater/internal/master/hub" "github.com/shcizo/package-updater/internal/master/jobs" "github.com/shcizo/package-updater/internal/master/poller" "github.com/shcizo/package-updater/internal/master/registry" "github.com/shcizo/package-updater/internal/master/store" "github.com/shcizo/package-updater/internal/master/web" "github.com/shcizo/package-updater/internal/master/webauth" "github.com/shcizo/package-updater/internal/master/webhook" "github.com/shcizo/package-updater/internal/pin" ) // Build info is injected at link time via -ldflags="-X main.version=...". var version = "dev" func main() { if err := run(); err != nil { fmt.Fprintln(os.Stderr, "fatal:", err) os.Exit(1) } } // pollFirstDelay is how long the master runs before its first registry // polling cycle, so agents have reported their inventory. const pollFirstDelay = time.Minute // pollWorkers is how many registry lookups run at once. const pollWorkers = 4 // registryCredentialsDir returns the directory the registry client reads // credentials from (DOCKER_CONFIG, else ~/.docker), whether a config.json is // in it, and the error from opening it when it exists but cannot be read (for // example because the master runs as another user than the file owner). The // file is opened and closed, never read; only the path is logged. func registryCredentialsDir() (dir string, found bool, readErr error) { dir = os.Getenv("DOCKER_CONFIG") if dir == "" { home, err := os.UserHomeDir() if err != nil { return "", false, nil } dir = filepath.Join(home, ".docker") } path := filepath.Join(dir, "config.json") if _, err := os.Stat(path); err != nil { return dir, false, nil } f, err := os.Open(path) if err != nil { return dir, true, err } _ = f.Close() return dir, true, nil } func run() error { cfg, err := config.Load() if err != nil { return err } logger := logging.New(cfg.LogLevel) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DatabaseURL) if err != nil { return err } defer st.Close() key, err := st.LoadOrCreateIdentity(ctx, pin.GenerateKey) if err != nil { return err } cert, err := pin.Certificate(key) if err != nil { return err } fingerprint, err := pin.Fingerprint(key.Public()) if err != nil { return err } logger.Info("master_starting", "version", version, "master_key", fingerprint) broker := events.NewBroker() agentHub := hub.New(st, broker, logger) jobService := jobs.New(st, agentHub, broker, cfg.AgentImage, logger) provider, err := webauth.NewOIDCProvider(ctx, webauth.OIDCConfig{ Issuer: cfg.OIDCIssuer, ClientID: cfg.OIDCClientID, ClientSecret: cfg.OIDCClientSecret, RedirectURL: cfg.PublicURL + "/auth/callback", }) if err != nil { return err } auth := webauth.New(provider, st, cfg.AllowedGroup, strings.HasPrefix(cfg.PublicURL, "https://"), logger) credDir, credFound, credReadErr := registryCredentialsDir() logger.Info("registry_credentials", "docker_config", credDir, "config_json_found", credFound, "config_json_readable", credFound && credReadErr == nil) if credReadErr != nil { logger.Warn("registry_credentials_unreadable", "docker_config", credDir, "error", credReadErr) } registryPoller := poller.New(st, jobService, registry.New("package-updater-master/"+version, logger), broker, poller.Config{ Interval: cfg.PollInterval, FirstDelay: pollFirstDelay, Workers: pollWorkers, }, logger) ui, err := web.New(st, agentHub, jobService, broker, registryPoller, web.Config{ AgentURL: cfg.AgentURL, MasterKey: fingerprint, AgentImage: cfg.AgentImage, PollInterval: cfg.PollInterval, Version: version, }, logger) if err != nil { return err } mux := http.NewServeMux() mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { if err := st.Ping(r.Context()); err != nil { http.Error(w, "database unreachable", http.StatusServiceUnavailable) return } _, _ = w.Write([]byte("ok")) }) // The webhook does its own bearer-token check, so it sits outside the // OIDC login like /healthz. mux.Handle("POST /update", webhook.New(st, jobService, broker, webhook.Config{ Token: cfg.UpdaterAPIKey, Timeout: cfg.UpdateTimeout, PublicURL: cfg.PublicURL, }, logger)) if cfg.UpdaterAPIKey == "" { logger.Warn("webhook_disabled", "reason", "UPDATER_API_KEY not set") } mux.HandleFunc("GET /auth/login", auth.Login) mux.HandleFunc("GET /auth/callback", auth.Callback) mux.Handle("POST /auth/logout", auth.Require(http.HandlerFunc(auth.Logout))) ui.Register(mux, auth.Require) // BaseContext ties every request context to the signal context, so on // shutdown the long-lived SSE handlers and agent connections return // instead of holding Shutdown until its timeout. baseContext := func(net.Listener) context.Context { return ctx } // No WriteTimeout on either server: SSE streams, agent connections and // POST /update (which waits up to UPDATE_TIMEOUT) are long-lived by design. webSrv := &http.Server{ Addr: ":" + cfg.Port, Handler: mux, ReadHeaderTimeout: 10 * time.Second, BaseContext: baseContext, } agentMux := http.NewServeMux() agentMux.Handle("GET /agent", agentHub) agentSrv := &http.Server{ Addr: ":" + cfg.AgentPort, Handler: agentMux, ReadHeaderTimeout: 10 * time.Second, TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13}, BaseContext: baseContext, } pollCtx, stopPolling := context.WithCancel(ctx) defer stopPolling() pollDone := make(chan struct{}) go func() { defer close(pollDone) registryPoller.Run(pollCtx) }() errCh := make(chan error, 2) go func() { errCh <- webSrv.ListenAndServe() }() go func() { errCh <- agentSrv.ListenAndServeTLS("", "") }() logger.Info("listening", "web", webSrv.Addr, "agents", agentSrv.Addr) var serveErr error select { case <-ctx.Done(): logger.Info("shutdown_signal") case err := <-errCh: if !errors.Is(err, http.ErrServerClosed) { serveErr = err } } shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() _ = webSrv.Shutdown(shutdownCtx) _ = agentSrv.Shutdown(shutdownCtx) // The poller writes to the store, which is closed when run returns. stopPolling() <-pollDone return serveErr }