diff --git a/go.mod b/go.mod index c0cd994..2e69f35 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ require ( github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect github.com/go-openapi/swag/jsonname v0.25.5 // indirect + github.com/golang-jwt/jwt/v5 v5.3.1 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect diff --git a/go.sum b/go.sum index fd4fa0e..32a1a17 100644 --- a/go.sum +++ b/go.sum @@ -17,6 +17,8 @@ github.com/go-openapi/swag/jsonname v0.25.5 h1:8p150i44rv/Drip4vWI3kGi9+4W9TdI3U github.com/go-openapi/swag/jsonname v0.25.5/go.mod h1:jNqqikyiAK56uS7n8sLkdaNY/uq6+D2m2LANat09pKU= github.com/go-openapi/testify/v2 v2.4.0 h1:8nsPrHVCWkQ4p8h1EsRVymA2XABB4OT40gcvAu+voFM= github.com/go-openapi/testify/v2 v2.4.0/go.mod h1:HCPmvFFnheKK2BuwSA0TbbdxJ3I16pjwMkYkP4Ywn54= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= diff --git a/internal/config/config.go b/internal/config/config.go index 2e7f0ed..7a82511 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -17,9 +17,11 @@ type Config struct { APIListen string // :8090 APIEnv string // dev, prod - // Auth (interim Phase 2: static bearer tokens; OIDC JWT later) + // Auth (Phase 2: static bearer tokens + OIDC JWT) APIToken string // operator/CI bearer token for the REST API MCPBearerToken string // shared secret for Hermes→API MCP calls + OIDCIssuer string // OIDC issuer URL for JWT validation (e.g. https://authentik.example.com/application/o/oikos/) + OIDCClientID string // OIDC client ID (aud claim expected in JWT) // Observability Debug bool // verbose logging, probe payloads, SQL @@ -55,6 +57,12 @@ func FromEnv() Config { if v := os.Getenv("OIKOS_ENV"); v != "" { c.APIEnv = v } + if v := os.Getenv("OIKOS_OIDC_ISSUER"); v != "" { + c.OIDCIssuer = v + } + if v := os.Getenv("OIKOS_OIDC_CLIENT_ID"); v != "" { + c.OIDCClientID = v + } if v := os.Getenv("OIKOS_API_TOKEN"); v != "" { c.APIToken = v } @@ -86,8 +94,8 @@ func (c Config) String() string { if c.MCPBearerToken != "" { token = "***" } - return fmt.Sprintf("Config{DB=%s Listen=%s Env=%s Debug=%v MCPToken=%s SeedsDir=%s}", - c.redactedDBURL(), c.APIListen, c.APIEnv, c.Debug, token, c.SeedsDir) + return fmt.Sprintf("Config{DB=%s Listen=%s Env=%s Debug=%v MCPToken=%s SeedsDir=%s OIDCIssuer=%s OIDCClientID=%s}", + c.redactedDBURL(), c.APIListen, c.APIEnv, c.Debug, token, c.SeedsDir, c.OIDCIssuer, c.OIDCClientID) } // LogValue implements slog.LogValuer so structured handlers (JSON) never @@ -105,5 +113,7 @@ func (c Config) LogValue() slog.Value { slog.Bool("debug", c.Debug), slog.String("mcp_token", token), slog.String("seeds_dir", c.SeedsDir), + slog.String("oidc_issuer", c.OIDCIssuer), + slog.String("oidc_client_id", c.OIDCClientID), ) } diff --git a/internal/domain/errors.go b/internal/domain/errors.go index 2344396..0d9d76b 100644 --- a/internal/domain/errors.go +++ b/internal/domain/errors.go @@ -18,4 +18,5 @@ var ( ErrAlreadyExists = errors.New("entity already exists") ErrQuarantined = errors.New("pattern is quarantined") ErrSkillDeprecated = errors.New("skill is deprecated") + ErrInvalidInput = errors.New("invalid input") ) diff --git a/internal/httpapi/impl.go b/internal/httpapi/impl.go index 2b2d8d0..6020a84 100644 --- a/internal/httpapi/impl.go +++ b/internal/httpapi/impl.go @@ -2,14 +2,18 @@ package httpapi import ( "context" + "crypto/sha256" "encoding/json" "fmt" "strconv" + "strings" "time" "github.com/dtoro/oikos/internal/db" + "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/domain" "github.com/dtoro/oikos/internal/httpapi/gen" + "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) @@ -746,3 +750,312 @@ func (s *Server) QueryAudit(ctx context.Context, req gen.QueryAuditRequestObject } return gen.QueryAudit200JSONResponse{Items: items}, rows.Err() } + +// ─── Entity mutations ────────────────────────────────────────────── + +func (s *Server) CreateEntity(ctx context.Context, req gen.CreateEntityRequestObject) (gen.CreateEntityResponseObject, error) { + if req.Body == nil { + return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) + } + + // Check idempotency if a key was provided. + actor := "operator" + var bodyHash string + if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" { + key := *req.Params.IdempotencyKey + q := sqlcgen.New(s.pool) + cached, err := q.GetIdempotentResponse(ctx, sqlcgen.GetIdempotentResponseParams{ + Actor: actor, + Key: key, + }) + if err == nil { + // Verify the request body hasn't changed. + bodyJSON, _ := json.Marshal(req.Body) + bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON)) + if cached.RequestHash != bodyHash { + return nil, fmt.Errorf("%w: idempotency key %s used with different request body", domain.ErrConflict, key) + } + // Replay the cached response. + if cached.ResponseCode != nil && *cached.ResponseCode == 201 { + var entity gen.Entity + if len(cached.ResponseBody) > 0 { + if err := json.Unmarshal(cached.ResponseBody, &entity); err != nil { + return nil, fmt.Errorf("unmarshal cached response: %w", err) + } + } + return gen.CreateEntity201JSONResponse{ + Body: entity, + Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, + }, nil + } + // Forward cached error response. + return gen.CreateEntitydefaultApplicationProblemPlusJSONResponse{ + Body: gen.Problem{Status: int(*cached.ResponseCode), Title: "replayed error"}, + StatusCode: int(*cached.ResponseCode), + }, nil + } + } + + id := uuid.New() + slug := req.Body.Slug + if slug == "" { + slug = req.Body.Type + ":" + req.Body.Name + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + q := sqlcgen.New(tx) + + // Validate type exists and is NOT abstract. + var isAbstract bool + if err := tx.QueryRow(ctx, `SELECT is_abstract FROM entity_types WHERE name = $1`, req.Body.Type).Scan(&isAbstract); err != nil { + if err == pgx.ErrNoRows { + return nil, fmt.Errorf("%w: entity type %q", domain.ErrNotFound, req.Body.Type) + } + return nil, err + } + if isAbstract { + return nil, fmt.Errorf("%w: %s", domain.ErrAbstractType, req.Body.Type) + } + + // Get default state from lifecycle. + var defaultState *string + var lcDefault string + if err := tx.QueryRow(ctx, `SELECT ld.default_state FROM lifecycle_defs ld + JOIN entity_types et ON et.lifecycle_id = ld.id + WHERE et.name = $1`, req.Body.Type).Scan(&lcDefault); err == nil { + defaultState = &lcDefault + } + + state := req.Body.State + if state == nil && defaultState != nil { + state = defaultState + } + + var attrsJSON []byte + if req.Body.Attributes != nil { + attrsJSON, _ = json.Marshal(req.Body.Attributes) + } + + // Insert the entity. + inserted, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{ + ID: id, + Slug: slug, + Type: req.Body.Type, + Name: req.Body.Name, + State: state, + Attributes: attrsJSON, + }) + if err != nil { + // Duplicate slug. + if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") { + return nil, fmt.Errorf("%w: slug %q already exists", domain.ErrAlreadyExists, slug) + } + return nil, err + } + + // Convert sqlcgen.Entity → gen.Entity. + entity := sqlcEntityToGen(inserted) + + // Cache idempotent response. + if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" { + respBody, _ := json.Marshal(entity) + code := int32(201) + if bodyHash == "" { + bodyJSON, _ := json.Marshal(req.Body) + bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON)) + } + if putErr := q.PutIdempotentResponse(ctx, sqlcgen.PutIdempotentResponseParams{ + Actor: actor, + Key: *req.Params.IdempotencyKey, + RequestHash: bodyHash, + ResponseCode: &code, + ResponseBody: respBody, + }); putErr != nil { + return nil, putErr + } + } + + // Audit. + entityID := inserted.ID + if auditErr := observability.Audit(ctx, q, "operator", actor, "create", + &entityID, "POST", "/api/v1/entities", "", + map[string]any{"type": req.Body.Type, "slug": slug}); auditErr != nil { + return nil, auditErr + } + + if eventErr := observability.Event(ctx, q, "entity.created", &entityID, + "info", "oikos-api", "", + map[string]any{"slug": slug, "type": req.Body.Type}); eventErr != nil { + return nil, eventErr + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + + return gen.CreateEntity201JSONResponse{ + Body: entity, + Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, + }, nil +} + +func (s *Server) PatchEntity(ctx context.Context, req gen.PatchEntityRequestObject) (gen.PatchEntityResponseObject, error) { + if req.Body == nil { + return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) + } + + id, err := s.resolveEntityID(ctx, req.Id) + if err != nil { + return nil, err + } + + // Parse If-Match header (quoted version string). + ifMatch := strings.Trim(req.Params.IfMatch, `"`) + expectedVersion, err := strconv.Atoi(ifMatch) + if err != nil { + return nil, fmt.Errorf("%w: invalid If-Match header %q", domain.ErrInvalidInput, req.Params.IfMatch) + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + // Get current entity for version check + lifecycle validation. + current, err := sqlcgen.New(tx).GetEntityByID(ctx, id) + if err != nil { + if err == pgx.ErrNoRows { + return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, req.Id) + } + return nil, err + } + + if int(current.Version) != expectedVersion { + return nil, fmt.Errorf("%w: expected version %d, current version %d", + domain.ErrConflict, expectedVersion, current.Version) + } + + // Validate lifecycle transition if state is being changed. + if req.Body.State != nil && *req.Body.State != "" { + // Get lifecycle def for the entity's type. + lc, err := sqlcgen.New(tx).GetLifecycleForType(ctx, current.Type) + if err != nil { + if err == pgx.ErrNoRows { + // No lifecycle defined — any state is allowed. + _ = lc + } else { + return nil, err + } + } else { + // Check the transition is valid. + var transitions map[string][]string + if err := json.Unmarshal(lc.Transitions, &transitions); err != nil { + return nil, fmt.Errorf("parse lifecycle transitions: %w", err) + } + + fromState := "" + if current.State != nil { + fromState = *current.State + } + toState := *req.Body.State + + if allowed, ok := transitions[fromState]; ok { + found := false + for _, s := range allowed { + if s == toState { + found = true + break + } + } + if !found { + return nil, fmt.Errorf("%w: %s → %s", domain.ErrInvalidTransition, fromState, toState) + } + } else if fromState != "" { + // No transitions defined from current state. + return nil, fmt.Errorf("%w: %s → %s", domain.ErrInvalidTransition, fromState, toState) + } + } + } + + // Check idempotency (note: the spec doesn't define Idempotency-Key for PATCH, + // but we handle it if the generated code ever adds it). + // For now, no idempotency check on PATCH. + + // Marshal attributes if provided. + var attrsJSON []byte + if req.Body.Attributes != nil { + attrsJSON, _ = json.Marshal(req.Body.Attributes) + } + + // Perform the update via sqlcgen. + q := sqlcgen.New(tx) + updated, err := q.UpdateEntity(ctx, sqlcgen.UpdateEntityParams{ + Name: req.Body.Name, + State: req.Body.State, + Attributes: attrsJSON, + SetMaintenance: req.Body.MaintenanceUntil != nil, + MaintenanceUntil: req.Body.MaintenanceUntil, + ID: id, + Version: int32(expectedVersion), + }) + if err != nil { + if err == pgx.ErrNoRows { + // Version mismatch or entity not found. + return nil, fmt.Errorf("%w: entity was modified concurrently", domain.ErrConflict) + } + return nil, err + } + + entity := sqlcEntityToGen(updated) + + // Audit. + if auditErr := observability.Audit(ctx, q, "operator", "operator", "patch", + &id, "PATCH", "/api/v1/entities/"+req.Id, "", + map[string]any{"version": expectedVersion}); auditErr != nil { + return nil, auditErr + } + + if eventErr := observability.Event(ctx, q, "entity.updated", &id, + "info", "oikos-api", "", + map[string]any{"slug": entity.Slug, "type": entity.Type, "version": updated.Version}); eventErr != nil { + return nil, eventErr + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + + return gen.PatchEntity200JSONResponse{ + Body: entity, + Headers: gen.PatchEntity200ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, + }, nil +} + +// sqlcEntityToGen converts a sqlcgen.Entity to a gen.Entity. +func sqlcEntityToGen(e sqlcgen.Entity) gen.Entity { + out := gen.Entity{ + Id: e.ID, + Slug: e.Slug, + Type: e.Type, + Name: e.Name, + State: e.State, + Version: int(e.Version), + CreatedAt: e.CreatedAt, + UpdatedAt: e.UpdatedAt, + } + if e.MaintenanceUntil != nil { + out.MaintenanceUntil = e.MaintenanceUntil + } + if len(e.Attributes) > 0 { + var attrs map[string]any + if json.Unmarshal(e.Attributes, &attrs) == nil && len(attrs) > 0 { + out.Attributes = &attrs + } + } + return out +} diff --git a/internal/httpapi/problem.go b/internal/httpapi/problem.go index ffb5f90..5ac14a4 100644 --- a/internal/httpapi/problem.go +++ b/internal/httpapi/problem.go @@ -26,6 +26,8 @@ func statusFor(err error) (status int, title string) { return http.StatusConflict, "relationship cardinality violation" case errors.Is(err, domain.ErrAbstractType), errors.Is(err, domain.ErrInvalidEdge): return http.StatusUnprocessableEntity, "ontology validation failed" + case errors.Is(err, domain.ErrInvalidInput): + return http.StatusBadRequest, "invalid input" case errors.Is(err, domain.ErrApprovalRequired): return http.StatusForbidden, "operator approval required" case errors.Is(err, domain.ErrAutonomyBlocked): diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index d428bbd..687a3ae 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -6,29 +6,62 @@ package httpapi import ( "context" + "crypto/rsa" "crypto/subtle" + "crypto/tls" + "encoding/base64" + "encoding/json" + "fmt" "log/slog" + "math/big" "net/http" "strings" + "sync" "time" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" + "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/httpapi/gen" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" + "github.com/golang-jwt/jwt/v5" ) +// 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 + pool *db.Pool + cfg config.Config + sseBroker *sseBroker + sseSubs map[*sseSubscriber]struct{} + sseMu sync.Mutex } // NewHandler builds the full HTTP handler: /healthz (unauthenticated, // SG18) + the OpenAPI surface under /api/v1 behind bearer auth. func NewHandler(pool *db.Pool, cfg config.Config) http.Handler { - s := &Server{pool: pool, cfg: cfg} + s := &Server{ + pool: pool, + cfg: cfg, + sseBroker: newSSEBroker(10000), + sseSubs: make(map[*sseSubscriber]struct{}), + } + + // Start background SSE listener + go s.sseListener(context.Background()) r := chi.NewRouter() r.Use(middleware.Recoverer) @@ -57,15 +90,334 @@ func NewHandler(pool *db.Pool, cfg config.Config) http.Handler { gen.HandlerWithOptions(strict, gen.ChiServerOptions{ BaseURL: "/api/v1", BaseRouter: r, - Middlewares: []gen.MiddlewareFunc{bearerAuth(cfg)}, + Middlewares: []gen.MiddlewareFunc{combinedAuth(cfg)}, ErrorHandlerFunc: func(w http.ResponseWriter, req *http.Request, err error) { writeProblem(w, req, http.StatusBadRequest, "bad request", err.Error()) }, }) + // Register SSE stream endpoint directly on the chi router with the + // same auth middleware, bypassing the oapi-codegen strict handler + // (which would buffer the entire response body). + r.With(combinedAuth(cfg)).Get("/api/v1/events/stream", s.serveSSE) + 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 +} + // bearerAuth is the interim Phase 2 auth: a static bearer token // (OIKOS_API_TOKEN / OIKOS_MCP_BEARER_TOKEN from Infisical in prod). In // dev mode with no token configured, requests pass as the operator. @@ -103,6 +455,9 @@ func bearerAuth(cfg config.Config) func(http.Handler) http.Handler { } } +// Ensure sqlcgen is imported (used in sse.go but referenced here for build safety). +var _ = &sqlcgen.Queries{} + // requestLogger logs one line per request with method, path, status, // duration, and the chi request id. func requestLogger(next http.Handler) http.Handler { @@ -144,4 +499,4 @@ func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config) error defer cancel() return srv.Shutdown(shutdownCtx) } -} +} \ No newline at end of file diff --git a/internal/httpapi/sse.go b/internal/httpapi/sse.go new file mode 100644 index 0000000..282a7d5 --- /dev/null +++ b/internal/httpapi/sse.go @@ -0,0 +1,409 @@ +package httpapi + +import ( + "container/list" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "strconv" + "sync" + "time" + + "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/httpapi/gen" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// SSE broker is a keep-last-event-id in-memory buffer used for subscriber +// fan-out. The database NOTIFY is the primary delivery mechanism; this buffer +// just supports the Last-Event-ID replay on connect. +type sseBroker struct { + mu sync.Mutex + buf *list.List // list of sqlcgen.Event + cache map[int64]*list.Element // id → list element for O(1) lookup + cap int + lastID int64 +} + +func newSSEBroker(capacity int) *sseBroker { + return &sseBroker{ + buf: list.New(), + cache: make(map[int64]*list.Element), + cap: capacity, + } +} + +func (b *sseBroker) push(ev sqlcgen.Event) { + b.mu.Lock() + defer b.mu.Unlock() + + // Evict oldest if at capacity + for b.buf.Len() >= b.cap && b.buf.Len() > 0 { + front := b.buf.Front() + b.cache[front.Value.(sqlcgen.Event).ID] = nil // don't delete, just nil + b.buf.Remove(front) + } + + elem := b.buf.PushBack(ev) + b.cache[ev.ID] = elem + if ev.ID > b.lastID { + b.lastID = ev.ID + } +} + +func (b *sseBroker) after(id int64) []sqlcgen.Event { + b.mu.Lock() + defer b.mu.Unlock() + + if id >= b.lastID { + return nil + } + + // Walk from the front to find the first element after id + var events []sqlcgen.Event + for e := b.buf.Front(); e != nil; e = e.Next() { + ev := e.Value.(sqlcgen.Event) + if ev.ID > id { + events = append(events, ev) + } + } + return events +} + +func (b *sseBroker) latestID() int64 { + b.mu.Lock() + defer b.mu.Unlock() + return b.lastID +} + +// notifyPayload is the JSON payload from the pg_notify trigger (migration 008). +type notifyPayload struct { + ID int64 `json:"id"` + Ts string `json:"ts"` + Type string `json:"type"` + EntityID *string `json:"entity_id"` + Severity string `json:"severity"` + Source string `json:"source"` + CorrelationID *string `json:"correlation_id"` +} + +// sseSubscriber holds the channels and cancel func for one SSE client. +type sseSubscriber struct { + ch chan sqlcgen.Event + done chan struct{} + cancel context.CancelFunc +} + +// serveSSE is the streaming handler for GET /api/v1/events/stream. +// +// Protocol: https://html.spec.whatwg.org/multipage/server-sent-events.html +// +// 1. If Last-Event-ID is present, replay buffered events from the in-memory +// broker (or fall back to ListEventsAfter for cold start). +// 2. Subscribe via in-memory channel and forward events from pg_notify. +// 3. Send a colon-comment heartbeat every 15 s. +// 4. Unsubscribe and clean up on client disconnect. +func (s *Server) serveSSE(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + writeProblem(w, r, http.StatusInternalServerError, "internal error", "streaming not supported") + return + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") // disable nginx buffering + w.WriteHeader(http.StatusOK) + flusher.Flush() + + ctx := r.Context() + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + sub := &sseSubscriber{ + ch: make(chan sqlcgen.Event, 64), + done: make(chan struct{}), + cancel: cancel, + } + + s.sseMu.Lock() + s.sseSubs[sub] = struct{}{} + s.sseMu.Unlock() + + defer func() { + s.sseMu.Lock() + delete(s.sseSubs, sub) + s.sseMu.Unlock() + close(sub.done) + }() + + // ── 1. Replay ───────────────────────────────────────────────── + if lastID := r.Header.Get("Last-Event-ID"); lastID != "" { + id, err := strconv.ParseInt(lastID, 10, 64) + if err == nil { + replayed := 0 + + // Try in-memory broker first + events := s.sseBroker.after(id) + if len(events) > 0 { + for _, ev := range events { + if !writeSSE(w, flusher, ev) { + return + } + replayed++ + } + } + + // If broker didn't have them all, fetch from DB + if replayed == 0 || events[len(events)-1].ID != s.sseBroker.latestID() { + q := sqlcgen.New(s.pool) + dbEvents, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ + ID: id, + Limit: 5000, + }) + if err != nil { + slog.Error("sse db replay failed", "error", err) + } else { + for _, ev := range dbEvents { + if !writeSSE(w, flusher, ev) { + return + } + } + } + } + } + } + + // ── 2. Subscribe and forward ────────────────────────────────── + heartbeat := time.NewTicker(15 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-ctx.Done(): + return + + case ev, ok := <-sub.ch: + if !ok { + return + } + if !writeSSE(w, flusher, ev) { + return + } + + case <-heartbeat.C: + _, err := fmt.Fprintf(w, ": heartbeat\n\n") + if err != nil { + return + } + flusher.Flush() + } + } +} + +// sseListener runs in a background goroutine: it opens a dedicated pgx +// connection, LISTENs on oikos_events, and fans out each notification to +// all live subscribers. Runs until ctx is cancelled. +func (s *Server) sseListener(ctx context.Context) { + poolConn, err := s.pool.Acquire(ctx) + if err != nil { + slog.Error("sse listener acquire failed", "error", err) + return + } + defer poolConn.Release() + + conn := poolConn.Conn() + if _, err := conn.Exec(ctx, "LISTEN oikos_events"); err != nil { + slog.Error("sse listener listen failed", "error", err) + return + } + + slog.Info("sse listener started on oikos_events") + defer slog.Info("sse listener stopped") + + for { + nt, err := conn.WaitForNotification(ctx) + if err != nil { + if ctx.Err() != nil { + return // normal shutdown + } + slog.Error("sse listener notification error", "error", err) + // Reconnect on error after a brief delay + select { + case <-time.After(5 * time.Second): + case <-ctx.Done(): + return + } + // Re-acquire connection + poolConn.Release() + var reconnErr error + poolConn, reconnErr = s.pool.Acquire(ctx) + if reconnErr != nil { + slog.Error("sse listener reconnect failed", "error", reconnErr) + return + } + conn = poolConn.Conn() + if _, err := conn.Exec(ctx, "LISTEN oikos_events"); err != nil { + slog.Error("sse listener re-listen failed", "error", err) + return + } + continue + } + + var p notifyPayload + if err := json.Unmarshal([]byte(nt.Payload), &p); err != nil { + slog.Error("sse listener unmarshal failed", "error", err) + continue + } + + // Fetch full event from DB + q := sqlcgen.New(s.pool) + events, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ + ID: p.ID - 1, + Limit: 1, + }) + if err != nil || len(events) == 0 { + slog.Warn("sse listener event fetch failed", "id", p.ID, "error", err) + continue + } + ev := events[0] + + // Push to broker + s.sseBroker.push(ev) + + // Fan out to subscribers (non-blocking send) + s.sseMu.Lock() + for sub := range s.sseSubs { + select { + case sub.ch <- ev: + default: + // Subscriber too slow — drop event for them + // (they'll reconnect via Last-Event-ID) + } + } + s.sseMu.Unlock() + } +} + +// writeSSE writes a single Event as an SSE message. Returns false if the +// write failed (client disconnected). +func writeSSE(w ioWriter, flusher http.Flusher, ev sqlcgen.Event) bool { + data, err := json.Marshal(ev) + if err != nil { + return true // skip un-serializable events + } + _, err = fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", ev.ID, ev.Type, data) + if err != nil { + return false + } + flusher.Flush() + return true +} + +// ioWriter is an interface satisfied by both http.ResponseWriter and +// io.StringWriter, letting writeSSE work with the writer directly. +type ioWriter interface { + Write([]byte) (int, error) +} + +// StreamEvents implements the OpenAPI interface (SSE event stream). +// Uses io.Pipe to bridge the streaming SSE goroutine to the response body. +func (s *Server) StreamEvents(ctx context.Context, req gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) { + pr, pw := io.Pipe() + go func() { + // Wrap the pipe writer as an http.ResponseWriter-like struct + // that implements http.Flusher via calling flush on the pipe + // (which isn't a real flusher — we use the io.Pipe writer directly + // via writeSSE's ioWriter interface). + s.serveSSEWriter(pw, req.Params) + pw.Close() + }() + return gen.StreamEvents200TexteventStreamResponse{ + Body: pr, + ContentLength: -1, // unknown length + }, nil +} + +// serveSSEWriter runs the SSE loop writing to an io.Writer. +func (s *Server) serveSSEWriter(w io.Writer, params gen.StreamEventsParams) { + // No explicit flusher for pipe writes — io.Pipe flushes on each Write. + ctx := context.Background() + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + sub := &sseSubscriber{ + ch: make(chan sqlcgen.Event, 64), + done: make(chan struct{}), + cancel: cancel, + } + s.sseMu.Lock() + s.sseSubs[sub] = struct{}{} + s.sseMu.Unlock() + defer func() { + s.sseMu.Lock() + delete(s.sseSubs, sub) + s.sseMu.Unlock() + close(sub.done) + }() + + // Replay on Last-Event-ID + if params.LastEventID != nil && *params.LastEventID != "" { + id, err := strconv.ParseInt(*params.LastEventID, 10, 64) + if err == nil { + events := s.sseBroker.after(id) + if len(events) > 0 { + for _, ev := range events { + if !writeSSE(w, nil, ev) { + return + } + } + } + // If broker didn't have them all, fetch from DB + if len(events) == 0 || events[len(events)-1].ID != s.sseBroker.latestID() { + q := sqlcgen.New(s.pool) + dbEvents, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ID: id, Limit: 5000}) + if err != nil { + slog.Error("sse db replay failed", "error", err) + } else { + for _, ev := range dbEvents { + if !writeSSE(w, nil, ev) { + return + } + } + } + } + } + } + + // Subscribe and forward + heartbeat := time.NewTicker(15 * time.Second) + defer heartbeat.Stop() + + for { + select { + case <-ctx.Done(): + return + case ev, ok := <-sub.ch: + if !ok { + return + } + if !writeSSE(w, nil, ev) { + return + } + case <-heartbeat.C: + _, err := fmt.Fprintf(w, ": heartbeat\n\n") + if err != nil { + return + } + } + } +} + +// Ensure pgxpool is imported — used via Acquire. +var _ = &pgxpool.Pool{} +var _ = pgx.ErrNoRows diff --git a/internal/httpapi/stubs.go b/internal/httpapi/stubs.go index 8d4a83a..c38f198 100644 --- a/internal/httpapi/stubs.go +++ b/internal/httpapi/stubs.go @@ -38,18 +38,6 @@ func (s *Server) ListClassifications(ctx context.Context, request gen.ListClassi return nil, errNotImplemented } -func (s *Server) CreateEntity(ctx context.Context, request gen.CreateEntityRequestObject) (gen.CreateEntityResponseObject, error) { - return nil, errNotImplemented -} - -func (s *Server) PatchEntity(ctx context.Context, request gen.PatchEntityRequestObject) (gen.PatchEntityResponseObject, error) { - return nil, errNotImplemented -} - -func (s *Server) StreamEvents(ctx context.Context, request gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) { - return nil, errNotImplemented -} - func (s *Server) ListExecutions(ctx context.Context, request gen.ListExecutionsRequestObject) (gen.ListExecutionsResponseObject, error) { return nil, errNotImplemented }