Janady13/missionmatch / cmd/api/main.go

1// Command api is the missionMATCH HTTP server entrypoint. 2// 3// It wires configuration, the FIPS startup guard and self test, the Postgres pool (NOBYPASSRLS app 4// role), the embedding provider, the authentication module, and the tenant scoped API. Cloud and 5// provider selection are configuration (the portability mandate); no cloud SDK is imported here. 6package main 7 8import ( 9 "context" 10 "crypto/fips140" 11 "crypto/pbkdf2" 12 "crypto/rand" 13 "crypto/sha256" 14 "encoding/json" 15 "errors" 16 "log" 17 "net/http" 18 "os/signal" 19 "syscall" 20 "time" 21 22 "github.com/go-chi/chi/v5" 23 "github.com/go-chi/chi/v5/middleware" 24 25 "github.com/missionworkforce/missionmatch/internal/modules/ai" 26 "github.com/missionworkforce/missionmatch/internal/modules/auth" 27 "github.com/missionworkforce/missionmatch/internal/modules/awards" 28 "github.com/missionworkforce/missionmatch/internal/modules/commandcenter" 29 "github.com/missionworkforce/missionmatch/internal/modules/eeo" 30 "github.com/missionworkforce/missionmatch/internal/modules/governance" 31 "github.com/missionworkforce/missionmatch/internal/modules/matching" 32 "github.com/missionworkforce/missionmatch/internal/modules/missionmatch" 33 "github.com/missionworkforce/missionmatch/internal/modules/opportunities" 34 oppintel "github.com/missionworkforce/missionmatch/internal/modules/opportunity_intelligence" 35 "github.com/missionworkforce/missionmatch/internal/modules/screenmatch" 36 "github.com/missionworkforce/missionmatch/internal/modules/slatematch" 37 "github.com/missionworkforce/missionmatch/internal/modules/sourcing" 38 "github.com/missionworkforce/missionmatch/internal/modules/staffing" 39 "github.com/missionworkforce/missionmatch/internal/modules/wages" 40 "github.com/missionworkforce/missionmatch/internal/platform/audit" 41 "github.com/missionworkforce/missionmatch/internal/platform/config" 42 "github.com/missionworkforce/missionmatch/internal/platform/database" 43 "github.com/missionworkforce/missionmatch/internal/platform/extraction" 44 "github.com/missionworkforce/missionmatch/internal/platform/inference" 45 "github.com/missionworkforce/missionmatch/internal/platform/obs" 46 "github.com/missionworkforce/missionmatch/internal/platform/ratelimit" 47 "github.com/missionworkforce/missionmatch/internal/platform/security" 48 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 49) 50 51func main() { 52 cfg, err := config.Load() 53 if err != nil { 54 log.Fatalf("config: %v", err) 55 } 56 57 // Structured JSON logging for the request path (debug level in dev, info in production): the obs 58 // middleware logs every request through slog. The one time boot and shutdown messages below stay on 59 // the standard logger, since they are operator facing startup output, not the high volume hot path. 60 obs.Setup(!cfg.IsProduction()) 61 62 // FIPS startup guard: in a FIPS deployment, refuse to start with crypto outside the validated module. 63 if cfg.RequireFIPS && !fips140.Enabled() { 64 log.Fatalf("FIPS required but OFF, build with GOFIPS140=v1.0.0 and run with GODEBUG=fips140=on") 65 } 66 // FIPS self test: actually derive a PBKDF2 key at the configured cost under the active module and 67 // fail closed if it errors, so a broken crypto path is caught at boot, not at first login. 68 if err := fipsSelfTest(); err != nil { 69 log.Fatalf("FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err) 70 } 71 log.Printf("missionMATCH starting: env=%s cloud=%s fips_enabled=%v module=%s", 72 cfg.Env, cfg.Cloud, fips140.Enabled(), fips140.Version()) 73 74 ctx := context.Background() 75 var db *tenant.DB 76 if cfg.AppDatabaseURL != "" { 77 pool, err := database.NewPool(ctx, cfg.AppDatabaseURL) 78 if err != nil { 79 log.Fatalf("db pool: %v", err) 80 } 81 // Fail closed unless this pool is the NOBYPASSRLS app role, so a swapped or over privileged DSN 82 // can never silently bypass RLS. 83 if err := database.AssertRole(ctx, pool, "missionmatch_app"); err != nil { 84 log.Fatalf("app pool role check: %v", err) 85 } 86 db = tenant.New(pool) 87 defer db.Close() 88 } else { 89 log.Printf("WARNING: MM_DATABASE_URL unset, DB backed routes return 503") 90 } 91 92 // The compliance pool authenticates as the compliance role and is the ONLY pool with eeo access. 93 // It drives adverse impact reporting; the app pool above can never read demographics. 94 var complianceDB *tenant.DB 95 if cfg.ComplianceDatabaseURL != "" { 96 pool, err := database.NewPool(ctx, cfg.ComplianceDatabaseURL) 97 if err != nil { 98 log.Fatalf("compliance db pool: %v", err) 99 } 100 // Fail closed unless this is the NOBYPASSRLS compliance role. A BYPASSRLS DSN here would leak 101 // every tenant's demographics cross org through the adverse impact join. 102 if err := database.AssertRole(ctx, pool, "missionmatch_compliance"); err != nil { 103 log.Fatalf("compliance pool role check: %v", err) 104 } 105 complianceDB = tenant.New(pool) 106 defer complianceDB.Close() 107 } else { 108 log.Printf("WARNING: MM_COMPLIANCE_DATABASE_URL unset, adverse impact reporting returns 503") 109 } 110 111 ext, err := extraction.New(cfg.Extraction) 112 if err != nil { 113 log.Fatalf("extraction: %v", err) 114 } 115 116 auditW := audit.New(db) 117 authSvc, err := buildAuthService(cfg, db, auditW) 118 if err != nil { 119 log.Fatalf("auth: %v", err) 120 } 121 csrf := security.NewCSRF(authKey(cfg.Auth.CSRFKey, 32, "MM_CSRF_KEY", cfg.IsProduction())) 122 authH := auth.NewHandler(authSvc, csrf, cfg.Auth.CookieSecure, cfg.AllowedOrigins, cfg.TrustedProxyHeader) 123 124 // The skill matcher reads the BACKFILLED skill embeddings (the HNSW indexed skills.embedding column) 125 // rather than embedding skill name strings at request time, so the vector layer is live in the scoring 126 // path and the fake default embedder is gone from it. 127 matcher := matching.NewStoredSkillMatcher(db) 128 svc := matching.NewService(db, matcher, auditW).WithExtractor(ext) 129 h := matching.NewHandler(svc) 130 screenH := screenmatch.NewHandler(h) 131 slateH := slatematch.NewHandler(h) 132 ccH := commandcenter.NewHandler(commandcenter.NewServiceWithAudit(db, auditW)) 133 missionH := missionmatch.NewHandler(h, ccH) 134 oppH := opportunities.NewHandler(opportunities.NewService(db)) 135 oppIntelH := oppintel.NewHandler(oppintel.NewService(db)) 136 awardsH := awards.NewHandler(awards.NewService(db)) 137 wagesH := wages.NewHandler(wages.NewService(db)) 138 sourcingH := sourcing.NewHandler(sourcing.NewService(db, auditW)) 139 eeoH := eeo.NewHandler(eeo.NewService(complianceDB)) 140 govH := governance.NewHandler(governance.NewService(db, auditW, cfg.DecisionRetentionDays)) 141 142 // W3 staffing recommendations. It reuses the ONE engine path (the same skill matcher) and narrates only 143 // through the optional inference plane; a nil client means deterministic rationales, so the feature 144 // works even when the plane is down. Generating a run is a write capability; the deterministic engine 145 // stays the floor and a person decides every recommendation. 146 var inferClient *inference.Client 147 if cfg.Inference.URL != "" { 148 ic, ierr := inference.New(inference.Config{ 149 BaseURL: cfg.Inference.URL, EmbedModel: cfg.Inference.EmbedModel, RerankModel: cfg.Inference.RerankModel, 150 GenerateModel: cfg.Inference.GenerateModel, Dimension: cfg.Inference.Dimension, 151 Timeout: time.Duration(cfg.Inference.TimeoutSeconds) * time.Second, MaxRetries: cfg.Inference.MaxRetries, 152 BreakerTrips: cfg.Inference.BreakerTrips, BreakerCool: time.Duration(cfg.Inference.BreakerCoolSec) * time.Second, 153 TeachingEnabled: cfg.Inference.TeachingEnabled, 154 }) 155 if ierr != nil { 156 log.Fatalf("inference client: %v", ierr) 157 } 158 inferClient = ic 159 } else { 160 log.Printf("WARNING: MM_MEGAMIND_URL unset, staffing recommendation rationales are deterministic only") 161 } 162 staffingH := staffing.NewHandler(staffing.NewService(db, matcher, auditW, inferClient, staffing.Load())) 163 164 // The AI brain, mounted LIVE on the shared RLS database (per tenant pools are a sequenced follow on, 165 // out of scope here). It reuses the same inference client and audit writer and is memory augmented; a 166 // nil inference client makes every answer deterministic (the graceful no plane posture). The 167 // deterministic engine remains the floor: the brain narrates, it never scores. 168 brainH := ai.NewHandler(ai.NewOrchestrator(db, inferClient, auditW).WithMemory(ai.NewMemory(db, inferClient))) 169 // Fail the boot if any curated TAG intent is unsafe (a write, an unallowlisted table, or eeo), so an 170 // unsafe intent can never ship live, as the registry guarantee promises. 171 if err := ai.ValidateRegistry(); err != nil { 172 log.Fatalf("ai brain: unsafe TAG intent in the registry: %v", err) 173 } 174 // The enforced re embed gate. With a configured plane, /ai/query stays dark until a boot self check 175 // confirms the stored vectors were produced by the configured embed model (the all-minilm to bge 176 // corruption mode mis ranks every retrieval with no error). A proven mismatch keeps the brain route 177 // disabled with a typed 503 while the rest of the API serves normally; a boot time plane outage lets 178 // the gate self heal on the first query after the plane recovers. This is never a fatal boot error. 179 if inferClient != nil { 180 gate := ai.NewVectorGate(func(ctx context.Context) error { 181 return ai.VerifyVectorModel(ctx, db, inferClient, cfg.Inference.Dimension) 182 }) 183 brainH = brainH.WithVectorGate(gate) 184 bootCtx, cancelBoot := context.WithTimeout(context.Background(), 30*time.Second) 185 switch err := gate.Check(bootCtx); { 186 case err == nil: 187 log.Printf("brain vector model check passed: stored vectors match %s", cfg.Inference.EmbedModel) 188 case errors.Is(err, ai.ErrVectorModelMismatch): 189 log.Printf("ERROR: brain vector model check FAILED, /ai/query is disabled until a re embed: %v", err) 190 default: 191 log.Printf("WARNING: brain vector model not yet verified at boot, /ai/query will retry on first use: %v", err) 192 } 193 cancelBoot() 194 } 195 196 // Resolve the tenant middleware. In a release build this is always the session resolver; the dev 197 // X-Org-ID path only exists in a dev tagged build. Assert that positively in production. 198 sessionMW, resolverKind := chooseResolver(cfg, authH) 199 if cfg.IsProduction() && resolverKind != "session" { 200 log.Fatalf("production must resolve the tenant from a validated session, got resolver %q", resolverKind) 201 } 202 log.Printf("tenant resolver: %s", resolverKind) 203 if !cfg.Auth.MFARequired { 204 log.Printf("WARNING: MM_MFA_REQUIRED is false, login is password only with no TOTP step (pilot only, this weakens the auth posture)") 205 } 206 207 csrfMW := csrf.Protect(authH.CSRFCookieName(), auth.CSRFHeader, cfg.AllowedOrigins) 208 209 // Metrics registry, fed the database pool gauges so /metrics reports connection saturation. 210 metrics := obs.NewRegistry() 211 metrics.SetPoolStats(func() map[string]float64 { 212 out := map[string]float64{} 213 if db != nil && db.Pool != nil { 214 s := db.Pool.Stat() 215 out["app_acquired"] = float64(s.AcquiredConns()) 216 out["app_idle"] = float64(s.IdleConns()) 217 out["app_total"] = float64(s.TotalConns()) 218 out["app_max"] = float64(s.MaxConns()) 219 } 220 if complianceDB != nil && complianceDB.Pool != nil { 221 out["compliance_total"] = float64(complianceDB.Pool.Stat().TotalConns()) 222 } 223 return out 224 }) 225 226 // Rate limiters: a generous per IP cap on the whole API, and a strict one on the public auth routes 227 // (login, CSRF, recovery) to blunt credential stuffing. Account lockout in the auth service is the 228 // deeper control; this is the transport layer defense in depth. Sweepers bound memory. 229 stopSweep := make(chan struct{}) 230 defer close(stopSweep) 231 apiLimiter := ratelimit.New(50, 100) 232 apiLimiter.StartSweeper(stopSweep, time.Minute, 10*time.Minute) 233 authLimiter := ratelimit.New(1, 10) 234 authLimiter.StartSweeper(stopSweep, time.Minute, 10*time.Minute) 235 236 r := chi.NewRouter() 237 r.Use(middleware.RequestID) 238 // NOTE: middleware.RealIP is deliberately NOT used; it trusts spoofable client headers (deprecated 239 // in chi for that reason). The auth handler and the rate limiter derive the client IP from the TCP 240 // peer, or from MM_TRUSTED_PROXY_HEADER only when the app is behind a trusted reverse proxy. 241 r.Use(middleware.Recoverer) 242 r.Use(metrics.Middleware) // access log + metrics, outside the timeout so timed out requests still record 243 r.Use(middleware.Timeout(60 * time.Second)) 244 245 // Liveness: the process is up. Cheap, never touches the database (a probe must not be coupled to the DB). 246 r.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) { 247 w.WriteHeader(http.StatusOK) 248 _, _ = w.Write([]byte("ok")) 249 }) 250 // Readiness: the process can serve, the database pools answer. A failing pool returns 503 so a load 251 // balancer stops routing to a node that cannot reach Postgres (the old /healthz reported healthy even 252 // with a dead DB). 253 r.Get("/readyz", func(w http.ResponseWriter, req *http.Request) { 254 ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second) 255 defer cancel() 256 if db != nil && db.Pool != nil { 257 if err := db.Pool.Ping(ctx); err != nil { 258 http.Error(w, "app database not ready", http.StatusServiceUnavailable) 259 return 260 } 261 } 262 if complianceDB != nil && complianceDB.Pool != nil { 263 if err := complianceDB.Pool.Ping(ctx); err != nil { 264 http.Error(w, "compliance database not ready", http.StatusServiceUnavailable) 265 return 266 } 267 } 268 w.WriteHeader(http.StatusOK) 269 _, _ = w.Write([]byte("ready")) 270 }) 271 r.Get("/healthz/fips", func(w http.ResponseWriter, _ *http.Request) { 272 w.Header().Set("Content-Type", "application/json") 273 _ = json.NewEncoder(w).Encode(map[string]any{ 274 "fips_enabled": fips140.Enabled(), 275 "fips_enforced": fips140.Enforced(), 276 "module": fips140.Version(), 277 "pbkdf2_self_test": true, // a failure would have aborted startup 278 }) 279 }) 280 // Metrics are unauthenticated operational data carrying no PII; production firewalls this to the 281 // scraping network. It sits outside /api so the API rate limiter never throttles a scrape. 282 r.Handle("/metrics", metrics.Handler()) 283 284 r.Route("/api", func(r chi.Router) { 285 r.Use(apiLimiter.Middleware(cfg.TrustedProxyHeader)) 286 // Public pre tenant auth: runs before any org is known. Behind the strict auth rate limiter, the 287 // CSRF Origin allowlist, and the signed double submit, but NOT the session resolver (these operate 288 // on a pre MFA session or no session and read the cookie themselves). 289 r.Group(func(r chi.Router) { 290 r.Use(authLimiter.Middleware(cfg.TrustedProxyHeader)) 291 r.Use(csrfMW) 292 authH.PublicRoutes(r) 293 }) 294 // Tenant scoped: the session resolver feeds app.current_org_id, then CSRF, then per route RBAC. 295 r.Group(func(r chi.Router) { 296 r.Use(sessionMW) 297 r.Use(csrfMW) 298 // Flat alias surface: the pre split paths (POST /match, /slates/..., /candidates, ...) stay 299 // working so existing clients and the screenmatch.thatwebhostingguy.com redirect never break. 300 h.Routes(r) 301 // Canonical tool namespaces over the same shared core: screenMATCH, slateMATCH, missionMATCH. 302 r.Route("/screenmatch", screenH.Routes) 303 r.Route("/slatematch", slateH.Routes) 304 r.Route("/missionmatch", missionH.Routes) 305 ccH.Routes(r) 306 oppH.Routes(r) 307 oppIntelH.Routes(r) 308 awardsH.Routes(r) 309 wagesH.Routes(r) 310 sourcingH.Routes(r) 311 eeoH.Routes(r) 312 govH.Routes(r) 313 staffingH.Routes(r) 314 brainH.Routes(r) 315 authH.TenantRoutes(r) 316 }) 317 }) 318 319 srv := &http.Server{ 320 Addr: cfg.HTTPAddr, 321 Handler: r, 322 ReadHeaderTimeout: 10 * time.Second, 323 ReadTimeout: 30 * time.Second, 324 WriteTimeout: 60 * time.Second, 325 IdleTimeout: 120 * time.Second, 326 } 327 328 // Graceful shutdown: on SIGINT or SIGTERM, stop accepting and let in flight requests (and their 329 // tenant transactions) drain before exit, so a redeploy never severs a write mid commit. 330 shutdownCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) 331 defer stop() 332 go func() { 333 log.Printf("listening on %s", cfg.HTTPAddr) 334 if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { 335 log.Fatalf("server: %v", err) 336 } 337 }() 338 <-shutdownCtx.Done() 339 log.Printf("shutdown signal received, draining in flight requests") 340 drainCtx, cancel := context.WithTimeout(context.Background(), 25*time.Second) 341 defer cancel() 342 if err := srv.Shutdown(drainCtx); err != nil { 343 log.Printf("graceful shutdown error: %v", err) 344 } 345 log.Printf("stopped") 346} 347 348// fipsSelfTest derives one PBKDF2 key at the production iteration count under the active crypto module. 349func fipsSelfTest() error { 350 _, err := pbkdf2.Key(sha256.New, "missionmatch fips self test", []byte("self test salt"), security.PBKDF2Iterations, 32) 351 return err 352} 353 354// buildAuthService assembles the auth service from config. In production every key is present 355// (config.Validate enforces it). In development a missing key falls back to an ephemeral random key 356// with a loud warning, so the server still starts for local work. 357func buildAuthService(cfg config.Config, db *tenant.DB, auditW *audit.Writer) (*auth.Service, error) { 358 prod := cfg.IsProduction() 359 pepperKeys := map[int][]byte{ 360 cfg.Auth.PepperVersion: authKey(cfg.Auth.Pepper, 32, "MM_CREDENTIAL_PEPPER", prod), 361 } 362 if len(cfg.Auth.PepperPrevious) > 0 { 363 pepperKeys[cfg.Auth.PepperVersion-1] = cfg.Auth.PepperPrevious 364 } 365 pepper, err := security.NewPepper(cfg.Auth.PepperVersion, pepperKeys) 366 if err != nil { 367 return nil, err 368 } 369 hasher, err := security.NewHasher(pepper, security.PBKDF2Iterations) 370 if err != nil { 371 return nil, err 372 } 373 tagger, err := security.NewTagger(authKey(cfg.Auth.RecoveryLookupKey, 32, "MM_RECOVERY_LOOKUP_KEY", prod)) 374 if err != nil { 375 return nil, err 376 } 377 // The KEK map carries the active version plus an optional prior version so a rotated KEK still 378 // decrypts TOTP secrets sealed under the old one, mirroring the pepper rotation above. 379 totpKEKs := map[int][]byte{ 380 cfg.Auth.TOTPKEKVersion: authKey(cfg.Auth.TOTPKEK, 32, "MM_TOTP_KEK", prod), 381 } 382 if len(cfg.Auth.TOTPKEKPrevious) > 0 { 383 totpKEKs[cfg.Auth.TOTPKEKVersion-1] = cfg.Auth.TOTPKEKPrevious 384 } 385 return auth.NewService(auth.Deps{ 386 DB: db, 387 Audit: auditW, 388 Hasher: hasher, 389 Tagger: tagger, 390 TOTP: security.TOTP{}, 391 TOTPKEKs: totpKEKs, 392 TOTPKEKActive: cfg.Auth.TOTPKEKVersion, 393 TOTPIssuer: cfg.Auth.TOTPIssuer, 394 IdleSeconds: cfg.Auth.SessionIdleSeconds, 395 AbsoluteSeconds: cfg.Auth.SessionAbsoluteSeconds, 396 PendingSeconds: cfg.Auth.PendingAbsoluteSeconds, 397 InviteSeconds: cfg.Auth.InviteExpirySeconds, 398 CookieSecure: cfg.Auth.CookieSecure, 399 MFADisabled: !cfg.Auth.MFARequired, 400 }) 401} 402 403// authKey returns the configured key, or an ephemeral random key in development when it is unset. 404func authKey(set []byte, size int, name string, prod bool) []byte { 405 if len(set) > 0 { 406 return set 407 } 408 if prod { 409 // config.Validate already fatals before here; this is a defensive backstop. 410 log.Fatalf("%s required in production", name) 411 } 412 b := make([]byte, size) 413 _, _ = rand.Read(b) 414 log.Printf("WARNING: %s unset, using an ephemeral dev key (sessions and credentials will not survive a restart; set %s for a stable dev database)", name, name) 415 return b 416} 417