Janady13/missionmatch / cmd/fairness-snapshot/main.go

1// Command fairness-snapshot appends the continuous adverse impact four fifths time series. For each 2// organization (or a single passed org), it computes the current adverse impact report and APPENDS one 3// eeo.fairness_snapshots row per demographic group bucket, stamping the AEDT model version in force. 4// Run on a schedule (a cron, the same way mm-opp-sync runs), it builds the time series that proves on 5// every release the platform is not drifting into disparate impact. It is append only and safe to rerun. 6// 7// Isolation: this job runs as the COMPLIANCE role, the ONLY role granted the eeo schema. The application 8// role and the AI path get nothing. When no -org is passed, the orgs to snapshot are enumerated from the 9// distinct organization_id values in audit_logs, which is outside RLS and granted to the compliance role, 10// so the job needs no cross org read of the tenant data. Each snapshot then runs inside a tenant 11// transaction, so RLS scopes the report join and the snapshot insert to that one org. 12// 13// Output is GROUP LEVEL AGGREGATE only: orgs snapshotted, group rows written, and any below threshold 14// count. No individual PII is read, computed, or printed. 15// 16// Env (server side only): MM_COMPLIANCE_DATABASE_URL (the compliance DSN, NOBYPASSRLS, required). 17// Flags: -org (optional, a single organization uuid; default is every org with AEDT activity). 18// 19// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run with GODEBUG=fips140=on. 20package main 21 22import ( 23 "context" 24 "crypto/fips140" 25 "crypto/pbkdf2" 26 "crypto/sha256" 27 "errors" 28 "flag" 29 "log" 30 "os" 31 "strings" 32 "time" 33 34 "github.com/google/uuid" 35 "github.com/jackc/pgx/v5/pgxpool" 36 37 "github.com/missionworkforce/missionmatch/internal/modules/eeo" 38 "github.com/missionworkforce/missionmatch/internal/platform/database" 39 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 40) 41 42func main() { 43 orgFlag := flag.String("org", "", "snapshot a single organization uuid (default: every org with AEDT activity)") 44 flag.Parse() 45 46 // FIPS self test: derive one PBKDF2 key under the active crypto module and fail closed if it errors, 47 // so a build that dropped the validated module is caught at boot, the same in build FIPS gate the api 48 // and import-bench binaries use. 49 if err := fipsSelfTest(); err != nil { 50 log.Fatalf("fairness-snapshot: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err) 51 } 52 log.Printf("fairness-snapshot: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version()) 53 54 dsn := strings.TrimSpace(os.Getenv("MM_COMPLIANCE_DATABASE_URL")) 55 if dsn == "" { 56 log.Fatal("fairness-snapshot: set MM_COMPLIANCE_DATABASE_URL (the compliance role DSN, the only path to the eeo schema)") 57 } 58 59 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) 60 defer cancel() 61 62 pool, err := database.NewPool(ctx, dsn) 63 if err != nil { 64 log.Fatalf("fairness-snapshot: connect: %v", err) 65 } 66 defer pool.Close() 67 68 // Fail closed unless this is the NOBYPASSRLS compliance role. A BYPASSRLS DSN here would let the 69 // report join and the snapshot insert escape RLS and mix demographics across orgs. 70 if err := database.AssertRole(ctx, pool, database.ComplianceRoleName()); err != nil { 71 log.Fatalf("fairness-snapshot: compliance pool role check: %v", err) 72 } 73 74 svc := eeo.NewService(tenant.New(pool)) 75 76 orgs, err := resolveOrgs(ctx, pool, *orgFlag) 77 if err != nil { 78 log.Fatalf("fairness-snapshot: resolve orgs: %v", err) 79 } 80 // THE ASSERTION THAT CLOSES THE LIE, finding D1b. Producing zero organizations while organizations 81 // demonstrably exist is a LOUD FAILURE, never a quiet exit 0. The old behaviour printed "no 82 // organizations to snapshot" and exited successfully every night for the life of the deployment, so 83 // systemd recorded success, nothing alerted, and the AEDT compliance artifact was simply never 84 // produced. An empty result is only acceptable when the platform genuinely holds no organizations. 85 if len(orgs) == 0 { 86 var exists bool 87 // The existence check must be INDEPENDENT of the enumeration. Asking the same function twice, which 88 // the first version of this fix did, produces an assertion that can never fire. 89 if e := pool.QueryRow(ctx, `SELECT eeo_any_organization_exists()`).Scan(&exists); e != nil { 90 log.Fatalf("fairness-snapshot: resolved zero organizations AND could not confirm whether any "+ 91 "exist: %v. Refusing to report success on an unknown state.", e) 92 } 93 if exists { 94 log.Fatal("fairness-snapshot: resolved ZERO organizations while the directory says organizations " + 95 "exist. This is the D1b failure mode and it must never exit 0: the adverse impact snapshot " + 96 "is a compliance artifact, and silently producing nothing is worse than crashing.") 97 } 98 log.Print("fairness-snapshot: no organizations exist to snapshot, which is a real empty state") 99 return 100 } 101 102 now := time.Now().UTC() 103 var snapshotted, totalGroups, totalBelow int 104 for _, org := range orgs { 105 sum, serr := svc.SnapshotAdverseImpact(tenant.WithOrg(ctx, org.String()), now) 106 if serr != nil { 107 // One org failing must not abort the schedule for the rest. Log the org id (not PII) and move on. 108 log.Printf("fairness-snapshot: org %s snapshot failed: %v", org, serr) 109 continue 110 } 111 snapshotted++ 112 totalGroups += sum.GroupsWritten 113 totalBelow += sum.BelowThreshold 114 log.Printf("fairness-snapshot: org %s snapshotted %d groups, %d below threshold, model %s", 115 org, sum.GroupsWritten, sum.BelowThreshold, sum.ModelVersion) 116 } 117 log.Printf("fairness-snapshot: done, %d orgs snapshotted, %d group rows written, %d below the 0.80 threshold", 118 snapshotted, totalGroups, totalBelow) 119} 120 121// resolveOrgs returns the orgs to snapshot. A passed -org is used verbatim (after validation). Otherwise 122// the orgs are the distinct non null organization_id values in audit_logs, which is OUTSIDE RLS and 123// granted to the compliance role, so the job enumerates the orgs with AEDT activity without a cross org 124// read of any tenant scoped table. 125func resolveOrgs(ctx context.Context, pool *pgxpool.Pool, orgFlag string) ([]uuid.UUID, error) { 126 if s := strings.TrimSpace(orgFlag); s != "" { 127 id, err := uuid.Parse(s) 128 if err != nil { 129 return nil, errors.New("fairness-snapshot: -org is not a valid uuid: " + s) 130 } 131 return []uuid.UUID{id}, nil 132 } 133 // ENUMERATED THROUGH THE DIRECTORY, via a narrow SECURITY DEFINER function that returns organization 134 // ids and nothing else. Reading audit_logs here was the D1b defect: this job holds the NOBYPASSRLS 135 // compliance role by design, and asking a role that can see nothing without an org context to discover 136 // which contexts exist can only ever return nothing. 137 rows, err := pool.Query(ctx, `SELECT organization_id FROM eeo_snapshot_organizations()`) 138 if err != nil { 139 return nil, err 140 } 141 defer rows.Close() 142 var out []uuid.UUID 143 for rows.Next() { 144 var id uuid.UUID 145 if e := rows.Scan(&id); e != nil { 146 return nil, e 147 } 148 out = append(out, id) 149 } 150 return out, rows.Err() 151} 152 153// fipsSelfTest derives one PBKDF2 HMAC SHA 256 key at a representative cost under the active crypto 154// module. It fails closed if the module is absent or errors, so a non FIPS build is caught at boot. 155func fipsSelfTest() error { 156 _, err := pbkdf2.Key(sha256.New, "fairness-snapshot-self-test", []byte("fairness-snapshot-salt"), 600000, 32) 157 return err 158} 159