diff --git a/Dockerfile b/Dockerfile index 114a1a4..8305278 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,7 +8,7 @@ COPY . . ARG VERSION=dev ARG COMMIT=unknown ARG BUILD_TIME=unknown -RUN for cmd in server:package-updater agent:updater-agent; do \ +RUN for cmd in server:package-updater master:updater-master agent:updater-agent; do \ CGO_ENABLED=0 go build \ -ldflags="-s -w \ -X main.version=${VERSION} \ diff --git a/cmd/master/main.go b/cmd/master/main.go new file mode 100644 index 0000000..34e0de7 --- /dev/null +++ b/cmd/master/main.go @@ -0,0 +1,137 @@ +// 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" + "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/store" + "github.com/shcizo/package-updater/internal/master/web" + "github.com/shcizo/package-updater/internal/master/webauth" + "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) + } +} + +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) + + 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) + + ui, err := web.New(st, agentHub, broker, web.Config{ + AgentURL: cfg.AgentURL, MasterKey: fingerprint, AgentImage: cfg.AgentImage, + }, 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")) + }) + 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 and agent connections + // 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, + } + + errCh := make(chan error, 2) + go func() { errCh <- webSrv.ListenAndServe() }() + go func() { errCh <- agentSrv.ListenAndServeTLS("", "") }() + logger.Info("listening", "web", webSrv.Addr, "agents", agentSrv.Addr) + + select { + case <-ctx.Done(): + logger.Info("shutdown_signal") + case err := <-errCh: + if !errors.Is(err, http.ErrServerClosed) { + return err + } + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = webSrv.Shutdown(shutdownCtx) + _ = agentSrv.Shutdown(shutdownCtx) + return nil +} diff --git a/internal/master/config/config.go b/internal/master/config/config.go new file mode 100644 index 0000000..0739ec1 --- /dev/null +++ b/internal/master/config/config.go @@ -0,0 +1,68 @@ +// Package config loads the master's configuration from environment variables. +package config + +import ( + "fmt" + "os" + "strings" +) + +// Config holds the master's runtime configuration. +type Config struct { + DatabaseURL string + PublicURL string // where people reach the UI, without trailing slash + AgentURL string // what agents dial, shown in the add-agent snippet + AgentImage string + OIDCIssuer string + OIDCClientID string + OIDCClientSecret string + AllowedGroup string + Port string + AgentPort string + LogLevel string +} + +// Load reads and validates the master configuration, failing fast on +// anything missing so a half-configured master never starts. +func Load() (*Config, error) { + cfg := &Config{ + DatabaseURL: os.Getenv("DATABASE_URL"), + PublicURL: strings.TrimSuffix(os.Getenv("PUBLIC_URL"), "/"), + AgentURL: os.Getenv("AGENT_URL"), + AgentImage: getenvDefault("AGENT_IMAGE", "gitea.shcizo.se/shcizo/package-updater:latest"), + OIDCIssuer: os.Getenv("OIDC_ISSUER"), + OIDCClientID: os.Getenv("OIDC_CLIENT_ID"), + OIDCClientSecret: os.Getenv("OIDC_CLIENT_SECRET"), + AllowedGroup: os.Getenv("ALLOWED_GROUP"), + Port: getenvDefault("PORT", "8080"), + AgentPort: getenvDefault("AGENT_PORT", "8090"), + LogLevel: getenvDefault("LOG_LEVEL", "info"), + } + for _, r := range []struct{ name, value string }{ + {"DATABASE_URL", cfg.DatabaseURL}, + {"PUBLIC_URL", cfg.PublicURL}, + {"AGENT_URL", cfg.AgentURL}, + {"OIDC_ISSUER", cfg.OIDCIssuer}, + {"OIDC_CLIENT_ID", cfg.OIDCClientID}, + {"OIDC_CLIENT_SECRET", cfg.OIDCClientSecret}, + {"ALLOWED_GROUP", cfg.AllowedGroup}, + } { + if r.value == "" { + return nil, fmt.Errorf("%s is required", r.name) + } + } + if !strings.HasPrefix(cfg.PublicURL, "http://") && !strings.HasPrefix(cfg.PublicURL, "https://") { + return nil, fmt.Errorf("PUBLIC_URL must start with http:// or https://") + } + if !strings.HasPrefix(cfg.AgentURL, "wss://") { + return nil, fmt.Errorf("AGENT_URL must start with wss://") + } + return cfg, nil +} + +func getenvDefault(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/internal/master/config/config_test.go b/internal/master/config/config_test.go new file mode 100644 index 0000000..3ef5614 --- /dev/null +++ b/internal/master/config/config_test.go @@ -0,0 +1,64 @@ +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var required = map[string]string{ + "DATABASE_URL": "postgres://u:p@db/pu", + "PUBLIC_URL": "https://updater.example.com", + "AGENT_URL": "wss://10.0.0.5:8090/agent", + "OIDC_ISSUER": "https://auth.example.com/application/o/pu/", + "OIDC_CLIENT_ID": "id", + "OIDC_CLIENT_SECRET": "secret", + "ALLOWED_GROUP": "updater-admins", +} + +func setAll(t *testing.T) { + for k, v := range required { + t.Setenv(k, v) + } + for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL"} { + t.Setenv(k, "") + } +} + +func TestLoad_Defaults(t *testing.T) { + setAll(t) + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, "8080", cfg.Port) + assert.Equal(t, "8090", cfg.AgentPort) + assert.Equal(t, "gitea.shcizo.se/shcizo/package-updater:latest", cfg.AgentImage) + assert.Equal(t, "info", cfg.LogLevel) + assert.Equal(t, "updater-admins", cfg.AllowedGroup) +} + +func TestLoad_EachRequiredVariable(t *testing.T) { + for key := range required { + t.Run(key, func(t *testing.T) { + setAll(t) + t.Setenv(key, "") + _, err := Load() + assert.ErrorContains(t, err, key+" is required") + }) + } +} + +func TestLoad_URLFormats(t *testing.T) { + cases := []struct{ key, value, wantErr string }{ + {"PUBLIC_URL", "updater.example.com", "PUBLIC_URL must start with http:// or https://"}, + {"AGENT_URL", "ws://10.0.0.5:8090/agent", "AGENT_URL must start with wss://"}, + } + for _, c := range cases { + t.Run(c.key, func(t *testing.T) { + setAll(t) + t.Setenv(c.key, c.value) + _, err := Load() + assert.ErrorContains(t, err, c.wantErr) + }) + } +}