// Package httpapi implements the Oikos REST API. The contract is // api/openapi.yaml (contract-first, ADR-0004); handlers implement the // oapi-codegen strict-server interface in gen/. Errors map to RFC 9457 // problem+json via domain sentinels. package httpapi import ( "context" "crypto/rsa" "crypto/subtle" "crypto/tls" "encoding/base64" "encoding/json" "fmt" "log/slog" "math/big" "net/http" "net/http/httputil" "net/url" "os" "strings" "sync" "time" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/httpapi/gen" mcphandler "github.com/dtoro/oikos/internal/mcp" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" "github.com/golang-jwt/jwt/v5" "github.com/google/uuid" ) // contextKey for storing actor identity in request context. type contextKey string const actorKey contextKey = "oikos_actor" // actor holds the resolved identity of the API caller. type actor struct { Type string // "operator", "agent", "system" Label string // human-readable label ID string // OIDC sub or static token identifier TokenType string // "static" or "oidc" } // Server implements gen.StrictServerInterface over the DB layer. type Server struct { pool *db.Pool cfg config.Config secretsManager secretsBackend sseBroker *sseBroker sseSubs map[*sseSubscriber]struct{} sseMu sync.Mutex } // secretsBackend is a minimal interface for secrets operations used by the // HTTP API (enrollment key storage, listing). Compatible with internal/secrets. type secretsBackend interface { Set(ctx context.Context, key string, value string) error List(ctx context.Context) ([]string, error) } // NewHandler builds the full HTTP handler: /healthz (unauthenticated, // SG18) + the OpenAPI surface under /api/v1 behind bearer auth. // // ctx governs the lifetime of the background SSE listener goroutine, which // holds a dedicated pooled connection for LISTEN. Callers MUST cancel ctx // before closing the pool — otherwise the held connection never releases // and pool.Close() deadlocks. func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, uiHandler http.Handler) http.Handler { s := &Server{ pool: pool, cfg: cfg, sseBroker: newSSEBroker(10000), sseSubs: make(map[*sseSubscriber]struct{}), } // Start background SSE listener, tied to ctx for clean shutdown. go s.sseListener(ctx) r := chi.NewRouter() r.Use(middleware.Recoverer) r.Use(middleware.RequestID) r.Use(requestLogger) // Liveness — no auth, no audit (plan SG18). Not exposed via Caddy. r.Get("/healthz", func(w http.ResponseWriter, req *http.Request) { ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second) defer cancel() if err := pool.Ping(ctx); err != nil { writeProblem(w, req, http.StatusServiceUnavailable, "database unreachable", "") return } w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"status":"ok"}`)) }) // Client enrollment — unauthenticated (IP-gated in handler). // Registered on base router BEFORE HandlerWithOptions so it // bypasses the combinedAuth middleware applied to all /api/v1/*. r.Post("/api/v1/clients/enroll", func(w http.ResponseWriter, req *http.Request) { var body gen.EnrollRequest if err := json.NewDecoder(req.Body).Decode(&body); err != nil { writeProblem(w, req, http.StatusBadRequest, "invalid request body", err.Error()) return } resp, err := s.EnrollClient(req.Context(), gen.EnrollClientRequestObject{Body: &body}) if err != nil { writeProblemFromErr(w, req, err) return } if err := resp.VisitEnrollClientResponse(w); err != nil { writeProblem(w, req, http.StatusInternalServerError, "response encoding failed", err.Error()) } }) strict := gen.NewStrictHandlerWithOptions(s, nil, gen.StrictHTTPServerOptions{ RequestErrorHandlerFunc: func(w http.ResponseWriter, req *http.Request, err error) { writeProblem(w, req, http.StatusBadRequest, "bad request", err.Error()) }, ResponseErrorHandlerFunc: writeProblemFromErr, }) gen.HandlerWithOptions(strict, gen.ChiServerOptions{ BaseURL: "/api/v1", BaseRouter: r, Middlewares: []gen.MiddlewareFunc{combinedAuth(cfg)}, ErrorHandlerFunc: func(w http.ResponseWriter, req *http.Request, err error) { writeProblem(w, req, http.StatusBadRequest, "bad request", err.Error()) }, }) // SSE stream: override the generated /events/stream route with a raw // flushing handler (registered AFTER HandlerWithOptions so chi's last // registration wins). The strict-server path can't Flush() per event; // this one uses the real ResponseWriter for real-time delivery. It // inherits the router's base middleware and applies auth via With(). r.With(combinedAuth(cfg)).Get("/api/v1/events/stream", s.serveSSE) // Mount MCP at /mcp (plan R3-10) nomosAgentID := uuid.Nil if cfg.NomosAgentID != "" { if id, err := uuid.Parse(cfg.NomosAgentID); err == nil { nomosAgentID = id } } if nomosAgentID == uuid.Nil && cfg.NomosAgentSlug != "" { _ = pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", cfg.NomosAgentSlug).Scan(&nomosAgentID) } r.With(combinedAuth(cfg)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken, nomosAgentID)) r.Get("/ui/*", func(w http.ResponseWriter, req *http.Request) { if uiHandler != nil { uiHandler.ServeHTTP(w, req) } }) r.Get("/ui", func(w http.ResponseWriter, req *http.Request) { http.Redirect(w, req, "/ui/", http.StatusMovedPermanently) }) r.Get("/", func(w http.ResponseWriter, req *http.Request) { http.Redirect(w, req, "/ui/", http.StatusMovedPermanently) }) if nomosURL := os.Getenv("NOMOS_PROXY_URL"); nomosURL != "" { target, _ := url.Parse(nomosURL) proxy := httputil.NewSingleHostReverseProxy(target) r.Mount("/agent", http.StripPrefix("/agent", proxy)) } return r } // combinedAuth tries OIDC JWT validation first (if configured), falls back to // static bearer token validation, and opens the gate in dev mode when no // credentials are configured. func combinedAuth(cfg config.Config) func(http.Handler) http.Handler { hasOIDC := cfg.OIDCIssuer != "" && cfg.OIDCClientID != "" hasStatic := cfg.APIToken != "" || cfg.MCPBearerToken != "" // Cache JWKS for OIDC var jwksURL string var jwksCache []jwtVerificationKey var jwksMu sync.RWMutex if hasOIDC { // Fetch JWKS URI from OIDC discovery jwksURL = discoverJWKSURI(cfg.OIDCIssuer) if jwksURL != "" { keys, err := fetchJWKS(jwksURL) if err != nil { slog.Warn("oidc initial jwks fetch failed, will retry on demand", "error", err) } else { jwksCache = keys } } } var staticTokens [][]byte if cfg.APIToken != "" { staticTokens = append(staticTokens, []byte(cfg.APIToken)) } if cfg.MCPBearerToken != "" { staticTokens = append(staticTokens, []byte(cfg.MCPBearerToken)) } devOpen := cfg.APIEnv == "dev" && !hasStatic && !hasOIDC return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if devOpen { ctx := context.WithValue(r.Context(), actorKey, actor{ Type: "system", Label: "dev:anonymous", ID: "dev", TokenType: "none", }) next.ServeHTTP(w, r.WithContext(ctx)) return } auth := r.Header.Get("Authorization") raw, ok := strings.CutPrefix(auth, "Bearer ") if !ok || raw == "" { writeProblem(w, r, http.StatusUnauthorized, "unauthorized", "missing bearer token") return } // Try OIDC first if configured if hasOIDC { jwksMu.RLock() keys := jwksCache jwksMu.RUnlock() // If cache is empty, try to refresh if len(keys) == 0 && jwksURL != "" { if freshKeys, err := fetchJWKS(jwksURL); err == nil { jwksMu.Lock() jwksCache = freshKeys keys = freshKeys jwksMu.Unlock() } } if act, err := validateOIDCToken(raw, cfg, keys); err == nil { ctx := context.WithValue(r.Context(), actorKey, act) next.ServeHTTP(w, r.WithContext(ctx)) return } else { slog.Debug("oidc validation failed", "error", err) } } // Fall back to static tokens if hasStatic { for _, t := range staticTokens { if subtle.ConstantTimeCompare([]byte(raw), t) == 1 { label := "operator:api" if cfg.MCPBearerToken != "" && subtle.ConstantTimeCompare([]byte(raw), []byte(cfg.MCPBearerToken)) == 1 { label = "agent:mcp" } ctx := context.WithValue(r.Context(), actorKey, actor{ Type: label[:strings.IndexByte(label, ':')], Label: label, ID: raw[:8] + "...", TokenType: "static", }) next.ServeHTTP(w, r.WithContext(ctx)) return } } } writeProblem(w, r, http.StatusUnauthorized, "unauthorized", "invalid or expired bearer token") }) } } // jwtVerificationKey holds a parsed RSA public key or HMAC secret for JWT // verification, identified by its key ID (kid). type jwtVerificationKey struct { Kid string Alg string Key any // *rsa.PublicKey or []byte for HMAC IsHMAC bool } // discoverJWKSURI fetches the OIDC discovery document and extracts the // jwks_uri field. func discoverJWKSURI(issuerURL string) string { discURL := strings.TrimRight(issuerURL, "/") + "/.well-known/openid-configuration" client := &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: false}, }} resp, err := client.Get(discURL) if err != nil { slog.Warn("oidc discovery failed", "url", discURL, "error", err) return "" } defer resp.Body.Close() var disc struct { JWKSURI string `json:"jwks_uri"` } if err := json.NewDecoder(resp.Body).Decode(&disc); err != nil { slog.Warn("oidc discovery decode failed", "error", err) return "" } if disc.JWKSURI == "" { slog.Warn("oidc discovery missing jwks_uri") return "" } return disc.JWKSURI } // fetchJWKS retrieves the JWK set from a URL and returns the parsed keys. func fetchJWKS(jwksURL string) ([]jwtVerificationKey, error) { client := &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: false}, }} resp, err := client.Get(jwksURL) if err != nil { return nil, fmt.Errorf("fetch jwks: %w", err) } defer resp.Body.Close() var jwks struct { Keys []json.RawMessage `json:"keys"` } if err := json.NewDecoder(resp.Body).Decode(&jwks); err != nil { return nil, fmt.Errorf("decode jwks: %w", err) } var keys []jwtVerificationKey for _, raw := range jwks.Keys { var header struct { Kty string `json:"kty"` Kid string `json:"kid,omitempty"` Alg string `json:"alg,omitempty"` Use string `json:"use,omitempty"` } if err := json.Unmarshal(raw, &header); err != nil { continue } // Skip keys not intended for signature verification if header.Use != "" && header.Use != "sig" { continue } var key jwtVerificationKey key.Kid = header.Kid key.Alg = header.Alg switch header.Kty { case "RSA": var rsaKey struct { N string `json:"n"` E string `json:"e"` } if err := json.Unmarshal(raw, &rsaKey); err != nil { continue } pubKey, err := parseRSAPublicKey(rsaKey.N, rsaKey.E) if err != nil { slog.Debug("oidc parse rsa key failed", "kid", header.Kid, "error", err) continue } key.Key = pubKey case "oct": // HMAC keys not expected for OIDC but handle gracefully key.IsHMAC = true default: continue } if key.Key != nil || key.IsHMAC { keys = append(keys, key) } } if len(keys) == 0 { return nil, fmt.Errorf("no usable keys in jwks") } return keys, nil } // parseRSAPublicKey decodes a base64url-encoded RSA modulus and exponent into // an *rsa.PublicKey. func parseRSAPublicKey(nB64, eB64 string) (any, error) { // Decode base64url modulus nBytes, err := base64.RawURLEncoding.DecodeString(nB64) if err != nil { return nil, fmt.Errorf("decode modulus: %w", err) } // Decode base64url exponent eBytes, err := base64.RawURLEncoding.DecodeString(eB64) if err != nil { return nil, fmt.Errorf("decode exponent: %w", err) } // Build RSA public key n := new(big.Int).SetBytes(nBytes) e := 0 for _, b := range eBytes { e = (e << 8) | int(b) } return &rsa.PublicKey{N: n, E: e}, nil } // validateOIDCToken parses and validates a JWT Bearer token against the OIDC // configuration. Returns the resolved actor on success. func validateOIDCToken(rawToken string, cfg config.Config, keys []jwtVerificationKey) (actor, error) { keyFunc := func(token *jwt.Token) (any, error) { kid, ok := token.Header["kid"].(string) if !ok { return nil, fmt.Errorf("no kid in token header") } // Find matching key for _, k := range keys { // If kid is empty in JWK, try algorithm match if k.Kid == kid || (k.Kid == "" && k.Alg == token.Header["alg"]) { return k.Key, nil } } // Fall back to any RSA key if no kid match (some providers don't set kid) if !ok { for _, k := range keys { if k.Key != nil { return k.Key, nil } } } return nil, fmt.Errorf("no matching key for kid: %s", kid) } token, err := jwt.Parse(rawToken, keyFunc, jwt.WithIssuer(cfg.OIDCIssuer), jwt.WithAudience(cfg.OIDCClientID), jwt.WithValidMethods([]string{"RS256", "RS384", "RS512", "ES256", "ES384", "ES512"}), ) if err != nil { return actor{}, fmt.Errorf("jwt validation: %w", err) } claims, ok := token.Claims.(jwt.MapClaims) if !ok { return actor{}, fmt.Errorf("invalid claims") } sub, _ := claims.GetSubject() if sub == "" { // Try the Azure/Entra ID oid claim fallback if oid, ok := claims["oid"].(string); ok { sub = oid } } preferredUsername, _ := claims["preferred_username"].(string) email, _ := claims["email"].(string) label := sub if preferredUsername != "" { label = preferredUsername } else if email != "" { label = email } return actor{ Type: "operator", Label: label, ID: sub, TokenType: "oidc", }, nil } // GetActor extracts the actor identity from the context. Returns nil if not // set (should not happen for authenticated routes). func GetActor(ctx context.Context) *actor { a, ok := ctx.Value(actorKey).(actor) if !ok { return nil } return &a } // requestLogger logs one line per request with method, path, status, // duration, and the chi request id. func requestLogger(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor) next.ServeHTTP(ww, r) slog.Info("http", "method", r.Method, "path", r.URL.Path, "status", ww.Status(), "duration_ms", time.Since(start).Milliseconds(), "request_id", middleware.GetReqID(r.Context()), ) }) } // ListenAndServe runs the API server with graceful shutdown on ctx cancel // (SG4): stop accepting, drain in-flight for up to 30s, then exit. func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, uiHandler http.Handler) error { srv := &http.Server{ Addr: cfg.APIListen, Handler: NewHandler(ctx, pool, cfg, uiHandler), ReadHeaderTimeout: 10 * time.Second, } errCh := make(chan error, 1) go func() { slog.Info("api listening", "addr", cfg.APIListen) errCh <- srv.ListenAndServe() }() select { case err := <-errCh: return err case <-ctx.Done(): slog.Info("api shutting down") shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() return srv.Shutdown(shutdownCtx) } }