diff --git a/README.md b/README.md index ada3c76e..b5d763db 100644 --- a/README.md +++ b/README.md @@ -467,7 +467,7 @@ CGO_ENABLED=0 go build -ldflags="-s -w -X main.version=2026.3.0" -o gateway ./cm | `GATEWAY_ID` | Stable gateway identifier (empty = generate ULID at startup) | | `GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE` | Template for the public terminal WebSocket URL, e.g. `wss://{id}.gateway.example.com/terminal` | | `GATEWAY_BOOTSTRAP_HOST` | Wildcard root hostname for agent bootstrap redirect, e.g. `gateway.example.com` | -| `GATEWAY_WEB_LISTEN_ADDR` | Listen address for the web TLS listener (terminal WebSocket), e.g. `:8443` | +| `GATEWAY_WEB_LISTEN_ADDR` | Listen address for the terminal-WebSocket HTTP listener (cleartext; Traefik terminates TLS upstream), e.g. `:8443` | | `GATEWAY_LOG_LEVEL` | Log level: `debug`, `info`, `warn`, `error` (default `info`) | | `GATEWAY_HEARTBEAT_INTERVAL` | Heartbeat cadence sent to agents (Go duration, 5s..5m; default `30s`) | | `GATEWAY_TRAEFIK_TTY_CERT_RESOLVER` | Traefik cert resolver name for the per-replica TTY HTTP router (e.g. `letsencrypt`) | diff --git a/cmd/control/README.md b/cmd/control/README.md index 3c6b8a27..5254cd93 100644 --- a/cmd/control/README.md +++ b/cmd/control/README.md @@ -103,7 +103,7 @@ export CONTROL_DATABASE_URL="postgres://powermanage:powermanage@localhost:5432/p export CONTROL_JWT_SECRET="your-secret-key" export CONTROL_CA_CERT="./dev/certs/ca.crt" export CONTROL_CA_KEY="./dev/certs/ca.key" -export CONTROL_GATEWAY_URL="https://gateway.example.com:8080" +export CONTROL_GATEWAY_URL="https://gateway.example.com" export CONTROL_ADMIN_EMAIL="admin@localhost.com" export CONTROL_ADMIN_PASSWORD="admin" diff --git a/cmd/control/main.go b/cmd/control/main.go index d1cf8610..2f4a1266 100644 --- a/cmd/control/main.go +++ b/cmd/control/main.go @@ -91,9 +91,23 @@ func main() { logger := logging.SetupLogger(cfg.LogLevel, cfg.LogFormat, os.Stderr) slog.SetDefault(logger) - logger.Info("starting control server", "version", version, "listen_addr", cfg.ListenAddr, "gateway_url", cfg.GatewayURL, "dynamic_group_eval_interval", cfg.DynamicGroupEvalInterval) - if cfg.GatewayURL == "" { - logger.Warn("CONTROL_GATEWAY_URL is not set - agents will not receive a gateway URL during registration") + // Redact the gateway URL on the startup line too. If a bad shape + // slipped in (e.g. https://u:p@host/ despite the validator, or an + // operator paste-mistake), it shouldn't land in every boot log. + logger.Info("starting control server", "version", version, "listen_addr", cfg.ListenAddr, "gateway_url", api.RedactGatewayURL(cfg.GatewayURL), "dynamic_group_eval_interval", cfg.DynamicGroupEvalInterval) + // CONTROL_GATEWAY_URL is fatal when invalid: registration hands + // it back to the agent verbatim, so any invalid shape — empty + // string, bare hostname (parses as a relative path), http:// + // (agents refuse h2c), userinfo, or non-https scheme — turns + // every successful enrollment into an agent that can never + // connect. api.ValidateGatewayURL is the shared validator + // (also invoked defensively in the registration handler). + if err := api.ValidateGatewayURL(cfg.GatewayURL); err != nil { + // Redact userinfo before logging — the validator rejects + // URLs that contain credentials, but those credentials + // shouldn't land in the startup error line regardless. + logger.Error("CONTROL_GATEWAY_URL is invalid", "gateway_url", api.RedactGatewayURL(cfg.GatewayURL), "error", err) + os.Exit(1) } // Setup signal handling diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index 7aa9f6d8..9a92cb65 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -55,6 +55,15 @@ func main() { logger := logging.SetupLogger(cfg.LogLevel, "json", os.Stdout) slog.SetDefault(logger) + // Config-shape checks that don't fit the simple "required env + // var empty" pattern below (TTY/MTLS host collision, etc.). + // Failing here keeps them visible at startup rather than at the + // first affected request. + if err := cfg.Validate(); err != nil { + logger.Error("invalid gateway configuration", "error", err) + os.Exit(1) + } + // Validate required config if cfg.ValkeyAddr == "" { logger.Error("GATEWAY_VALKEY_ADDR is required") @@ -137,27 +146,71 @@ func main() { shutdownCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - // Wire the multi-gateway registry. Reuses the same Valkey - // instance the Asynq queue uses, no extra connection pool. The - // registry is enabled only when the operator has set the - // public terminal URL template — without it the gateway can't - // know its own public URL, so we just leave the registry off - // (single-gateway / no-terminal mode). + // Wire the multi-gateway registry lazily. The registry is shared by + // three independent features: + // - terminal URL lookup for browser sessions, + // - internal gateway fan-out for control -> gateway RPCs, + // - Traefik Redis-KV self-registration. + // + // Keep those code paths independent. A malformed optional terminal + // URL must not prevent agent mTLS routing or internal gateway + // discovery from coming up. var ( gatewayReg *registry.Registry + registryRDB *redis.Client assignedHost string ) + ensureGatewayRegistry := func() *registry.Registry { + if gatewayReg != nil { + return gatewayReg + } + registryRDB = redis.NewClient(&redis.Options{ + Addr: cfg.ValkeyAddr, + Password: cfg.ValkeyPassword, + DB: cfg.ValkeyDB, + Protocol: 2, + }) + gatewayReg = registry.New(registry.NewValkeyBackend(registryRDB), logger.With("component", "registry")) + return gatewayReg + } + defer func() { + if registryRDB != nil { + if err := registryRDB.Close(); err != nil { + logger.Warn("failed to close gateway registry Valkey client", "error", err) + } + } + }() // Compute the agent redirect hostname independently of the terminal // URL. This supports multi-gateway agent routing without requiring // the terminal feature to be enabled. - if cfg.PublicAgentURLTemplate != "" { + // + // A malformed template (e.g. one that references an unset env var + // like ${TTY_DOMAIN} and expands to "https:///…" with an empty host) + // is treated as "bootstrap redirects disabled" with a loud warning + // rather than a fatal exit. The gateway still serves agent mTLS — + // which is what matters for an enrolled device. Crashing on a + // misconfigured optional feature used to kill the gateway on every + // restart, and because the Traefik Redis KV entry expires 45 s after + // each exit, the pm-mtls TCP router fell through to the HTTP router + // on the shared :443 and served the Let's Encrypt cert — giving + // agents the misleading x509 "unknown authority" error. + // Track whether the agent URL template was *configured* separately + // from whether it *resolved*. Without this split, a malformed + // template (e.g. "https://${UNSET_VAR}" → "https://") leaves + // assignedHost empty and the terminal-template block below would + // silently paper over the misconfiguration by substituting the TTY + // host — re-enabling bootstrap redirects to the wrong hostname. + // Operators who explicitly set GATEWAY_PUBLIC_AGENT_URL_TEMPLATE + // want the feature off when the template is broken, not fallen- + // back to a different URL. + agentURLTemplateConfigured := cfg.PublicAgentURLTemplate != "" + if agentURLTemplateConfigured { agentURL := strings.ReplaceAll(cfg.PublicAgentURLTemplate, "{id}", gatewayID) assignedHost = hostFromURL(agentURL) if assignedHost == "" { - logger.Error("could not extract host from GATEWAY_PUBLIC_AGENT_URL_TEMPLATE", + logger.Warn("GATEWAY_PUBLIC_AGENT_URL_TEMPLATE resolved to a URL with no host — bootstrap redirects disabled; check for unset env vars in the template", "template", cfg.PublicAgentURLTemplate, "resolved", agentURL) - os.Exit(1) } } @@ -167,97 +220,59 @@ func main() { // gateway never constructs hostnames from the request side. terminalURL := strings.ReplaceAll(cfg.PublicTerminalURLTemplate, "{id}", gatewayID) - // If no agent URL template was set, fall back to deriving the - // agent redirect hostname from the terminal URL (legacy - // single-hostname mode). - if assignedHost == "" { - assignedHost = hostFromURL(terminalURL) - if assignedHost == "" { - logger.Error("could not extract host from GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE", - "template", cfg.PublicTerminalURLTemplate, "resolved", terminalURL) - os.Exit(1) + terminalHost := hostFromURL(terminalURL) + if terminalHost == "" { + // Malformed template — skip the registry work instead of + // os.Exit(1). See the long comment above the agent-template + // check for why this has to be non-fatal. + logger.Warn("GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE resolved to a URL with no host — terminal session registration disabled on this gateway; check for unset env vars in the template", + "template", cfg.PublicTerminalURLTemplate, "resolved", terminalURL) + } else { + // Legacy single-hostname fallback: only substitute the + // terminal host when NO agent template was configured. + // If the operator set the agent template and it + // resolved to a broken URL, respect their intent to + // disable bootstrap redirects rather than silently + // masking the misconfiguration with the TTY host. + if assignedHost == "" && !agentURLTemplateConfigured { + assignedHost = terminalHost } - } - rdb := redis.NewClient(&redis.Options{ - Addr: cfg.ValkeyAddr, - Password: cfg.ValkeyPassword, - DB: cfg.ValkeyDB, - Protocol: 2, - }) - defer rdb.Close() - - gatewayReg = registry.New(registry.NewValkeyBackend(rdb), logger.With("component", "registry")) - stop, err := gatewayReg.RegisterGateway( - context.Background(), - gatewayID, - terminalURL, - registry.DefaultGatewayTTL, - registry.DefaultGatewayRefreshInterval, - ) - if err != nil { - logger.Error("failed to register gateway in registry", "error", err) - os.Exit(1) - } - defer stop() - - // Also publish the internal mTLS URL so the control server - // can discover this gateway for admin fan-out (List/Terminate - // terminal sessions). Uses the same TTL as the terminal URL. - // - // rc6 note: auto-derive when GATEWAY_INTERNAL_URL is empty. - // The GatewayService RPC is mounted on the same mTLS listener - // that accepts agents, so the internal URL is just - // `https://`. Auto-derivation keeps the - // common case zero-config — without it, operators who don't - // set the env silently lose the admin list/terminate feature - // (Terminal Sessions page shows empty). - internalURL := cfg.InternalURL - if internalURL == "" { - ip, err := routableIP() + gatewayReg = ensureGatewayRegistry() + // Match the 5s bound used on the other Redis-touching + // calls in this file (PublishTraefikRoute, + // RegisterGatewayInternal). Without it, a slow or hung + // Valkey at startup stalls gateway boot past the point + // where SIGTERM should be respected. The refresh + // goroutine stop() returned from RegisterGateway carries + // its own shutdown wiring, so the bound only affects the + // one-shot register at line 242. + registerCtx, cancelRegister := context.WithTimeout(shutdownCtx, 5*time.Second) + stop, err := gatewayReg.RegisterGateway( + registerCtx, + gatewayID, + terminalURL, + registry.DefaultGatewayTTL, + registry.DefaultGatewayRefreshInterval, + ) + cancelRegister() if err != nil { - logger.Error("cannot auto-derive GATEWAY_INTERNAL_URL", "error", err) - os.Exit(1) + // Fail-open: the terminal feature is optional, so a + // transient registry failure at startup must not kill + // the gateway's agent-mTLS service. Log loudly and + // leave terminal sessions disabled for this replica + // until the operator restarts. + logger.Warn("failed to register gateway in terminal registry — terminal sessions disabled on this replica", + "error", err) + } else { + defer stop() + logger.Info("multi-gateway routing enabled", + "gateway_id", gatewayID, + "terminal_url", terminalURL, + "agent_redirect_host", assignedHost, + ) } - internalURL = "https://" + ip + portOfListenAddr(cfg.ListenAddr) - logger.Info("auto-derived GATEWAY_INTERNAL_URL", "internal_url", internalURL) } - if internalURL != "" { - if err := gatewayReg.RegisterGatewayInternal( - context.Background(), gatewayID, internalURL, registry.DefaultGatewayTTL, - ); err != nil { - logger.Warn("failed to register gateway internal URL", "error", err) - } - // Refresh the internal URL on the same cadence as the - // terminal URL so it does not expire while the gateway is - // running. Derive from shutdownCtx so the goroutine exits - // as soon as a signal arrives. - internalRefreshCtx, stopInternalRefresh := context.WithCancel(shutdownCtx) - defer stopInternalRefresh() - go func() { - ticker := time.NewTicker(registry.DefaultGatewayRefreshInterval) - defer ticker.Stop() - for { - select { - case <-ticker.C: - if err := gatewayReg.RegisterGatewayInternal( - context.Background(), gatewayID, internalURL, registry.DefaultGatewayTTL, - ); err != nil { - logger.Warn("failed to refresh gateway internal URL", "error", err) - } - case <-internalRefreshCtx.Done(): - return - } - } - }() - } - - logger.Info("multi-gateway routing enabled", - "gateway_id", gatewayID, - "terminal_url", terminalURL, - "agent_redirect_host", assignedHost, - "internal_url", internalURL, - ) } // Traefik Redis-KV self-registration. Opt-in; when enabled, every @@ -268,20 +283,11 @@ func main() { // load-balanced across all replicas; each replica owns a unique // /gw/ path prefix on the shared tty host for TTY routing. if cfg.TraefikSelfRegister { - if gatewayReg == nil { - // Need a Valkey-backed registry. When PublicTerminalURLTemplate - // was empty we skipped the Valkey connection earlier; set one - // up here so Traefik self-reg still works in deployments that - // only want the routing layer (no terminal feature). - rdb := redis.NewClient(&redis.Options{ - Addr: cfg.ValkeyAddr, - Password: cfg.ValkeyPassword, - DB: cfg.ValkeyDB, - Protocol: 2, - }) - defer rdb.Close() - gatewayReg = registry.New(registry.NewValkeyBackend(rdb), logger.With("component", "registry")) - } + // Need a Valkey-backed registry even when terminal URLs are + // disabled; Traefik self-registration is the agent mTLS + // routing layer. ensureGatewayRegistry is idempotent, so an + // outer nil check would be noise. + gatewayReg = ensureGatewayRegistry() // Auto-derive per-replica backend addresses when not set. We use // the replica's own routable IP on the shared Docker/k8s network @@ -295,85 +301,192 @@ func main() { if mtlsBackend == "" || ttyBackend == "" { ip, err := routableIP() if err != nil { - logger.Error("cannot auto-derive Traefik backends", "error", err) - os.Exit(1) - } - if mtlsBackend == "" { - mtlsBackend = ip + portOfListenAddr(cfg.ListenAddr) - } - if ttyBackend == "" && cfg.WebListenAddr != "" { - ttyBackend = "http://" + ip + portOfListenAddr(cfg.WebListenAddr) + // Fail-open: the operator can still publish a manual + // backend via GATEWAY_TRAEFIK_{MTLS,TTY}_BACKEND, so + // missing auto-derivation is not fatal. If nothing is + // set we skip Traefik publication entirely below and + // the gateway keeps serving agent mTLS on whatever + // static route already points at it. + logger.Warn("cannot auto-derive Traefik backends from network interfaces — set GATEWAY_TRAEFIK_{MTLS,TTY}_BACKEND explicitly if self-registration is needed", + "error", err) + } else { + if mtlsBackend == "" { + mtlsBackend = ip + portOfListenAddr(cfg.ListenAddr) + } + if ttyBackend == "" && cfg.WebListenAddr != "" { + ttyBackend = "http://" + ip + portOfListenAddr(cfg.WebListenAddr) + } } } - traefikCfg := registry.TraefikRouteConfig{ - RootKey: cfg.TraefikRootKey, - MTLSHost: cfg.TraefikMTLSHost, - MTLSBackend: mtlsBackend, - MTLSEntryPoint: cfg.TraefikMTLSEntryPoint, - TTYHost: cfg.TraefikTTYHost, - TTYBackend: ttyBackend, - TTYEntryPoint: cfg.TraefikTTYEntryPoint, - TTYCertResolver: cfg.TraefikTTYCertResolver, - } + // Publish only when we actually have a usable mTLS backend — + // an empty backend in the Redis KV entry would poison the + // pm-mtls TCP router for every replica. + if mtlsBackend == "" { + logger.Warn("Traefik self-registration skipped — no MTLS backend available (routable-IP auto-derive failed and GATEWAY_TRAEFIK_MTLS_BACKEND is unset)") + } else { + traefikCfg := registry.TraefikRouteConfig{ + RootKey: cfg.TraefikRootKey, + MTLSHost: cfg.TraefikMTLSHost, + MTLSBackend: mtlsBackend, + MTLSEntryPoint: cfg.TraefikMTLSEntryPoint, + TTYHost: cfg.TraefikTTYHost, + TTYBackend: ttyBackend, + TTYEntryPoint: cfg.TraefikTTYEntryPoint, + TTYCertResolver: cfg.TraefikTTYCertResolver, + } - if err := gatewayReg.PublishTraefikRoute( - context.Background(), gatewayID, traefikCfg, registry.DefaultGatewayTTL, - ); err != nil { - logger.Error("failed to publish Traefik routing config", "error", err) - os.Exit(1) - } + // Bound the publish with shutdownCtx + a 5s timeout so a + // slow Redis can't stall startup, and the same bound is + // used on refresh so shutdown is prompt even during a + // hung publish. + publishCtx, cancelPublish := context.WithTimeout(shutdownCtx, 5*time.Second) + err := gatewayReg.PublishTraefikRoute( + publishCtx, gatewayID, traefikCfg, registry.DefaultGatewayTTL, + ) + cancelPublish() + if err != nil { + // Fail-open: the refresh goroutine below retries on + // the normal cadence, so a transient publish failure + // at startup recovers within one refresh interval. + // Killing the gateway used to be worse — the Redis + // KV entry expired during the restart backoff and + // Traefik fell through to the HTTP router (wrong + // cert). Log loudly and keep serving agent mTLS. + logger.Warn("failed to publish initial Traefik routing config — refresh goroutine will retry", + "error", err) + } - // Refresh on the same cadence as the gateway terminal URL and - // internal URL so all three keys share a lifecycle. Derive from - // shutdownCtx so the goroutine exits cleanly on SIGTERM. - traefikRefreshCtx, stopTraefikRefresh := context.WithCancel(shutdownCtx) - defer stopTraefikRefresh() - go func() { - ticker := time.NewTicker(registry.DefaultGatewayRefreshInterval) - defer ticker.Stop() - for { - select { - case <-ticker.C: - if err := gatewayReg.PublishTraefikRoute( - context.Background(), gatewayID, traefikCfg, registry.DefaultGatewayTTL, - ); err != nil { - logger.Warn("failed to refresh Traefik routing config", "error", err) + // Refresh on the same cadence as the gateway terminal URL + // so both keys share a lifecycle. Derive from shutdownCtx + // so the goroutine exits cleanly on SIGTERM. + traefikRefreshCtx, stopTraefikRefresh := context.WithCancel(shutdownCtx) + defer stopTraefikRefresh() + go func() { + ticker := time.NewTicker(registry.DefaultGatewayRefreshInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + refreshCtx, cancelRefresh := context.WithTimeout(traefikRefreshCtx, 5*time.Second) + err := gatewayReg.PublishTraefikRoute( + refreshCtx, gatewayID, traefikCfg, registry.DefaultGatewayTTL, + ) + cancelRefresh() + if err != nil { + logger.Warn("failed to refresh Traefik routing config", "error", err) + } + case <-traefikRefreshCtx.Done(): + return } - case <-traefikRefreshCtx.Done(): - return } - } - }() + }() - // Clean shutdown revokes only per-replica keys so other - // replicas' routes stay up. Uses a bounded context so a flaky - // Valkey can't stall the shutdown. - defer func() { - cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := gatewayReg.RevokeTraefikRoute(cleanupCtx, gatewayID, traefikCfg); err != nil { - logger.Warn("failed to revoke Traefik routing config", "error", err) - } - }() + // Clean shutdown revokes only per-replica keys so other + // replicas' routes stay up. Uses a bounded context so a + // flaky Valkey can't stall the shutdown. + defer func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := gatewayReg.RevokeTraefikRoute(cleanupCtx, gatewayID, traefikCfg); err != nil { + logger.Warn("failed to revoke Traefik routing config", "error", err) + } + }() - logger.Info("traefik self-registration enabled", - "gateway_id", gatewayID, - "mtls_host", cfg.TraefikMTLSHost, - "mtls_backend", cfg.TraefikMTLSBackend, - "tty_host", cfg.TraefikTTYHost, - "tty_backend", cfg.TraefikTTYBackend, - "tty_cert_resolver", cfg.TraefikTTYCertResolver, - ) + // Surface whether the initial publish succeeded. After a + // Valkey wobble the refresh loop will self-heal within + // one DefaultGatewayRefreshInterval, but an operator + // reading the log right after startup should be able to + // tell "registered now" from "will retry in N seconds" + // without having to cross-reference the Warn line above. + logger.Info("traefik self-registration enabled", + "gateway_id", gatewayID, + "initial_publish_ok", err == nil, + "mtls_host", cfg.TraefikMTLSHost, + "mtls_backend", mtlsBackend, + "tty_host", cfg.TraefikTTYHost, + "tty_backend", ttyBackend, + "tty_cert_resolver", cfg.TraefikTTYCertResolver, + ) + } } - // Fail fast if BootstrapHost is set but we have no assignedHost - // (because PublicTerminalURLTemplate was empty). Without this - // guard, BootstrapRedirectMiddleware would panic on an empty - // assignedHost further down. - if cfg.BootstrapHost != "" && assignedHost == "" { - logger.Error("GATEWAY_BOOTSTRAP_HOST is set but neither GATEWAY_PUBLIC_AGENT_URL_TEMPLATE nor GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE is set; cannot derive the per-gateway hostname for bootstrap redirects", - "bootstrap_host", cfg.BootstrapHost) - os.Exit(1) + + // Publish the internal mTLS URL so the control server can discover + // this gateway for admin fan-out (List/Terminate terminal sessions). + // This is intentionally independent of terminal URL registration: + // a bad optional public terminal URL (which may have left + // gatewayReg unset) should not disable the internal control-plane + // route. ensureGatewayRegistry is idempotent — if the Traefik + // block above already created the registry, this is a no-op; if + // terminal + Traefik are both off, this is where the registry + // gets built so admin fan-out still works. + gatewayReg = ensureGatewayRegistry() + { + internalURL := cfg.InternalURL + if internalURL == "" { + ip, err := routableIP() + if err != nil { + // Fail-open: the internal URL feeds the admin + // fan-out path (List/Terminate terminal sessions). + // It is strictly optional for agent mTLS service. + // Crashing the gateway here would take the agent + // routing layer down with it, which is exactly the + // rc8 failure mode we're fixing. Log and move on. + logger.Warn("cannot auto-derive GATEWAY_INTERNAL_URL — admin fan-out disabled for this replica; set GATEWAY_INTERNAL_URL explicitly to re-enable", + "error", err) + } else { + internalURL = "https://" + ip + portOfListenAddr(cfg.ListenAddr) + logger.Info("auto-derived GATEWAY_INTERNAL_URL", "internal_url", internalURL) + } + } + if internalURL != "" { + // Bound both the initial register and the periodic + // refresh with shutdownCtx + 5s so a slow Redis can't + // delay shutdown or pile up goroutines waiting on the + // registry during degraded Valkey health. + registerCtx, cancelRegister := context.WithTimeout(shutdownCtx, 5*time.Second) + err := gatewayReg.RegisterGatewayInternal( + registerCtx, gatewayID, internalURL, registry.DefaultGatewayTTL, + ) + cancelRegister() + if err != nil { + logger.Warn("failed to register gateway internal URL", "error", err) + } + internalRefreshCtx, stopInternalRefresh := context.WithCancel(shutdownCtx) + defer stopInternalRefresh() + go func() { + ticker := time.NewTicker(registry.DefaultGatewayRefreshInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + refreshCtx, cancelRefresh := context.WithTimeout(internalRefreshCtx, 5*time.Second) + err := gatewayReg.RegisterGatewayInternal( + refreshCtx, gatewayID, internalURL, registry.DefaultGatewayTTL, + ) + cancelRefresh() + if err != nil { + logger.Warn("failed to refresh gateway internal URL", "error", err) + } + case <-internalRefreshCtx.Done(): + return + } + } + }() + } + } + + // If BootstrapHost is set but no assignedHost is available + // (because both agent + terminal URL templates were empty or + // malformed), disable the bootstrap-redirect middleware instead + // of crashing. Agent mTLS still works — operators just lose the + // convenience redirect that points newly-enrolled agents at a + // stable per-gateway hostname. + bootstrapHost := cfg.BootstrapHost + if bootstrapHost != "" && assignedHost == "" { + logger.Warn("GATEWAY_BOOTSTRAP_HOST is set but no assigned host could be derived from GATEWAY_PUBLIC_AGENT_URL_TEMPLATE / GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE — bootstrap redirects disabled", + "bootstrap_host", bootstrapHost) + bootstrapHost = "" // disable middleware below } // Terminal session registry — shared between the agent bidi @@ -388,11 +501,11 @@ func main() { // Setup HTTP mux for agent connections (mTLS-protected) mux := http.NewServeMux() - // Create agent handler (always mTLS) + // Create agent handler (always mTLS). gatewayReg is always non-nil + // by this point — ensureGatewayRegistry() is called unconditionally + // in the internal-URL block above — so no guard on SetGatewayRouting. agentHandler := handler.NewAgentHandlerWithTLS(manager, aqClient, controlProxy, workerMgr, version, cfg.HeartbeatInterval, logger) - if gatewayReg != nil { - agentHandler.SetGatewayRouting(gatewayReg, gatewayID) - } + agentHandler.SetGatewayRouting(gatewayReg, gatewayID) agentHandler.SetTerminalSessions(terminalSessions) path, h := pmv1connect.NewAgentServiceHandler(agentHandler) @@ -402,7 +515,7 @@ func main() { // ↑ BootstrapRedirectMiddleware (returns 307 to assignedHost // when the request landed on the wildcard root via LB) mtlsHandler := handler.MTLSMiddleware(h, logger) - bootstrappedHandler := handler.BootstrapRedirectMiddleware(mtlsHandler, cfg.BootstrapHost, assignedHost, logger) + bootstrappedHandler := handler.BootstrapRedirectMiddleware(mtlsHandler, bootstrapHost, assignedHost, logger) mux.Handle(path, bootstrappedHandler) // Mount GatewayService on the mTLS listener (internal-only, @@ -635,10 +748,17 @@ func hostFromURL(raw string) string { default: return "" } - // u.Host includes the port (e.g. "gw-01.example.com:8443"). - // u.Hostname() strips it, which would break redirects on - // non-default ports. - if u.Host == "" { + // u.Host includes the port (e.g. "gw-01.example.com:8443"), which + // is what the caller needs for bootstrap redirects and registry + // registration. u.Hostname() strips the port, so we return u.Host. + // + // But both checks matter: a template like "https://${UNSET}:8443" + // collapses to "https://:8443" when the env var is missing. + // u.Host = ":8443" is non-empty (would pass the first check) + // while u.Hostname() = "" (no hostname). Require both — same + // invariant api.ValidateGatewayURL enforces for the control + // server's outbound URL. + if u.Host == "" || u.Hostname() == "" { return "" } return u.Host diff --git a/deploy/.env.example b/deploy/.env.example index c5de8f37..0853b38d 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -56,10 +56,22 @@ CONTROL_DOMAIN=power-manage.example.com # gateway TCP passthrough for the mTLS handshake on this subdomain. GATEWAY_DOMAIN=gateway.example.com -# Public domain for TTY WebSocket traffic. Defaults to GATEWAY_DOMAIN when -# unset, so single-domain deployments just leave this empty. Each gateway -# replica takes a /gw/ path prefix on this host for session-specific -# routing. +# Public domain for TTY WebSocket traffic. +# +# When the terminal WebSocket listener is enabled (the default when +# GATEWAY_WEB_LISTEN_ADDR is set and Traefik self-registration is on), +# this MUST be a different hostname than GATEWAY_DOMAIN — otherwise +# Traefik's TCP passthrough router for mTLS shadows the TTY HTTP +# router on the shared SNI and WebSocket sessions break silently. +# The gateway refuses to start in that collision state. +# +# When the terminal listener is disabled (GATEWAY_WEB_LISTEN_ADDR +# empty), this falls back to GATEWAY_DOMAIN harmlessly — the +# fallback only exists so the Traefik-registry entry still has a +# non-empty host field. +# +# Each gateway replica takes a /gw/ path prefix on this host +# for session-specific routing. # GATEWAY_TTY_DOMAIN=tty.example.com # ============================================================================= @@ -139,7 +151,13 @@ LOG_LEVEL=info # CONTROL_SSH_ACCESS_FOR_ALL=false # --- Docker Images ---------------------------------------------------------- -# Tag for control, gateway, and indexer images (default: latest) +# Tag for control, gateway, and indexer images. +# latest — latest stable release (default; CI promotes to this +# tag only after a stable release is cut) +# latest-rc — latest pre-release / rc, use this to track the +# current release-candidate line +# v2026.05 — pin to a specific release for stricter rollback +# semantics # IMAGE_TAG=latest # --- Gateway scaling (Traefik Redis KV self-registration) ------------------- diff --git a/deploy/compose.yml b/deploy/compose.yml index 65d7ec1c..7b53951e 100644 --- a/deploy/compose.yml +++ b/deploy/compose.yml @@ -81,7 +81,10 @@ services: - internal valkey: - image: docker.io/redis/redis-stack-server:latest + # Pinned to avoid silent upstream upgrades. Bump deliberately when + # you've verified the RediSearch module + data format still work + # for your deployment. + image: docker.io/redis/redis-stack-server:7.4.0-v0 container_name: pm-valkey restart: unless-stopped volumes: @@ -191,6 +194,7 @@ services: environment: - GATEWAY_VALKEY_ADDR=valkey:6379 - GATEWAY_VALKEY_PASSWORD=${VALKEY_PASSWORD} + - GATEWAY_VALKEY_DB=${GATEWAY_VALKEY_DB:-0} - GATEWAY_CONTROL_URL=https://control:8082 - GATEWAY_LISTEN_ADDR=:8080 - GATEWAY_LOG_LEVEL=${LOG_LEVEL:-info} diff --git a/deploy/deploy.sh b/deploy/deploy.sh index 649847b3..e94efc08 100755 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -202,7 +202,10 @@ log_info "Transfer complete" log_step "Loading images and restarting services on $SSH_HOST..." # Build the remote commands -# Read IMAGE_TAG from server's .env so we retag images to match what compose expects +# Read IMAGE_TAG from the server's .env so we retag images to match +# what compose expects. Falls back to "latest" when unset — matches +# compose.yml's ${IMAGE_TAG:-latest} default, which points at the +# curated latest-stable tag CI promotes after a release is cut. REMOTE_CMDS="cd ~/deploy && " REMOTE_CMDS+="IMAGE_TAG=\$(grep -oP '(?<=^IMAGE_TAG=).*' .env 2>/dev/null || echo latest) && " REMOTE_CMDS+="echo '[INFO] Server IMAGE_TAG='\$IMAGE_TAG && " diff --git a/deploy/initdb.d/01-indexer-user.sh b/deploy/initdb.d/01-indexer-user.sh index 3d92936e..e71c3f13 100755 --- a/deploy/initdb.d/01-indexer-user.sh +++ b/deploy/initdb.d/01-indexer-user.sh @@ -6,9 +6,14 @@ # ALTER ROLE pm_indexer PASSWORD 'your_password'; set -e +# Hard-fail on unset password: returning 0 caused the indexer container +# to crash-loop later with an obscure "no password supplied" error at +# first connect, hours after `setup.sh` had already reported success. +# Failing the init script keeps the failure at setup time where the +# operator is still paying attention. if [ -z "$INDEXER_POSTGRES_PASSWORD" ]; then - echo "WARN: INDEXER_POSTGRES_PASSWORD not set, skipping pm_indexer user creation" - exit 0 + echo "ERROR: INDEXER_POSTGRES_PASSWORD is required — set it in .env before initialising postgres (must be URL-safe; see .env.example)" >&2 + exit 1 fi psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL diff --git a/internal/api/auth_handler.go b/internal/api/auth_handler.go index 509c831f..8673d803 100644 --- a/internal/api/auth_handler.go +++ b/internal/api/auth_handler.go @@ -19,17 +19,24 @@ import ( // AuthHandler handles authentication RPCs. type AuthHandler struct { - store *store.Store - logger *slog.Logger - jwtManager *auth.JWTManager + store *store.Store + logger *slog.Logger + jwtManager *auth.JWTManager + passwordAuthEnabled bool } -// NewAuthHandler creates a new auth handler. -func NewAuthHandler(st *store.Store, logger *slog.Logger, jwtManager *auth.JWTManager) *AuthHandler { +// NewAuthHandler creates a new auth handler. The passwordAuthEnabled flag +// is the global `CONTROL_PASSWORD_AUTH_ENABLED` operator switch. When +// false, Login rejects every password attempt regardless of the per-user +// HasPassword column — previous revs only gated the SSO UI's auth-method +// list on this flag, leaving the RPC itself open to direct password +// attempts against accounts that still had a password hash on disk. +func NewAuthHandler(st *store.Store, logger *slog.Logger, jwtManager *auth.JWTManager, passwordAuthEnabled bool) *AuthHandler { return &AuthHandler{ - store: st, - logger: logger, - jwtManager: jwtManager, + store: st, + logger: logger, + jwtManager: jwtManager, + passwordAuthEnabled: passwordAuthEnabled, } } @@ -39,6 +46,14 @@ func (h *AuthHandler) Login(ctx context.Context, req *connect.Request[pm.LoginRe return nil, err } + // Global password-auth switch — enforced BEFORE the user lookup so a + // burned bcrypt cycle doesn't leak account existence via timing when + // the operator has disabled password login entirely. + if !h.passwordAuthEnabled { + auth.VerifyPassword(req.Msg.Password, auth.DummyHash) + return nil, apiErrorCtx(ctx, ErrPasswordLoginDisabled, connect.CodeUnauthenticated, "password login is disabled on this server") + } + user, err := h.store.Queries().GetUserByEmail(ctx, req.Msg.Email) if err != nil { if errors.Is(err, pgx.ErrNoRows) { diff --git a/internal/api/auth_handler_test.go b/internal/api/auth_handler_test.go index 06607cfc..45108788 100644 --- a/internal/api/auth_handler_test.go +++ b/internal/api/auth_handler_test.go @@ -19,7 +19,7 @@ import ( func TestLogin_Success(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "correct-password", "admin") @@ -40,7 +40,7 @@ func TestLogin_Success(t *testing.T) { func TestLogin_WrongPassword(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "correct-password", "user") @@ -56,7 +56,7 @@ func TestLogin_WrongPassword(t *testing.T) { func TestLogin_NonexistentUser(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) _, err := h.Login(context.Background(), connect.NewRequest(&pm.LoginRequest{ Email: "nonexistent@test.com", @@ -66,10 +66,41 @@ func TestLogin_NonexistentUser(t *testing.T) { assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err)) } +// TestLogin_GlobalPasswordAuthDisabled regression-tests the bypass +// discovered in the rc10 audit: the `CONTROL_PASSWORD_AUTH_ENABLED=false` +// operator switch was only plumbed into the SSO handler's +// ListAuthMethods response, so a direct POST to /Login with valid +// credentials still authenticated. Now the switch is enforced in +// AuthHandler.Login itself before the DB lookup, so even an account +// with a password hash on disk cannot authenticate when the operator +// disables password login globally. +func TestLogin_GlobalPasswordAuthDisabled(t *testing.T) { + st := testutil.SetupPostgres(t) + jwtMgr := testutil.NewJWTManager() + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, false) // global switch off + + email := testutil.NewID() + "@test.com" + testutil.CreateTestUser(t, st, email, "correct-password", "admin") + + _, err := h.Login(context.Background(), connect.NewRequest(&pm.LoginRequest{ + Email: email, + Password: "correct-password", // valid credentials + })) + require.Error(t, err, "login must fail when global password auth is disabled") + assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err)) + // Also assert the user-facing error-surface carries the + // "password login is disabled" sentinel (not the generic + // "invalid credentials"). A future refactor that collapses this + // branch into the invalid-credentials path would look correct + // under the code check but fail this substring — which is + // exactly the signal we want. + assert.Contains(t, err.Error(), "password login is disabled") +} + func TestLogin_DisabledUser(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" userID := testutil.CreateTestUser(t, st, email, "password", "user") @@ -89,7 +120,7 @@ func TestLogin_DisabledUser(t *testing.T) { func TestLogin_NoCookiesSet(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") @@ -115,7 +146,7 @@ func TestLogin_NoCookiesSet(t *testing.T) { func TestRefreshToken_RequiresBodyToken(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) // RefreshToken with empty body should fail. // Proto validation catches the empty refresh_token field. @@ -127,7 +158,7 @@ func TestRefreshToken_RequiresBodyToken(t *testing.T) { func TestRefreshToken_NoCookieFallback(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") @@ -153,7 +184,7 @@ func TestRefreshToken_NoCookieFallback(t *testing.T) { func TestRefreshToken_BodyToken(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") @@ -183,7 +214,7 @@ func TestRefreshToken_BodyToken(t *testing.T) { func TestLogout_NoCookiesCleared(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") @@ -209,7 +240,7 @@ func TestLogout_NoCookiesCleared(t *testing.T) { func TestGetCurrentUser(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" userID := testutil.CreateTestUser(t, st, email, "pass", "admin") @@ -226,7 +257,7 @@ func TestLogin_TOTPRequired(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() enc := testutil.NewEncryptor(t) - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" userID := testutil.CreateTestUser(t, st, email, "password", "user") @@ -251,7 +282,7 @@ func TestLogin_TOTPRequired(t *testing.T) { func TestLogin_TOTPNotRequired(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") @@ -273,7 +304,7 @@ func TestLogin_TOTPNotRequired(t *testing.T) { func TestLogin_PasswordDisabledByProvider(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" userID := testutil.CreateTestUser(t, st, email, "password", "user") @@ -315,7 +346,7 @@ func TestLogin_PasswordDisabledByProvider(t *testing.T) { func TestLogin_SSOOnlyUserNoPassword(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" userID := testutil.NewID() @@ -335,7 +366,7 @@ func TestLogin_SSOOnlyUserNoPassword(t *testing.T) { func TestLogin_UserHasPassword(t *testing.T) { st := testutil.SetupPostgres(t) jwtMgr := testutil.NewJWTManager() - h := api.NewAuthHandler(st, slog.Default(), jwtMgr) + h := api.NewAuthHandler(st, slog.Default(), jwtMgr, true) email := testutil.NewID() + "@test.com" testutil.CreateTestUser(t, st, email, "password", "user") diff --git a/internal/api/errors_parity_test.go b/internal/api/errors_parity_test.go new file mode 100644 index 00000000..efcdf1db --- /dev/null +++ b/internal/api/errors_parity_test.go @@ -0,0 +1,154 @@ +package api + +import ( + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "testing" +) + +// TestErrorCodeParityWithTSSDK guards against drift between the +// snake_case error codes the server emits (errors.go) and the Err* +// constants the TS SDK exports (sdk/ts/errors.ts). When the two +// lists disagree, the web client falls back to raw code strings +// instead of the paraglide-localized message, which is exactly the +// kind of silent UX regression the rc10 audit flagged. +// +// Fails with a diff when the sets don't match. To fix: +// - If the server added a new code, add a matching Err* const to +// sdk/ts/errors.ts AND a matching error_ paraglide key to +// web/messages/{en,de}.json. +// - If the server removed a code, delete the const from +// sdk/ts/errors.ts and the matching paraglide key. +func TestErrorCodeParityWithTSSDK(t *testing.T) { + serverCodes := extractServerCodes(t) + sdkCodes := extractSDKCodes(t) + + missingFromSDK := diff(serverCodes, sdkCodes) + extraInSDK := diff(sdkCodes, serverCodes) + + if len(missingFromSDK) > 0 { + t.Errorf("error codes emitted by server but not exported by sdk/ts/errors.ts: %v\n"+ + "→ add a matching Err* const to sdk/ts/errors.ts and an error_ key to web/messages/{en,de}.json", + missingFromSDK) + } + if len(extraInSDK) > 0 { + t.Errorf("error codes exported by sdk/ts/errors.ts but never emitted by the server: %v\n"+ + "→ delete the stale consts (and any matching paraglide keys) to stop lying to future developers", + extraInSDK) + } +} + +// extractServerCodes walks errors.go and returns every snake_case +// string literal assigned to an Err* constant. Deliberately NOT +// importing the constants directly — we want to catch the case where +// a const is declared but never used, or where a code is hard-coded +// in a handler without going through the constant. +func extractServerCodes(t *testing.T) []string { + t.Helper() + data, err := os.ReadFile("errors.go") + if err != nil { + t.Fatalf("read server errors.go: %v", err) + } + // Match lines like: ErrWhatever = "snake_case_value" + re := regexp.MustCompile(`Err\w+\s*=\s*"([a-z][a-z0-9_]*)"`) + matches := re.FindAllStringSubmatch(string(data), -1) + seen := make(map[string]struct{}, len(matches)) + for _, m := range matches { + seen[m[1]] = struct{}{} + } + out := make([]string, 0, len(seen)) + for code := range seen { + out = append(out, code) + } + sort.Strings(out) + return out +} + +// extractSDKCodes walks sdk/ts/errors.ts and returns every +// snake_case string literal assigned to an exported Err* const. +// +// Resolution order: +// 1. PM_SDK_TS_ERRORS env var (absolute or relative path) — CI sets +// this when it checks the SDK out beside the server repo, so +// standalone server CI can still exercise the parity guard. +// 2. ../../../../sdk/ts/errors.ts — the local dev-workspace layout +// /home//.../power-manage/{server,sdk}. +// +// If neither resolves AND PM_SDK_PARITY_REQUIRED=1 is set, the test +// fails loudly — this is the mode CI should use when it expects the +// SDK to be available. Without the env var the test skips with a +// clear log line so a local `go test ./...` in a standalone server +// checkout still passes. +func extractSDKCodes(t *testing.T) []string { + t.Helper() + + var candidates []string + if env := os.Getenv("PM_SDK_TS_ERRORS"); env != "" { + candidates = append(candidates, env) + } + // go test sets cwd to the package dir (server/internal/api), so + // ../../../ resolves to the multi-repo workspace root where + // /sdk lives alongside /server. + candidates = append(candidates, filepath.Join("..", "..", "..", "sdk", "ts", "errors.ts")) + // Fallback: some harnesses run tests from one level higher (e.g. + // when the repo is checked out directly without the workspace + // wrapper). Keeping the older depth as a second candidate lets + // both shapes succeed without env-var wrangling. + candidates = append(candidates, filepath.Join("..", "..", "..", "..", "sdk", "ts", "errors.ts")) + + var data []byte + var tried []string + for _, path := range candidates { + tried = append(tried, path) + b, err := os.ReadFile(path) + if err == nil { + data = b + break + } + } + if data == nil { + msg := "cannot read sdk/ts/errors.ts from any candidate path: " + strings.Join(tried, ", ") + if os.Getenv("PM_SDK_PARITY_REQUIRED") == "1" { + t.Fatalf("PM_SDK_PARITY_REQUIRED=1 but %s — CI should check out the sdk repo beside server or set PM_SDK_TS_ERRORS", msg) + } + t.Skipf("%s — set PM_SDK_TS_ERRORS or PM_SDK_PARITY_REQUIRED=1 (with the file available) to exercise the parity guard", msg) + return nil + } + + // Accept single, double, or backtick-quoted string literals plus + // an optional `: string` type annotation so a future refactor to + // `export const ErrFoo: string = "…"` or a template literal + // doesn't silently hide the code from the parity check. + re := regexp.MustCompile(`export\s+const\s+Err\w+(?:\s*:\s*string)?\s*=\s*['"`+"`"+`]([a-z][a-z0-9_]*)['"`+"`"+`]`) + matches := re.FindAllStringSubmatch(string(data), -1) + seen := make(map[string]struct{}, len(matches)) + for _, m := range matches { + seen[m[1]] = struct{}{} + } + out := make([]string, 0, len(seen)) + for code := range seen { + out = append(out, code) + } + sort.Strings(out) + return out +} + +// diff returns elements in a that are not in b, sorted. +func diff(a, b []string) []string { + present := make(map[string]struct{}, len(b)) + for _, s := range b { + present[s] = struct{}{} + } + var out []string + for _, s := range a { + if _, ok := present[s]; !ok { + out = append(out, s) + } + } + sort.Strings(out) + return out +} + diff --git a/internal/api/internal_handler.go b/internal/api/internal_handler.go index 24c68526..6b62a267 100644 --- a/internal/api/internal_handler.go +++ b/internal/api/internal_handler.go @@ -304,18 +304,29 @@ func (h *InternalHandler) ProxyStoreLpsPasswords(ctx context.Context, req *conne // returns the session metadata the gateway needs to bridge the // connection. // -// Validation does NOT consume the entry — the same gateway uses the -// metadata for the lifetime of the WebSocket. Revocation happens -// explicitly via ControlService.StopTerminal or -// TerminateTerminalSession. This contract is documented on the RPC -// in manchtools/power-manage-sdk#27. +// rc10 single-use contract: a successful validation CONSUMES the +// token atomically (Valkey GETDEL), so a second call with the same +// bearer returns Unauthenticated. This blocks the replay surface +// where a token leaks via a reverse-proxy access log that captured +// the query-string — the attacker can no longer mint additional +// WebSocket connections during the 60 s TTL. // -// Distinguishes 'unknown / expired' (Unauthenticated, with a generic -// message so a forgery probe cannot tell the difference between an -// expired token and a never-existed one) from 'mismatched token' -// (Unauthenticated, but logged separately so the audit pipeline can -// flag forgery attempts). 'Token store not configured' is Unavailable -// — that's an operator misconfiguration, not a client bug. +// Real flow only validates once per WS: the gateway calls this RPC +// from terminal_bridge.go at connection acceptance, stashes the +// returned metadata for the WebSocket's lifetime, and never re- +// validates. So the single-use contract is consistent with normal +// operation; only attacker replays break. +// +// Forgery attempts (valid session_id, wrong bearer) do NOT consume +// the entry — the terminal store restores the session with its +// remaining TTL so a legitimate client isn't locked out by a guess. +// +// Distinguishes 'unknown / expired / already consumed' (Unauthenticated, +// with a generic message so a forgery probe cannot tell the +// difference) from 'mismatched token' (Unauthenticated, but logged +// separately so the audit pipeline can flag forgery attempts). 'Token +// store not configured' is Unavailable — operator misconfiguration, +// not a client bug. func (h *InternalHandler) ProxyValidateTerminalToken(ctx context.Context, req *connect.Request[pm.InternalValidateTerminalTokenRequest]) (*connect.Response[pm.InternalValidateTerminalTokenResponse], error) { if h.terminalTokenStore == nil { return nil, connect.NewError(connect.CodeUnavailable, diff --git a/internal/api/internal_handler_test.go b/internal/api/internal_handler_test.go index b5b286da..d636b58f 100644 --- a/internal/api/internal_handler_test.go +++ b/internal/api/internal_handler_test.go @@ -230,11 +230,19 @@ func TestProxyValidateTerminalToken_HappyPath(t *testing.T) { assert.Equal(t, uint32(40), resp.Msg.Rows) } -func TestProxyValidateTerminalToken_DoesNotConsumeToken(t *testing.T) { - // The contract documented above the RPC says validation must NOT - // consume the entry — the same gateway uses the metadata for the - // lifetime of the WebSocket. Validate twice and assert both - // succeed. +func TestProxyValidateTerminalToken_IsSingleUse(t *testing.T) { + // rc10 contract: a successful validation atomically consumes the + // token so replays within the TTL fail with Unauthenticated. This + // blocks the leaked-token replay surface (reverse-proxy access + // logs capturing query strings, browser history snooping, etc.) + // without affecting normal operation — the real gateway flow in + // terminal_bridge.go validates the token exactly once per + // WebSocket connection and uses the returned metadata for the + // lifetime of the connection. + // + // Supersedes the pre-rc10 TestProxyValidateTerminalToken_DoesNotConsumeToken + // which asserted the opposite and is exactly the contract the + // audit flagged as a replay vulnerability. h, tokenStore := newInternalHandlerWithTokenStore(t) mintRes, err := tokenStore.Mint(context.Background(), terminal.MintParams{ @@ -244,13 +252,21 @@ func TestProxyValidateTerminalToken_DoesNotConsumeToken(t *testing.T) { }) require.NoError(t, err) - for i := 0; i < 3; i++ { - _, err := h.ProxyValidateTerminalToken(context.Background(), connect.NewRequest(&pm.InternalValidateTerminalTokenRequest{ - SessionId: mintRes.SessionID, - Token: mintRes.Token, - })) - require.NoErrorf(t, err, "validation %d should succeed", i+1) - } + // First validation succeeds and consumes the token. + _, err = h.ProxyValidateTerminalToken(context.Background(), connect.NewRequest(&pm.InternalValidateTerminalTokenRequest{ + SessionId: mintRes.SessionID, + Token: mintRes.Token, + })) + require.NoError(t, err, "first validation should succeed") + + // Second validation with the same bearer fails with + // Unauthenticated — the token is gone. + _, err = h.ProxyValidateTerminalToken(context.Background(), connect.NewRequest(&pm.InternalValidateTerminalTokenRequest{ + SessionId: mintRes.SessionID, + Token: mintRes.Token, + })) + require.Error(t, err, "second validation must fail (single-use contract)") + assert.Equal(t, connect.CodeUnauthenticated, connect.CodeOf(err)) } func TestProxyValidateTerminalToken_UnknownSession(t *testing.T) { diff --git a/internal/api/registration_handler.go b/internal/api/registration_handler.go index 0f447c72..726d03b5 100644 --- a/internal/api/registration_handler.go +++ b/internal/api/registration_handler.go @@ -5,7 +5,9 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "fmt" "log/slog" + "net/url" "time" "connectrpc.com/connect" @@ -17,6 +19,75 @@ import ( "github.com/manchtools/power-manage/server/internal/store" ) +// ValidateGatewayURL returns nil when raw is a gateway URL an agent +// can actually connect to over mTLS. A surprising number of shapes +// pass `url.Parse` without being usable: +// - bare hostnames like `gateway.example.com` parse with Scheme="", +// Host="", Path="gateway.example.com" — the agent would try to +// dial a relative path; +// - `http://...` is refused because rc10 agents refuse h2c; +// - `wss://...` or other schemes are refused because the agent's +// gateway client uses HTTPS transport; +// - user-info (`https://user:pass@host/`) is refused because +// credentials in the URL are never the right answer and would +// leak on every enrollment response; +// - fragments are meaningless on the wire and refused to keep the +// shape tight. +// +// Used both at control server startup (fatal on violation, so a +// misconfiguration is visible at boot) and defensively in the +// registration handler before the URL is handed to the agent. +func ValidateGatewayURL(raw string) error { + if raw == "" { + return fmt.Errorf("gateway URL is empty") + } + u, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("gateway URL parse failed: %w", err) + } + if u.Scheme != "https" { + return fmt.Errorf("gateway URL must use https scheme, got %q", u.Scheme) + } + // u.Hostname() strips port and brackets so "https://:8443" (port + // only, no host) and "https://[::1]:443" (IPv6) both validate + // under the same rule. u.Host would accept ":8443" silently. + if u.Hostname() == "" { + return fmt.Errorf("gateway URL has no host — bare hostnames like %q are not absolute URLs", RedactGatewayURL(raw)) + } + if u.User != nil { + return fmt.Errorf("gateway URL must not contain userinfo (credentials in URL leak on every enrollment response)") + } + if u.Fragment != "" { + return fmt.Errorf("gateway URL must not contain a fragment") + } + return nil +} + +// RedactGatewayURL strips userinfo from a URL-shaped string for safe +// logging / panic messages. Exported so cmd/control/main.go can use +// the same redaction on the startup-log error path. +// +// If the input is unparseable, returns a placeholder rather than the +// raw value — a malformed url.Parse input that still carries +// credentials in a substring shouldn't leak just because the parser +// rejected it (e.g. "https://u:p@host:notaport" fails to parse as +// a URL but still contains "u:p" in-band). +func RedactGatewayURL(raw string) string { + if raw == "" { + return "" + } + u, err := url.Parse(raw) + if err != nil { + return "" + } + if u.User == nil { + return raw + } + // Rebuild without userinfo. + u.User = nil + return u.String() +} + // RegistrationHandler handles agent registration requests. type RegistrationHandler struct { store *store.Store @@ -25,8 +96,19 @@ type RegistrationHandler struct { logger *slog.Logger } -// NewRegistrationHandler creates a new registration handler. +// NewRegistrationHandler creates a new registration handler. Panics +// when gatewayURL fails ValidateGatewayURL — caller is expected to +// have validated at startup, so reaching the handler constructor +// with a bad value is a programmer error, not a runtime condition. +// Startup-time validation (cmd/control/main.go) surfaces the same +// check with a clean operator-facing error message. func NewRegistrationHandler(st *store.Store, certAuth *ca.CA, gatewayURL string, logger *slog.Logger) *RegistrationHandler { + if err := ValidateGatewayURL(gatewayURL); err != nil { + // Redact userinfo before panicking — a gateway URL that + // contains credentials (which the validator is rejecting) + // would otherwise leak them into the crash log. + panic(fmt.Sprintf("NewRegistrationHandler: invalid gateway URL %q: %v", RedactGatewayURL(gatewayURL), err)) + } return &RegistrationHandler{ store: st, ca: certAuth, @@ -41,6 +123,19 @@ func (h *RegistrationHandler) Register(ctx context.Context, req *connect.Request logger := h.logger.With("hostname", req.Msg.Hostname, "agent_version", req.Msg.AgentVersion) logger.Info("processing registration request") + // Defence in depth: the startup guard in cmd/control/main.go + // plus the NewRegistrationHandler constructor both run + // ValidateGatewayURL, so reaching this check with an invalid + // URL would mean both earlier layers regressed. We re-run the + // full validator (not just the emptiness check) so the agent + // never receives a URL shape that the URL validators missed — + // bare hostnames, http://, userinfo, etc. + if err := ValidateGatewayURL(h.gatewayURL); err != nil { + logger.Error("registration refused: gatewayURL failed validation", + "gateway_url", RedactGatewayURL(h.gatewayURL), "error", err) + return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeFailedPrecondition, "server misconfiguration: gateway URL is invalid") + } + // Validate CSR is present if len(req.Msg.Csr) == 0 { return nil, apiErrorCtx(ctx, ErrValidationFailed, connect.CodeInvalidArgument, "CSR is required") diff --git a/internal/api/registration_handler_test.go b/internal/api/registration_handler_test.go index 6513578a..3ae01374 100644 --- a/internal/api/registration_handler_test.go +++ b/internal/api/registration_handler_test.go @@ -112,7 +112,7 @@ func createTestTokenWithValue(t *testing.T, st *store.Store, actorID string, one func TestRegister_ValidToken(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") _, tokenValue := createTestTokenWithValue(t, st, adminID, false) @@ -130,7 +130,7 @@ func TestRegister_ValidToken(t *testing.T) { assert.NotEmpty(t, resp.Msg.DeviceId.Value) assert.NotEmpty(t, resp.Msg.Certificate) assert.NotEmpty(t, resp.Msg.CaCert) - assert.Equal(t, "wss://gateway.test:443", resp.Msg.GatewayUrl) + assert.Equal(t, "https://gateway.test:8080", resp.Msg.GatewayUrl) // Verify device projection exists device, err := st.Queries().GetDeviceByID(context.Background(), db.GetDeviceByIDParams{ @@ -143,7 +143,7 @@ func TestRegister_ValidToken(t *testing.T) { func TestRegister_OneTimeTokenDisabledAfterUse(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") _, tokenValue := createTestTokenWithValue(t, st, adminID, true) @@ -174,7 +174,7 @@ func TestRegister_OneTimeTokenDisabledAfterUse(t *testing.T) { func TestRegister_DisabledTokenFails(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") tokenID, tokenValue := createTestTokenWithValue(t, st, adminID, false) @@ -204,7 +204,7 @@ func TestRegister_DisabledTokenFails(t *testing.T) { func TestRegister_InvalidTokenFails(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) csr := generateCSR(t) _, err := h.Register(context.Background(), connect.NewRequest(&pm.RegisterRequest{ @@ -220,7 +220,7 @@ func TestRegister_InvalidTokenFails(t *testing.T) { func TestRegister_MissingCSRFails(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") _, tokenValue := createTestTokenWithValue(t, st, adminID, false) @@ -238,7 +238,7 @@ func TestRegister_MissingCSRFails(t *testing.T) { func TestRegister_ReusableTokenAllowsMultipleUses(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") _, tokenValue := createTestTokenWithValue(t, st, adminID, false) @@ -259,7 +259,7 @@ func TestRegister_ReusableTokenAllowsMultipleUses(t *testing.T) { func TestRegister_ExpiredTokenFails(t *testing.T) { st := testutil.SetupPostgres(t) testCA := newTestCA(t) - h := api.NewRegistrationHandler(st, testCA, "wss://gateway.test:443", slog.Default()) + h := api.NewRegistrationHandler(st, testCA, "https://gateway.test:8080", slog.Default()) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") ctx := context.Background() diff --git a/internal/api/service.go b/internal/api/service.go index 32424427..b8c5bb34 100644 --- a/internal/api/service.go +++ b/internal/api/service.go @@ -61,7 +61,7 @@ func NewControlService(st *store.Store, jwtManager *auth.JWTManager, signer Acti settingsHandler := NewSettingsHandler(st, logger, systemActions) return &ControlService{ registration: NewRegistrationHandler(st, certAuth, gatewayURL, logger), - auth: NewAuthHandler(st, logger.With("component", "auth_handler"), jwtManager), + auth: NewAuthHandler(st, logger.With("component", "auth_handler"), jwtManager, cfg.PasswordAuthEnabled), totp: NewTOTPHandler(st, logger.With("component", "totp_handler"), jwtManager, enc, ""), user: NewUserHandler(st, logger.With("component", "user_handler"), systemActions), device: NewDeviceHandler(st, enc, logger.With("component", "device_handler")), diff --git a/internal/api/validate_gateway_url_test.go b/internal/api/validate_gateway_url_test.go new file mode 100644 index 00000000..49e61147 --- /dev/null +++ b/internal/api/validate_gateway_url_test.go @@ -0,0 +1,68 @@ +package api_test + +import ( + "strings" + "testing" + + "github.com/manchtools/power-manage/server/internal/api" +) + +// TestValidateGatewayURL covers the shapes the control server may be +// handed via CONTROL_GATEWAY_URL and the ones registration_handler +// rechecks defensively. Each case captures a real operator footgun +// rc10 reviewers flagged: +// +// - empty string — rc10 closed the "enroll with empty URL" path. +// - bare hostname like "gateway.example.com" — parses via url.Parse +// without error but isn't an absolute URL; the agent would dial +// a relative path and fail with a cryptic error. +// - http:// — agents refuse h2c as of rc10 (see agent main.go). +// - ws:// / wss:// — agent uses HTTPS transport for the streaming +// client; accepting wss here would confuse the operator. +// - userinfo — credentials in the URL leak on every enrollment +// response and are never the right answer. +// - fragment — meaningless on the wire. +func TestValidateGatewayURL(t *testing.T) { + cases := []struct { + name string + in string + wantErr bool + wantWord string // substring that must appear in the error message + }{ + {"empty", "", true, "empty"}, + {"bare hostname", "gateway.example.com", true, "scheme"}, + {"http downgrade", "http://gateway.example.com", true, "https"}, + {"wss scheme", "wss://gateway.example.com", true, "https"}, + {"ws scheme", "ws://gateway.example.com", true, "https"}, + {"userinfo", "https://user:pass@gateway.example.com", true, "userinfo"}, + {"userinfo username only", "https://user@gateway.example.com", true, "userinfo"}, + {"fragment", "https://gateway.example.com#frag", true, "fragment"}, + {"host only", "https://", true, "host"}, + // CR called out that u.Host would accept ":8443" silently + // (it treats the empty part as a host with a port). Switching + // to u.Hostname() plus this regression test closes the gap. + {"port without hostname", "https://:8443", true, "host"}, + + {"happy path host", "https://gateway.example.com", false, ""}, + {"happy path with port", "https://gateway.example.com:8443", false, ""}, + {"happy path with path", "https://gateway.example.com/gw/01ABC", false, ""}, + {"happy path with trailing slash", "https://gateway.example.com/", false, ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := api.ValidateGatewayURL(tc.in) + if tc.wantErr { + if err == nil { + t.Fatalf("ValidateGatewayURL(%q) = nil, want error containing %q", tc.in, tc.wantWord) + } + if tc.wantWord != "" && !strings.Contains(err.Error(), tc.wantWord) { + t.Errorf("ValidateGatewayURL(%q) err = %q, want substring %q", tc.in, err.Error(), tc.wantWord) + } + return + } + if err != nil { + t.Errorf("ValidateGatewayURL(%q) = %v, want nil", tc.in, err) + } + }) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index a7c84d09..49c23b1f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,6 +2,7 @@ package config import ( + "fmt" "os" "strconv" "time" @@ -71,10 +72,11 @@ type Config struct { // Example: "gateway.example.com" BootstrapHost string - // Web listener for non-mTLS traffic (terminal WebSocket). Uses - // standard TLS (server cert only, no client cert) so web browsers - // can connect. Empty disables the web listener — terminal - // sessions won't work but agent connections are unaffected. + // Web listener for non-mTLS traffic (terminal WebSocket). This + // listener serves cleartext HTTP on the private network; public TLS + // terminates at Traefik before proxying to it. Empty disables the + // web listener — terminal sessions won't work but agent connections + // are unaffected. WebListenAddr string // InternalURL is the mTLS URL the control server uses to call @@ -164,7 +166,8 @@ func FromEnv() *Config { // * RootKey = traefik — matches Traefik default --providers.redis.rootkey // * MTLSHost = GATEWAY_DOMAIN (not Traefik-prefixed — one name per thing) // * MTLSEntryPoint = websecure (same :443 as control, SNI-separated) - // * TTYHost = GATEWAY_TTY_DOMAIN (falls back to empty → TTY router disabled) + // * TTYHost = GATEWAY_TTY_DOMAIN, then GATEWAY_DOMAIN + // (Validate rejects unsafe shared-host terminal+self-register configs) // * TTYEntryPoint = websecure // * TTYCertResolver = letsencrypt // @@ -192,7 +195,25 @@ func FromEnv() *Config { TraefikMTLSHost: firstNonEmpty(os.Getenv("GATEWAY_TRAEFIK_MTLS_HOST"), os.Getenv("GATEWAY_DOMAIN")), TraefikMTLSBackend: getEnv("GATEWAY_TRAEFIK_MTLS_BACKEND", ""), TraefikMTLSEntryPoint: getEnv("GATEWAY_TRAEFIK_MTLS_ENTRYPOINT", "websecure"), - TraefikTTYHost: firstNonEmpty(os.Getenv("GATEWAY_TRAEFIK_TTY_HOST"), os.Getenv("GATEWAY_TTY_DOMAIN")), + // TTYHost falls back through GATEWAY_TTY_DOMAIN (dedicated + // TTY subdomain) to GATEWAY_DOMAIN. Without this chain rc9 + // hit the hidden empty-host trap that crashed gateway + // startup in staging. + // + // Note on the GATEWAY_DOMAIN rung: it only works when the + // terminal WebSocket listener is disabled + // (GATEWAY_WEB_LISTEN_ADDR empty) or when Traefik self- + // registration is off. With terminal + self-register both + // enabled, a shared TTY/MTLS hostname is refused by + // Validate() below — Traefik's TCP passthrough would + // match the shared SNI and shadow the TTY HTTP router, + // silently breaking WebSocket sessions. Single-domain + // terminal deployments need a separate GATEWAY_TTY_DOMAIN. + TraefikTTYHost: firstNonEmpty( + os.Getenv("GATEWAY_TRAEFIK_TTY_HOST"), + os.Getenv("GATEWAY_TTY_DOMAIN"), + os.Getenv("GATEWAY_DOMAIN"), + ), TraefikTTYBackend: getEnv("GATEWAY_TRAEFIK_TTY_BACKEND", ""), TraefikTTYEntryPoint: getEnv("GATEWAY_TRAEFIK_TTY_ENTRYPOINT", "websecure"), TraefikTTYCertResolver: getEnv("GATEWAY_TRAEFIK_TTY_CERT_RESOLVER", "letsencrypt"), @@ -201,6 +222,47 @@ func FromEnv() *Config { } } +// Validate returns a non-nil error when the loaded config has a +// combination that the gateway cannot serve coherently. Called once +// at startup from cmd/gateway/main.go; keeping it on the Config +// struct (rather than inline in main) so tests can exercise the +// shape checks without booting a full process. +func (c *Config) Validate() error { + // TTY / MTLS host collision: when Traefik self-registration is + // on AND the terminal feature is in play AND the mTLS + TTY + // routers bind the same entrypoint, the shared hostname means + // Traefik's TCP-passthrough router for mTLS wins the SNI match + // and the TTY HTTP router never gets the request. The TLS + // handshake lands on an agent-mTLS backend that isn't speaking + // HTTP, so the WebSocket upgrade fails silently. + // + // Split the preconditions explicitly so the check handles both + // the auto-derived backend case (operator only set + // GATEWAY_WEB_LISTEN_ADDR) and the explicit backend case + // (operator set GATEWAY_TRAEFIK_TTY_BACKEND directly). Also + // narrow to "same entrypoint" — different entrypoints mean the + // routers don't actually collide even on a shared host, so + // flagging that shape would be a false positive for + // bring-your-own-Traefik topologies. + // + // Refuse startup with a clear message instead of letting the + // operator discover this when a terminal session silently fails. + terminalEnabled := c.WebListenAddr != "" || c.TraefikTTYBackend != "" + entrypointsCollide := c.TraefikTTYEntryPoint != "" && c.TraefikTTYEntryPoint == c.TraefikMTLSEntryPoint + if c.TraefikSelfRegister && + terminalEnabled && + entrypointsCollide && + c.TraefikTTYHost != "" && + c.TraefikMTLSHost != "" && + c.TraefikTTYHost == c.TraefikMTLSHost { + return fmt.Errorf( + "GATEWAY_TTY_DOMAIN / TraefikTTYHost cannot equal GATEWAY_DOMAIN / TraefikMTLSHost when the terminal router and mTLS router share an entrypoint (both %q on entrypoint %q): Traefik TCP passthrough for mTLS matches the same SNI as the TTY HTTP router and breaks WebSocket sessions — set a distinct GATEWAY_TTY_DOMAIN or split the entrypoints", + c.TraefikTTYHost, c.TraefikTTYEntryPoint, + ) + } + return nil +} + // firstNonEmpty returns the first argument that isn't the empty // string. Used to resolve an env-var name (the legacy one) with a // more-preferred name as the primary source. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 8bde0dfa..686858fb 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -121,6 +121,105 @@ func TestFromEnv_TraefikTTYCertResolver(t *testing.T) { } } +// TestValidate_TTYMTLSHostCollision captures the config-shape invariant +// CodeRabbit flagged on the rc10 review: when the terminal WebSocket +// listener is enabled AND Traefik self-registration is on, the TTY +// host must not equal the mTLS host. If they match, Traefik's TCP +// passthrough router for the shared SNI wins over the TTY HTTP router +// and the WebSocket handshake fails against the mTLS backend. +func TestValidate_TTYMTLSHostCollision(t *testing.T) { + cases := []struct { + name string + cfg Config + wantErr bool + }{ + { + name: "collision with terminal enabled (WebListenAddr) → error", + cfg: Config{ + TraefikSelfRegister: true, + WebListenAddr: ":8443", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "gw.example.com", + TraefikMTLSEntryPoint: "websecure", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: true, + }, + { + name: "collision with terminal enabled (explicit TTYBackend) → error", + cfg: Config{ + TraefikSelfRegister: true, + TraefikTTYBackend: "http://10.0.0.5:8443", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "gw.example.com", + TraefikMTLSEntryPoint: "websecure", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: true, + }, + { + name: "collision but terminal disabled → OK", + cfg: Config{ + TraefikSelfRegister: true, + WebListenAddr: "", // terminal off + TraefikTTYBackend: "", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "gw.example.com", + TraefikMTLSEntryPoint: "websecure", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: false, + }, + { + name: "distinct hosts with terminal enabled → OK", + cfg: Config{ + TraefikSelfRegister: true, + WebListenAddr: ":8443", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "tty.example.com", + TraefikMTLSEntryPoint: "websecure", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: false, + }, + { + name: "shared host but different entrypoints → OK (routers don't collide)", + cfg: Config{ + TraefikSelfRegister: true, + WebListenAddr: ":8443", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "gw.example.com", + TraefikMTLSEntryPoint: "mtls", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: false, + }, + { + name: "collision but self-register off → OK (operator owns routing)", + cfg: Config{ + TraefikSelfRegister: false, + WebListenAddr: ":8443", + TraefikMTLSHost: "gw.example.com", + TraefikTTYHost: "gw.example.com", + TraefikMTLSEntryPoint: "websecure", + TraefikTTYEntryPoint: "websecure", + }, + wantErr: false, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := tc.cfg.Validate() + if tc.wantErr && err == nil { + t.Fatalf("Validate() = nil, want error for %+v", tc.cfg) + } + if !tc.wantErr && err != nil { + t.Fatalf("Validate() = %v, want nil for %+v", err, tc.cfg) + } + }) + } +} + func TestGetEnvInt_ValidValues(t *testing.T) { tests := []struct { name string diff --git a/internal/control/inbox_worker.go b/internal/control/inbox_worker.go index e2a62112..ecedd43f 100644 --- a/internal/control/inbox_worker.go +++ b/internal/control/inbox_worker.go @@ -535,7 +535,56 @@ func (w *InboxWorker) handleRevokeLuksDeviceKeyResult(ctx context.Context, t *as "success", payload.Success, ) - luksStreamID := ulid.Make().String() + // Look up the stream ID minted at request time so the Revoked / + // Failed event lands on the SAME stream as the Requested / + // Dispatched phases. Earlier versions generated a fresh ULID + // here, which split every revocation across two streams and + // broke the projection's three-phase stitch — fixed in rc10. + // + // Correctness assumption: at most one outstanding revocation + // per (device, action) at a time. Enforced at the API layer: + // RevokeLuksDeviceKey checks the projection for an already- + // dispatched-and-unterminal request before accepting a new + // one, so duplicates via concurrent operator clicks are + // rejected upstream. If that invariant ever regresses, the + // ORDER BY sequence_num DESC + LIMIT 1 here will pick the + // LATEST matching request — which is the expected "the + // operator re-requested and here's the result" semantic. + // Older abandoned streams would then lack a terminal event; + // not a correctness issue for the projection (it keys by + // stream_id), but worth flagging for the audit export. + // + // If the lookup fails (e.g. the Requested event never made it + // to disk because the original API call crashed), fall back to + // a fresh stream ID so we still record the Failed outcome + // durably — an orphan Failed event is better than dropping the + // agent-reported failure on the floor. + luksStreamID, err := w.store.Queries().GetLuksRevocationStreamID(ctx, db.GetLuksRevocationStreamIDParams{ + DeviceID: payload.DeviceID, + ActionID: payload.ActionID, + }) + switch { + case err == nil: + // Happy path — stream ID recovered. + case errors.Is(err, pgx.ErrNoRows): + // Genuinely absent: the Requested event never landed + // (original RPC crashed before append). Fall back to a + // fresh ULID so we still record the terminal outcome — + // an orphan Failed event is better than silently dropping + // the agent-reported failure on the floor. + w.logger.Warn("LUKS revocation stream ID not found — appending to a fresh stream; projection will show only the terminal event", + "device_id", payload.DeviceID, + "action_id", payload.ActionID, + ) + luksStreamID = ulid.Make().String() + default: + // Transient DB / context error. Return so Asynq retries; + // previously we masked these as "not found" and forked + // the stream, which would compound audit fragmentation + // under DB flakes. + return fmt.Errorf("look up LUKS revocation stream ID for device %s action %s: %w", payload.DeviceID, payload.ActionID, err) + } + if payload.Success { return w.store.AppendEvent(ctx, store.Event{ StreamType: "luks_key", diff --git a/internal/store/generated/luks.sql.go b/internal/store/generated/luks.sql.go index d1273f8f..2e0df1bf 100644 --- a/internal/store/generated/luks.sql.go +++ b/internal/store/generated/luks.sql.go @@ -166,6 +166,36 @@ func (q *Queries) GetLuksKeyHistory(ctx context.Context, deviceID string) ([]Luk return items, nil } +const getLuksRevocationStreamID = `-- name: GetLuksRevocationStreamID :one +SELECT stream_id +FROM events +WHERE stream_type = 'luks_key' + AND event_type IN ('LuksDeviceKeyRevocationRequested', 'LuksDeviceKeyRevocationDispatched') + AND data->>'device_id' = $1::text + AND data->>'action_id' = $2::text +ORDER BY sequence_num DESC +LIMIT 1 +` + +type GetLuksRevocationStreamIDParams struct { + DeviceID string `json:"device_id"` + ActionID string `json:"action_id"` +} + +// GetLuksRevocationStreamID looks up the luks_key event-stream ID that +// was minted when api/device_handler.go appended the +// LuksDeviceKeyRevocationRequested event for this (device, action). +// The inbox worker uses it to append the final Revoked / Failed event +// to the SAME stream so the three-phase projection stitches together. +// Returns the most recent request if somehow there are multiple (there +// should only ever be one; LIMIT 1 is belt-and-braces). +func (q *Queries) GetLuksRevocationStreamID(ctx context.Context, arg GetLuksRevocationStreamIDParams) (string, error) { + row := q.db.QueryRow(ctx, getLuksRevocationStreamID, arg.DeviceID, arg.ActionID) + var stream_id string + err := row.Scan(&stream_id) + return stream_id, err +} + const validateAndConsumeLuksToken = `-- name: ValidateAndConsumeLuksToken :one UPDATE luks_tokens SET used = TRUE diff --git a/internal/store/generated/models.go b/internal/store/generated/models.go index 4213c6d2..66d99fdf 100644 --- a/internal/store/generated/models.go +++ b/internal/store/generated/models.go @@ -388,6 +388,19 @@ type ScimGroupMappingProjection struct { ProjectionVersion int64 `json:"projection_version"` } +type SecurityAlertsProjection struct { + EventID uuid.UUID `json:"event_id"` + DeviceID string `json:"device_id"` + AlertType string `json:"alert_type"` + Message string `json:"message"` + Details []byte `json:"details"` + RaisedAt time.Time `json:"raised_at"` + Acknowledged bool `json:"acknowledged"` + AcknowledgedAt *time.Time `json:"acknowledged_at"` + AcknowledgedBy *string `json:"acknowledged_by"` + CreatedAt time.Time `json:"created_at"` +} + type ServerSettingsProjection struct { ID string `json:"id"` UserProvisioningEnabled bool `json:"user_provisioning_enabled"` diff --git a/internal/store/generated/security_alerts.sql.go b/internal/store/generated/security_alerts.sql.go new file mode 100644 index 00000000..76cf048a --- /dev/null +++ b/internal/store/generated/security_alerts.sql.go @@ -0,0 +1,219 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: security_alerts.sql + +package generated + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +const countSecurityAlertsForDevice = `-- name: CountSecurityAlertsForDevice :one +SELECT COUNT(*)::bigint +FROM security_alerts_projection +WHERE device_id = $1 + AND ($2::bool OR NOT acknowledged) +` + +type CountSecurityAlertsForDeviceParams struct { + DeviceID string `json:"device_id"` + IncludeAcknowledged bool `json:"include_acknowledged"` +} + +// Companion count for ListSecurityAlertsForDevice. Needed by +// buildNextPageToken to compute totalCount and emit a correct +// next-page token; mirrors the same include_acknowledged filter +// semantics as the list query so the two stay in lockstep. +func (q *Queries) CountSecurityAlertsForDevice(ctx context.Context, arg CountSecurityAlertsForDeviceParams) (int64, error) { + row := q.db.QueryRow(ctx, countSecurityAlertsForDevice, arg.DeviceID, arg.IncludeAcknowledged) + var column_1 int64 + err := row.Scan(&column_1) + return column_1, err +} + +const countUnacknowledgedSecurityAlerts = `-- name: CountUnacknowledgedSecurityAlerts :one +SELECT COUNT(*)::bigint FROM security_alerts_projection WHERE NOT acknowledged +` + +// COUNT(*) in PostgreSQL is bigint; keep the full precision so +// buildNextPageToken (which works in int64) doesn't see a silently +// truncated int32 once device counts climb past 2.1B aggregate +// alerts across all time. Matches the pagination helper contract. +func (q *Queries) CountUnacknowledgedSecurityAlerts(ctx context.Context) (int64, error) { + row := q.db.QueryRow(ctx, countUnacknowledgedSecurityAlerts) + var column_1 int64 + err := row.Scan(&column_1) + return column_1, err +} + +const getSecurityAlert = `-- name: GetSecurityAlert :one +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE event_id = $1 +` + +type GetSecurityAlertRow struct { + EventID uuid.UUID `json:"event_id"` + DeviceID string `json:"device_id"` + AlertType string `json:"alert_type"` + Message string `json:"message"` + Details []byte `json:"details"` + RaisedAt time.Time `json:"raised_at"` + Acknowledged bool `json:"acknowledged"` + AcknowledgedAt *time.Time `json:"acknowledged_at"` + AcknowledgedBy *string `json:"acknowledged_by"` +} + +func (q *Queries) GetSecurityAlert(ctx context.Context, eventID uuid.UUID) (GetSecurityAlertRow, error) { + row := q.db.QueryRow(ctx, getSecurityAlert, eventID) + var i GetSecurityAlertRow + err := row.Scan( + &i.EventID, + &i.DeviceID, + &i.AlertType, + &i.Message, + &i.Details, + &i.RaisedAt, + &i.Acknowledged, + &i.AcknowledgedAt, + &i.AcknowledgedBy, + ) + return i, err +} + +const listSecurityAlertsForDevice = `-- name: ListSecurityAlertsForDevice :many + +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE device_id = $1 + AND ($2::bool OR NOT acknowledged) +ORDER BY raised_at DESC +LIMIT $4::int +OFFSET $3::int +` + +type ListSecurityAlertsForDeviceParams struct { + DeviceID string `json:"device_id"` + IncludeAcknowledged bool `json:"include_acknowledged"` + PageOffset int32 `json:"page_offset"` + PageSize int32 `json:"page_size"` +} + +type ListSecurityAlertsForDeviceRow struct { + EventID uuid.UUID `json:"event_id"` + DeviceID string `json:"device_id"` + AlertType string `json:"alert_type"` + Message string `json:"message"` + Details []byte `json:"details"` + RaisedAt time.Time `json:"raised_at"` + Acknowledged bool `json:"acknowledged"` + AcknowledgedAt *time.Time `json:"acknowledged_at"` + AcknowledgedBy *string `json:"acknowledged_by"` +} + +// Phase 1 read queries for the security_alerts_projection (added in +// migration 010). The RPC surface (ControlService.ListSecurityAlerts, +// AcknowledgeSecurityAlert) and the web UI wiring land in a follow-up +// PR because they need proto changes; the queries below are the +// Go-side primitives those handlers will call. +// +// Keeping them in this PR means the projection is not a dead table — +// internal callers and tests can already exercise it, and future +// handler work only needs to wire the proto envelope. +func (q *Queries) ListSecurityAlertsForDevice(ctx context.Context, arg ListSecurityAlertsForDeviceParams) ([]ListSecurityAlertsForDeviceRow, error) { + rows, err := q.db.Query(ctx, listSecurityAlertsForDevice, + arg.DeviceID, + arg.IncludeAcknowledged, + arg.PageOffset, + arg.PageSize, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListSecurityAlertsForDeviceRow{} + for rows.Next() { + var i ListSecurityAlertsForDeviceRow + if err := rows.Scan( + &i.EventID, + &i.DeviceID, + &i.AlertType, + &i.Message, + &i.Details, + &i.RaisedAt, + &i.Acknowledged, + &i.AcknowledgedAt, + &i.AcknowledgedBy, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listUnacknowledgedSecurityAlerts = `-- name: ListUnacknowledgedSecurityAlerts :many +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE NOT acknowledged +ORDER BY raised_at DESC +LIMIT $2::int +OFFSET $1::int +` + +type ListUnacknowledgedSecurityAlertsParams struct { + PageOffset int32 `json:"page_offset"` + PageSize int32 `json:"page_size"` +} + +type ListUnacknowledgedSecurityAlertsRow struct { + EventID uuid.UUID `json:"event_id"` + DeviceID string `json:"device_id"` + AlertType string `json:"alert_type"` + Message string `json:"message"` + Details []byte `json:"details"` + RaisedAt time.Time `json:"raised_at"` + Acknowledged bool `json:"acknowledged"` + AcknowledgedAt *time.Time `json:"acknowledged_at"` + AcknowledgedBy *string `json:"acknowledged_by"` +} + +func (q *Queries) ListUnacknowledgedSecurityAlerts(ctx context.Context, arg ListUnacknowledgedSecurityAlertsParams) ([]ListUnacknowledgedSecurityAlertsRow, error) { + rows, err := q.db.Query(ctx, listUnacknowledgedSecurityAlerts, arg.PageOffset, arg.PageSize) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListUnacknowledgedSecurityAlertsRow{} + for rows.Next() { + var i ListUnacknowledgedSecurityAlertsRow + if err := rows.Scan( + &i.EventID, + &i.DeviceID, + &i.AlertType, + &i.Message, + &i.Details, + &i.RaisedAt, + &i.Acknowledged, + &i.AcknowledgedAt, + &i.AcknowledgedBy, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/store/migrations/010_security_alerts_projection.sql b/internal/store/migrations/010_security_alerts_projection.sql new file mode 100644 index 00000000..fb75a833 --- /dev/null +++ b/internal/store/migrations/010_security_alerts_projection.sql @@ -0,0 +1,125 @@ +-- SecurityAlert projection — enables the UI to list security alerts +-- without scanning the raw events table. +-- +-- Rationale: agents emit SecurityAlert events via the inbox worker +-- (internal/control/inbox_worker.go handleSecurityAlert). They are +-- persisted on the device stream, but nothing projected them into a +-- queryable table, so compliance / dashboard consumers had to either +-- scan the append-only events log or ignore alerts entirely. The +-- rc10 audit flagged this as silent data loss for the SIEM path. +-- +-- Shape follows the LpsPasswordRotated / LuksKey rotation pattern: +-- a derived projection keyed by a new UUID with an acknowledged +-- flag so "unacknowledged alerts" is a cheap query. alert_type is +-- the free-form tag from the payload (e.g. "file_integrity_violation", +-- "auditd_rule_trip"); details is a small key/value blob from the +-- agent's evidence bundle. +-- +-- Wiring: rather than rewriting project_event() (the central +-- dispatcher in migration 004 uses BEGIN/EXCEPTION isolation per +-- case arm), we install a second AFTER INSERT trigger on events +-- that ONLY fires for SecurityAlert-shaped rows. This keeps the +-- change additive and prevents an accidental dropped case arm from +-- breaking an unrelated projection. + +-- +goose Up + +-- The primary key is the originating event_id rather than a fresh +-- UUID. This is the event-sourcing idempotency pattern: if the +-- projection is ever replayed (backfill, rebuild, trigger re-fire), +-- the same SecurityAlert event produces the same row, so ON CONFLICT +-- DO NOTHING prevents duplicates without needing deduplication +-- logic in the acknowledge path. SecurityAlertAcknowledged carries +-- the alert_id explicitly in its payload and UPDATEs by that key. +CREATE TABLE security_alerts_projection ( + event_id UUID PRIMARY KEY REFERENCES events(id), + device_id TEXT NOT NULL, + alert_type TEXT NOT NULL, + message TEXT NOT NULL, + details JSONB, + raised_at TIMESTAMPTZ NOT NULL, + acknowledged BOOLEAN NOT NULL DEFAULT FALSE, + acknowledged_at TIMESTAMPTZ, + acknowledged_by TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_security_alerts_device ON security_alerts_projection(device_id, acknowledged, raised_at DESC); +CREATE INDEX idx_security_alerts_type ON security_alerts_projection(alert_type, raised_at DESC); +CREATE INDEX idx_security_alerts_unack ON security_alerts_projection(acknowledged, raised_at DESC) WHERE acknowledged = FALSE; + +-- +goose StatementBegin +CREATE OR REPLACE FUNCTION project_security_alert_event(event events) RETURNS void AS $$ +BEGIN + CASE event.event_type + WHEN 'SecurityAlert' THEN + INSERT INTO security_alerts_projection ( + event_id, device_id, alert_type, message, details, raised_at + ) VALUES ( + event.id, + event.stream_id, + event.data->>'alert_type', + event.data->>'message', + event.data->'details', + event.occurred_at + ) + ON CONFLICT (event_id) DO NOTHING; + WHEN 'SecurityAlertAcknowledged' THEN + -- Cast the right-hand side to UUID (not event_id to + -- TEXT) so the primary-key index on event_id is used + -- and a malformed alert_id surfaces as a projection + -- error instead of silently full-scanning and matching + -- zero rows. + UPDATE security_alerts_projection + SET acknowledged = TRUE, + acknowledged_at = event.occurred_at, + acknowledged_by = event.data->>'acknowledged_by' + WHERE event_id = (event.data->>'alert_id')::uuid; + IF NOT FOUND THEN + -- Out-of-order replay (ack before the alert row + -- exists) or an alert that was purged. Raise so + -- the sidecar trigger's EXCEPTION handler logs it + -- into projection_errors for operator visibility; + -- previously this was a silent no-op. + RAISE EXCEPTION 'SecurityAlertAcknowledged references unknown alert_id=%', event.data->>'alert_id'; + END IF; + ELSE + NULL; + END CASE; +END; +$$ LANGUAGE plpgsql; +-- +goose StatementEnd + +-- Sidecar trigger: fires AFTER INSERT alongside the existing +-- event_projector trigger, but only for the SecurityAlert event +-- types so unrelated streams are unaffected. Error isolation +-- follows the same pattern as the central dispatcher: failures +-- route to projection_errors instead of aborting the event append. +-- +goose StatementBegin +CREATE OR REPLACE FUNCTION project_security_alert_trigger() RETURNS TRIGGER AS $$ +BEGIN + IF NEW.stream_type = 'device' + AND NEW.event_type IN ('SecurityAlert', 'SecurityAlertAcknowledged') THEN + BEGIN + PERFORM project_security_alert_event(NEW); + EXCEPTION WHEN OTHERS THEN + INSERT INTO projection_errors (event_id, event_type, stream_type, error_message) + VALUES (NEW.id, NEW.event_type, NEW.stream_type, SQLERRM); + END; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; +-- +goose StatementEnd + +CREATE TRIGGER security_alert_projector + AFTER INSERT ON events + FOR EACH ROW + EXECUTE FUNCTION project_security_alert_trigger(); + +-- +goose Down + +DROP TRIGGER IF EXISTS security_alert_projector ON events; +DROP FUNCTION IF EXISTS project_security_alert_trigger; +DROP FUNCTION IF EXISTS project_security_alert_event; +DROP TABLE IF EXISTS security_alerts_projection; diff --git a/internal/store/queries/luks.sql b/internal/store/queries/luks.sql index f59946e3..99ba059e 100644 --- a/internal/store/queries/luks.sql +++ b/internal/store/queries/luks.sql @@ -31,3 +31,21 @@ WHERE token = $1 AND NOT used AND expires_at > NOW() RETURNING *; + +-- GetLuksRevocationStreamID looks up the luks_key event-stream ID that +-- was minted when api/device_handler.go appended the +-- LuksDeviceKeyRevocationRequested event for this (device, action). +-- The inbox worker uses it to append the final Revoked / Failed event +-- to the SAME stream so the three-phase projection stitches together. +-- Returns the most recent request if somehow there are multiple (there +-- should only ever be one; LIMIT 1 is belt-and-braces). +-- +-- name: GetLuksRevocationStreamID :one +SELECT stream_id +FROM events +WHERE stream_type = 'luks_key' + AND event_type IN ('LuksDeviceKeyRevocationRequested', 'LuksDeviceKeyRevocationDispatched') + AND data->>'device_id' = sqlc.arg(device_id)::text + AND data->>'action_id' = sqlc.arg(action_id)::text +ORDER BY sequence_num DESC +LIMIT 1; diff --git a/internal/store/queries/security_alerts.sql b/internal/store/queries/security_alerts.sql new file mode 100644 index 00000000..589fb902 --- /dev/null +++ b/internal/store/queries/security_alerts.sql @@ -0,0 +1,51 @@ +-- Phase 1 read queries for the security_alerts_projection (added in +-- migration 010). The RPC surface (ControlService.ListSecurityAlerts, +-- AcknowledgeSecurityAlert) and the web UI wiring land in a follow-up +-- PR because they need proto changes; the queries below are the +-- Go-side primitives those handlers will call. +-- +-- Keeping them in this PR means the projection is not a dead table — +-- internal callers and tests can already exercise it, and future +-- handler work only needs to wire the proto envelope. + +-- name: ListSecurityAlertsForDevice :many +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE device_id = $1 + AND (sqlc.arg(include_acknowledged)::bool OR NOT acknowledged) +ORDER BY raised_at DESC +LIMIT sqlc.arg(page_size)::int +OFFSET sqlc.arg(page_offset)::int; + +-- name: ListUnacknowledgedSecurityAlerts :many +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE NOT acknowledged +ORDER BY raised_at DESC +LIMIT sqlc.arg(page_size)::int +OFFSET sqlc.arg(page_offset)::int; + +-- name: GetSecurityAlert :one +SELECT event_id, device_id, alert_type, message, details, raised_at, + acknowledged, acknowledged_at, acknowledged_by +FROM security_alerts_projection +WHERE event_id = $1; + +-- COUNT(*) in PostgreSQL is bigint; keep the full precision so +-- buildNextPageToken (which works in int64) doesn't see a silently +-- truncated int32 once device counts climb past 2.1B aggregate +-- alerts across all time. Matches the pagination helper contract. +-- name: CountUnacknowledgedSecurityAlerts :one +SELECT COUNT(*)::bigint FROM security_alerts_projection WHERE NOT acknowledged; + +-- Companion count for ListSecurityAlertsForDevice. Needed by +-- buildNextPageToken to compute totalCount and emit a correct +-- next-page token; mirrors the same include_acknowledged filter +-- semantics as the list query so the two stay in lockstep. +-- name: CountSecurityAlertsForDevice :one +SELECT COUNT(*)::bigint +FROM security_alerts_projection +WHERE device_id = $1 + AND (sqlc.arg(include_acknowledged)::bool OR NOT acknowledged); diff --git a/internal/terminal/fake_backend.go b/internal/terminal/fake_backend.go index 87c585f5..babbbca7 100644 --- a/internal/terminal/fake_backend.go +++ b/internal/terminal/fake_backend.go @@ -70,3 +70,22 @@ func (b *FakeBackend) Delete(ctx context.Context, sessionID string) error { delete(b.values, sessionID) return nil } + +// GetAndDelete mirrors the Valkey GETDEL primitive: returns the payload +// and removes the entry in a single atomic step under the mutex, so +// two concurrent callers cannot both observe it. Essential for the +// single-use token semantics Validate relies on. +func (b *FakeBackend) GetAndDelete(ctx context.Context, sessionID string) ([]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + entry, ok := b.values[sessionID] + if !ok { + return nil, ErrTokenNotFound + } + if !b.now().Before(entry.expiresAt) { + delete(b.values, sessionID) + return nil, ErrTokenNotFound + } + delete(b.values, sessionID) + return append([]byte(nil), entry.payload...), nil +} diff --git a/internal/terminal/store.go b/internal/terminal/store.go index 1ea102b1..a27a98df 100644 --- a/internal/terminal/store.go +++ b/internal/terminal/store.go @@ -101,6 +101,16 @@ type SessionBackend interface { // Delete removes the session_id. Idempotent: returns nil whether // or not the key existed. Delete(ctx context.Context, sessionID string) error + // GetAndDelete atomically returns the payload and removes the + // session_id in one operation. Used by Validate to enforce + // single-use tokens: two concurrent connect attempts with the + // same bearer can only succeed once — the loser sees + // ErrTokenNotFound. Implementations must use a primitive that + // cannot race (Valkey GETDEL, an in-process mutex on the fake, + // etc.). A naïve Get-then-Delete pair does NOT satisfy this + // contract; returning a nil payload and nil error MUST be + // translated to ErrTokenNotFound. + GetAndDelete(ctx context.Context, sessionID string) ([]byte, error) } // TokenStore is the high-level façade used by the API handlers. It @@ -242,20 +252,45 @@ func (s *TokenStore) Lookup(ctx context.Context, sessionID string) (*Session, er } // Validate verifies that the supplied bearer token matches the one -// stored for the given session_id and returns the Session. Used by -// the gateway-side validation path (will be wired via an internal -// RPC in a follow-up PR). Distinguishes ErrTokenNotFound (expired or -// never minted) from ErrTokenMismatch (forgery attempt) so the audit -// log can record them differently. +// stored for the given session_id, atomically consumes the token on +// success, and returns the Session. Single-use: a second call with +// the same bearer returns ErrTokenNotFound even within the TTL. +// +// Distinguishes ErrTokenNotFound (expired, never minted, or already +// consumed) from ErrTokenMismatch (bearer forgery attempt) so the +// audit log can record forgeries separately. Used by the +// gateway-side InternalService.ValidateTerminalToken path. +// +// On mismatch the session entry is re-persisted with the same +// remaining TTL so a forged bearer cannot DoS a legitimate session +// that has not yet been claimed. (GETDEL has already removed it; if +// we did not re-set, the real client's subsequent Validate would +// hit ErrTokenNotFound.) func (s *TokenStore) Validate(ctx context.Context, sessionID, bearerToken string) (*Session, error) { - session, err := s.Lookup(ctx, sessionID) + payload, err := s.backend.GetAndDelete(ctx, sessionID) if err != nil { return nil, err } + var session Session + if err := json.Unmarshal(payload, &session); err != nil { + return nil, fmt.Errorf("terminal: decode session %s: %w", sessionID, err) + } if subtle.ConstantTimeCompare([]byte(session.TokenHash), []byte(hashToken(bearerToken))) != 1 { + // Forgery attempt — restore the real session so the + // legitimate client isn't locked out. Compute the remaining + // TTL from ExpiresAt; if already expired we just drop it. + remaining := session.ExpiresAt.Sub(s.now()) + if remaining > 0 { + if restoreErr := s.backend.Set(ctx, sessionID, payload, remaining); restoreErr != nil { + // Log via caller — we don't have a logger here. Returning + // mismatch is the priority; the caller surfaces it as + // Unauthenticated and the audit pipeline flags it. + return nil, ErrTokenMismatch + } + } return nil, ErrTokenMismatch } - return session, nil + return &session, nil } // Revoke removes the session entry, making subsequent Validate / diff --git a/internal/terminal/store_test.go b/internal/terminal/store_test.go index e3047943..fd558656 100644 --- a/internal/terminal/store_test.go +++ b/internal/terminal/store_test.go @@ -117,6 +117,69 @@ func TestTokenStore_Validate_UnknownSession(t *testing.T) { } } +// TestTokenStore_Validate_IsSingleUse covers the rc10 hardening: a +// bearer token can only be successfully validated once. A second +// Validate call with the same (session_id, bearer) must return +// ErrTokenNotFound even within the TTL. This blocks the replay +// surface where a token leaks (e.g. via a reverse-proxy access log +// that captures query params) and an attacker mints additional +// WebSocket connections during the 60 s window. +func TestTokenStore_Validate_IsSingleUse(t *testing.T) { + store := NewTokenStore(NewFakeBackend(nil)) + ctx := context.Background() + + res, err := store.Mint(ctx, MintParams{ + UserID: "user-1", + DeviceID: "device-1", + TtyUser: "pm-tty-alice", + }) + if err != nil { + t.Fatalf("mint: %v", err) + } + + if _, err := store.Validate(ctx, res.SessionID, res.Token); err != nil { + t.Fatalf("first validate should succeed, got %v", err) + } + + _, err = store.Validate(ctx, res.SessionID, res.Token) + if !errors.Is(err, ErrTokenNotFound) { + t.Errorf("second validate must return ErrTokenNotFound (single-use), got %v", err) + } +} + +// TestTokenStore_Validate_MismatchPreservesSession covers the +// companion invariant: a forgery attempt consumes-and-restores the +// entry so the legitimate client can still validate once later. +// Without this behavior, any attacker guess of a session_id would +// lock out the real web client for the remaining TTL. +func TestTokenStore_Validate_MismatchPreservesSession(t *testing.T) { + store := NewTokenStore(NewFakeBackend(nil)) + ctx := context.Background() + + res, err := store.Mint(ctx, MintParams{ + UserID: "user-1", + DeviceID: "device-1", + TtyUser: "pm-tty-alice", + }) + if err != nil { + t.Fatalf("mint: %v", err) + } + + if _, err := store.Validate(ctx, res.SessionID, "wrong-token"); !errors.Is(err, ErrTokenMismatch) { + t.Fatalf("forgery should return ErrTokenMismatch, got %v", err) + } + + // Legitimate client still gets through on the next attempt. + if _, err := store.Validate(ctx, res.SessionID, res.Token); err != nil { + t.Errorf("legitimate validate after forgery should succeed, got %v", err) + } + + // ... and the token is consumed after that one success. + if _, err := store.Validate(ctx, res.SessionID, res.Token); !errors.Is(err, ErrTokenNotFound) { + t.Errorf("re-validation after legitimate use should be ErrTokenNotFound, got %v", err) + } +} + func TestTokenStore_Revoke_IsIdempotent(t *testing.T) { store := NewTokenStore(NewFakeBackend(nil)) ctx := context.Background() diff --git a/internal/terminal/valkey_backend.go b/internal/terminal/valkey_backend.go index 62957b69..2fff7644 100644 --- a/internal/terminal/valkey_backend.go +++ b/internal/terminal/valkey_backend.go @@ -52,3 +52,19 @@ func (b *ValkeyBackend) Delete(ctx context.Context, sessionID string) error { } return nil } + +// GetAndDelete atomically returns the payload and removes the key in a +// single Valkey/Redis round-trip using GETDEL (available since Redis +// 6.2; redis-stack-server ships well past that). This is the primitive +// that makes terminal tokens single-use — two concurrent validators +// cannot both observe the payload. +func (b *ValkeyBackend) GetAndDelete(ctx context.Context, sessionID string) ([]byte, error) { + payload, err := b.client.GetDel(ctx, keyPrefix+sessionID).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrTokenNotFound + } + return nil, fmt.Errorf("terminal: valkey getdel: %w", err) + } + return payload, nil +}