"""HTTP surface for the security auditor domain. All routes mount under ``/api/v1/auditor``. RBAC: * ``can_view_security`` — read endpoints * ``can_manage_security`` — write / sync / decide / build endpoints """ from __future__ import annotations import os import uuid from collections.abc import Sequence from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import FileResponse from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.exceptions import ForbiddenError, NotFoundError from app.core.rbac import can_manage_security, can_view_security from app.core.response import success_response from app.domains.auditor import findings_service from app.domains.auditor import schemas as audsch from app.domains.auditor.constants import ALL_MFA_PROVIDERS from app.domains.auditor.evidence_query.router import ( register_evidence_query_routes, ) from app.domains.auditor.integration_adapters import ( AtlassianAuditorAdapter, IntegrationMissingError, ) from app.domains.auditor.models import ( AccessCertificationItem, AuditorCertExpiry, AuditorEvidencePack, AuditorMfaStatus, AuditorPatAudit, AuditorPublicExposureFinding, AuditorRiskScore, AuditorSecretRotationRecord, AuditorSigninEvent, ) from app.domains.auditor.services import ( CertManualService, DashboardService, EvidencePackService, ExposureService, RiskScoreService, SigninAcknowledgeService, ) from app.domains.auditor.sync_services import ( CertExpirySyncService, MfaSyncService, PatAuditService, PublicExposureScanService, SecretRotationSyncService, SigninAnomalyService, ) from app.identity.person_models import IdentityPerson from app.scheduled_jobs.models import ScheduledJob router = APIRouter( prefix="/api/v1/auditor", tags=["auditor"], ) # ───────────────────────────────────────────────────────────────────── # Auth helpers # ───────────────────────────────────────────────────────────────────── def _require_view(user: IdentityPerson) -> None: if not can_view_security(user.role_names): raise ForbiddenError( "auditor or global_admin role required to view security data" ) def _require_forensic_read(user: IdentityPerson) -> None: """Read a frozen evidence bundle — narrower than ``_require_view``. ``can_view_security`` admits ``management`` as a read-only leadership tier, which is right for aggregate posture (risk score, control counts, cert expiry). It is NOT right here: a bundle is one named person's complete access footprint — every seat, permission, cloud grant, approval chain and audit action, with display names in the clear — and any holder could enumerate any colleague's, executives included. The MCP surface already drew this same line for support transcripts a few days ago, and this keeps the two consistent instead of leaving one of them the soft way in. Producing a bundle is already restricted to this tier; reading one back must not be broader than creating it. """ if not can_manage_security(user.role_names): raise ForbiddenError( "auditor or global_admin role required to read frozen " "evidence bundles" ) def _require_manage(user: IdentityPerson) -> None: if not can_manage_security(user.role_names): raise ForbiddenError( "auditor or global_admin role required to mutate security data" ) def _trace_id(request: Request) -> uuid.UUID: raw = getattr(request.state, "trace_id", None) if isinstance(raw, uuid.UUID): return raw if raw: try: return uuid.UUID(str(raw)) except ValueError: pass return uuid.uuid4() async def _persons_index( db: AsyncSession, person_ids: list[uuid.UUID], ) -> dict[uuid.UUID, IdentityPerson]: if not person_ids: return {} rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(person_ids), ) )).scalars().all() return {p.id: p for p in rows} # ───────────────────────────────────────────────────────────────────── # Dashboard # ───────────────────────────────────────────────────────────────────── @router.get("/coverage") async def get_coverage( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Per-control integration-coverage status. For each of the 8 auditor controls, lists which upstream integrations the control depends on, whether each is configured (an enabled ``integrations`` row exists for the right kind / provider) and whether data has actually landed in the auditor table for it (``last_synced_at`` newer than 7 days). The frontend uses this to render an honest empty-state on every page: ┌─────────────────────────────────┐ │ MFA compliance — 2/4 connected │ │ ✓ google │ │ ✓ github │ │ ✗ feishu → 配置 integration │ │ ✗ atlassian → 配置 integration │ └─────────────────────────────────┘ Never lies: a control with zero connected integrations is shown as "no data source configured" with a deep-link to /admin/integrations, not "暂无数据". """ _require_view(user) from app.integrations.models import Integration integration_rows = (await db.execute( select( Integration.kind, Integration.provider, Integration.enabled, Integration.last_test_status, Integration.last_tested_at, ) )).all() have: dict[tuple[str, str], dict[str, Any]] = {} for k, p, enabled, status, tested_at in integration_rows: if not enabled: continue # Multi-instance kinds (cloud) keep the freshest test result so # the dashboard reflects the most recently-validated row. prev = have.get((k, p)) if prev is None or ( tested_at is not None and (prev.get("last_tested_at") is None or tested_at > prev["last_tested_at"]) ): have[(k, p)] = { "last_test_status": status, "last_tested_at": tested_at, } # Atlassian's Org Admin API key is held inside the Jira integration's # secrets blob; presence of the row is NOT enough — without that # specific field the MFA / Public-exposure / PAT controls' # atlassian-flavored adapters all fail with IntegrationMissingError. # Probe once here so the dashboard shows the honest state. atlassian_org_admin_configured = False try: await AtlassianAuditorAdapter._resolve_org_admin_creds(db) atlassian_org_admin_configured = True except IntegrationMissingError: atlassian_org_admin_configured = False except Exception: # noqa: BLE001 — resolver bug ≠ integration gap # Don't lie either way; degrade to the row-presence check below. atlassian_org_admin_configured = ( ("software_provisioning", "jira") in have ) def _entry( key: str, label: str, *, kind: str | None, provider: str | None, configured: bool, ) -> dict[str, Any]: meta = have.get((kind, provider)) if (kind and provider) else None out: dict[str, Any] = { "key": key, "label": label, "configured": configured, "kind": kind, "provider": provider, } if meta: out["last_test_status"] = meta.get("last_test_status") tested_at = meta.get("last_tested_at") out["last_tested_at"] = ( tested_at.isoformat() if tested_at else None ) return out seven_days_ago = datetime.now(UTC) - timedelta(days=7) # Pre-launch hardening 2026-05-04: explicit per-model timestamp # column registry instead of getattr-with-default. The previous # ``getattr(model, "last_synced_at", None) or getattr(model, # "last_seen_at", None)`` would silently return False on a future # column rename — every Security Auditor "data-present" indicator # would flip to "no recent data" with zero error, hiding a stealth # regression of every control. Same family of bugs the user has # called out four PRs in a row (memory: feedback_no_getattr_default # _for_model_columns). _data_present_timestamp_col: dict[type, Any] = { AuditorMfaStatus: AuditorMfaStatus.last_synced_at, AuditorPatAudit: AuditorPatAudit.last_synced_at, AuditorCertExpiry: AuditorCertExpiry.last_synced_at, AuditorSecretRotationRecord: ( AuditorSecretRotationRecord.last_synced_at ), AuditorPublicExposureFinding: ( AuditorPublicExposureFinding.last_seen_at ), # AuditorSigninEvent stamps each ingested row at ``event_at`` (the # provider's audit-log event time, not a sync-marker column). # Missing this entry was the 2026-05-05 prod regression: the # registry was added in pre-launch hardening but Signin was the # only one of six controls left out, so /coverage 500'd whenever # the route reached signin's data_present check (always — it # runs unconditionally). AuditorSigninEvent: AuditorSigninEvent.event_at, } async def _has_recent(model, *, where=None) -> bool: col = _data_present_timestamp_col.get(model) if col is None: raise RuntimeError( f"_has_recent: model {model.__name__} not registered in " "_data_present_timestamp_col — add the timestamp column " "explicitly." ) q = select(func.count()).select_from(model) q = q.where(col >= seven_days_ago) if where is not None: q = q.where(where) return int((await db.execute(q)).scalar_one()) > 0 # Per-control dependency declaration. Format: # integrations: list of (kind, provider, label) the control reads # data_present_check: callable returning bool (recent rows exist) controls: dict[str, dict[str, Any]] = {} # 1. MFA — Google + Feishu + GitHub + Atlassian (Jira integration # carries the org_admin_api_key needed for Atlassian Guard # 2-step-verification policy probe) mfa_per = [ _entry( "google", "Google Workspace", kind="software_provisioning", provider="google_workspace", configured=("software_provisioning", "google_workspace") in have, ), _entry( "feishu", "Feishu", kind="chat_approval", provider="feishu", configured=("chat_approval", "feishu") in have, ), _entry( "github", "GitHub Enterprise", kind="software_provisioning", provider="github", configured=("software_provisioning", "github") in have, ), _entry( "atlassian", "Atlassian (via Jira integration's org_admin_api_key)", kind="software_provisioning", provider="jira", # Tighter than just-row-exists — the MFA call requires # org_admin_api_key inside the secrets blob. configured=atlassian_org_admin_configured, ), ] controls["mfa"] = { "providers": mfa_per, "data_present": await _has_recent(AuditorMfaStatus), } # 2. Privileged review — same set as MFA + AWS for root accounts priv_per = list(mfa_per) + [ _entry( "aws", "AWS (root + IAM admin holders)", kind="cloud", provider="aws", configured=("cloud", "aws") in have, ), ] controls["privileged_review"] = { "providers": priv_per, "data_present": int( (await db.execute( select(func.count()) .select_from(AccessCertificationItem) .where(AccessCertificationItem.access_type == "privileged_role") )).scalar_one() ) > 0, } # 3. Public exposure — AWS (SG + S3) + Atlassian (anonymous spaces) controls["public_exposure"] = { "providers": [ _entry( "aws", "AWS (Security Groups + S3)", kind="cloud", provider="aws", configured=("cloud", "aws") in have, ), _entry( "atlassian", "Atlassian (Confluence anon + Jira public)", kind="software_provisioning", provider="jira", # Public-spaces also rides org_admin_api_key. configured=atlassian_org_admin_configured, ), ], "data_present": await _has_recent( AuditorPublicExposureFinding, ), } # 4. Secret rotation — Helix native + AWS IAM/ACM + Aliyun RAM + # Tencent CAM + Azure storage keys + Azure SP secrets controls["secret_rotation"] = { "providers": [ {"key": "helix", "label": "Helix API keys (native)", "configured": True, "kind": None, "provider": None}, _entry( "aws", "AWS IAM access keys + ACM certs", kind="cloud", provider="aws", configured=("cloud", "aws") in have, ), _entry( "aliyun", "Aliyun RAM access keys", kind="cloud", provider="aliyun", configured=("cloud", "aliyun") in have, ), _entry( "tencent", "Tencent CAM access keys", kind="cloud", provider="tencent", configured=("cloud", "tencent") in have, ), _entry( "azure", "Azure storage keys + SP client secrets", kind="cloud", provider="azure", configured=("cloud", "azure") in have, ), ], "data_present": await _has_recent(AuditorSecretRotationRecord), } # 5. Signin anomaly — Google Workspace + Feishu audit controls["signin_anomaly"] = { "providers": [ _entry( "google", "Google Workspace Reports API", kind="software_provisioning", provider="google_workspace", configured=("software_provisioning", "google_workspace") in have, ), _entry( "feishu", "Feishu audit log API", kind="chat_approval", provider="feishu", configured=("chat_approval", "feishu") in have, ), ], # The registered timestamp col (``event_at``) already constrains # to within seven_days_ago, so no extra ``where=`` is needed. "data_present": await _has_recent(AuditorSigninEvent), } # 6. PAT audit — Helix native + GitHub + OAuth + Atlassian org tokens controls["pat_audit"] = { "providers": [ {"key": "helix_pat", "label": "Helix PAT (native)", "configured": True, "kind": None, "provider": None}, {"key": "helix_apikey", "label": "Helix API keys (native)", "configured": True, "kind": None, "provider": None}, _entry( "github", "GitHub PATs", kind="software_provisioning", provider="github", configured=("software_provisioning", "github") in have, ), _entry( "atlassian", "Atlassian org API tokens", kind="software_provisioning", provider="jira", # Org tokens require the same org_admin_api_key. configured=atlassian_org_admin_configured, ), ], "data_present": await _has_recent(AuditorPatAudit), } # 7. Cert + ICP expiry — AWS ACM + manual ledger + Aliyun & Tencent # registered domains (the cloud domain APIs surface ICP filing # number for .cn / .com.cn domains) controls["cert_expiry"] = { "providers": [ {"key": "manual", "label": "Manual ledger (Helix-internal)", "configured": True, "kind": None, "provider": None}, _entry( "aws_acm", "AWS ACM", kind="cloud", provider="aws", configured=("cloud", "aws") in have, ), _entry( "aliyun_domain", "Aliyun domains (+ ICP filings)", kind="cloud", provider="aliyun", configured=("cloud", "aliyun") in have, ), _entry( "tencent_domain", "Tencent domains", kind="cloud", provider="tencent", configured=("cloud", "tencent") in have, ), ], "data_present": await _has_recent(AuditorCertExpiry), } # 8. Risk score — derived; data_present iff any RiskScore row exists risk_rows = int((await db.execute( select(func.count()).select_from(AuditorRiskScore) )).scalar_one()) controls["risk_score"] = { "providers": [ {"key": "derived", "label": "Computed from MFA + signin + secrets + PATs + certs", "configured": True, "kind": None, "provider": None}, ], "data_present": risk_rows > 0, } return success_response( data={"controls": controls}, request=request, ) # ───────────────────────────────────────────────────────────────────── # Control freshness (scheduled-job state) + unified manual trigger # ───────────────────────────────────────────────────────────────────── # Each auditor control is fed by exactly one scheduled-job row. The # mapping lives here (not in the DB) because the relationship is part of # the API contract — the frontend reads it via /control-status to # render "上次运行 X 分钟前 / 上次错误 / 下次运行" on every card. CONTROL_HANDLER_KEYS: dict[str, str] = { "mfa": "auditor.mfa_sync_daily", "privileged_review": "auditor.privileged_review_auto_open_quarterly", "public_exposure": "auditor.public_exposure_scan_daily", "secret_rotation": "auditor.secret_rotation_check_daily", "signin_anomaly": "auditor.signin_anomaly_score_hourly", "pat_audit": "auditor.pat_audit_daily", "cert_expiry": "auditor.cert_expiry_check_daily", "risk_score": "auditor.risk_score_recompute_hourly", } def _truncate(s: str | None, n: int = 500) -> str | None: if not s: return None return s if len(s) <= n else s[:n] + "…" @router.get("/control-status") async def get_control_status( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Per-control freshness & last-error status. Returns the matching ``scheduled_jobs`` row's last-run state for each of the 8 controls. The frontend uses this to render an inline "上次运行 / 上次错误 + 立刻同步" strip on each control card so users never have to guess whether a 0/0 dashboard means "actually clean" or "the cron never fired". Deliberately no next-run field: ``scheduled_jobs`` has no stored next-run column (dropped in migration 620 — nothing ever wrote it). A next-run time comes from ``scheduled_jobs.service.schedule_forecast``, which computes it from the live trigger. """ _require_view(user) handler_keys = list(CONTROL_HANDLER_KEYS.values()) rows = (await db.execute( select(ScheduledJob).where( ScheduledJob.handler_key.in_(handler_keys) ) )).scalars().all() by_handler: dict[str, ScheduledJob] = {r.handler_key: r for r in rows} out: dict[str, dict[str, Any]] = {} for control_key, handler_key in CONTROL_HANDLER_KEYS.items(): row = by_handler.get(handler_key) if row is None: out[control_key] = { "handler_key": handler_key, "job_id": None, "display_name": None, "enabled": False, "last_run_at": None, "last_status": None, "last_duration_ms": None, "last_error": None, "last_result_summary": None, "missing_handler": True, } continue out[control_key] = { "handler_key": row.handler_key, "job_id": str(row.id), "display_name": row.display_name, "enabled": bool(row.enabled), "last_run_at": ( row.last_run_at.isoformat() if row.last_run_at else None ), "last_status": row.last_status, "last_duration_ms": row.last_duration_ms, "last_error": _truncate(row.last_error, 800), "last_result_summary": row.last_result_summary, "missing_handler": False, } return success_response( data={ "controls": out, "as_of": datetime.now(UTC).isoformat(), }, request=request, ) @router.post("/control/{control_key}/sync") async def trigger_control_sync( control_key: str, request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Synchronously fire the scheduled job that feeds ``control_key``. Reuses :func:`app.scheduled_jobs.service.execute_job` so the run is recorded in ``scheduled_job_runs`` and ``scheduled_jobs.last_run_at`` refreshes in the same way the worker would do it. The HTTP call returns when the run finishes — the frontend uses this for the "立刻同步" button so the user sees fresh data on the next refresh. """ _require_manage(user) handler_key = CONTROL_HANDLER_KEYS.get(control_key) if not handler_key: raise HTTPException( status_code=404, detail=f"unknown control_key {control_key!r}", ) job_row = (await db.execute( select(ScheduledJob).where( ScheduledJob.handler_key == handler_key ) )).scalar_one_or_none() if job_row is None: raise HTTPException( status_code=503, detail=( f"scheduled job for handler_key={handler_key!r} not found " "— migrations may be stale." ), ) job_id = job_row.id # Run synchronously so the user sees fresh numbers on next page # load. ``execute_job`` opens its own session — release ours first # to avoid holding a transaction across what may be a multi-second # upstream sync. await db.commit() from app.scheduled_jobs.service import execute_job result = await execute_job( job_id, triggered_by=f"user_manual:{user.id}", ) return success_response( data={ "control_key": control_key, "handler_key": handler_key, "job_id": str(job_id), "status": result.get("status"), "duration_ms": result.get("duration_ms"), "result_summary": result.get("result_summary"), "error": result.get("error"), "ran_at": datetime.now(UTC).isoformat(), }, request=request, ) @router.get("/summary") async def get_summary( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) summary = await DashboardService.summary(db) cycle = summary["privileged_open_cycle"] last_ep = summary["last_evidence_pack"] payload = audsch.AuditorDashboardSummary( mfa_unenrolled_count=summary["mfa_unenrolled_count"], mfa_total_count=summary["mfa_total_count"], privileged_open_cycle=( audsch.PrivilegedReviewCycleSummary.model_validate(cycle) if cycle else None ), public_exposure_open_count=summary["public_exposure_open_count"], public_exposure_critical_count=summary["public_exposure_critical_count"], secret_overdue_count=summary["secret_overdue_count"], signin_high_anomaly_24h=summary["signin_high_anomaly_24h"], pat_risky_count=summary["pat_risky_count"], cert_expiring_30d=summary["cert_expiring_30d"], risk_critical_count=summary["risk_critical_count"], risk_high_count=summary["risk_high_count"], last_evidence_pack=( audsch.EvidencePackRow.model_validate(last_ep) if last_ep else None ), ) return success_response(data=payload.model_dump(mode="json"), request=request) # ───────────────────────────────────────────────────────────────────── # Unified findings (cross-control worklist) # ───────────────────────────────────────────────────────────────────── @router.get("/findings") async def list_findings( request: Request, page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), status: str | None = Query(None), severity: str | None = Query(None), control_key: str | None = Query(None), open_only: bool = Query(False), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) result = await findings_service.list_findings( db, status=status, severity=severity, control_key=control_key, open_only=open_only, page=page, page_size=page_size, ) summary = await findings_service.findings_summary(db) payload = audsch.FindingListResponse( items=[audsch.FindingRow(**row) for row in result["items"]], meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=result["total"], ), summary=summary, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.get("/findings/export.xlsx") async def export_findings( request: Request, status: str | None = Query(None), severity: str | None = Query(None), control_key: str | None = Query(None), open_only: bool = Query(False), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Export the filtered findings worklist as XLSX. Filter signature mirrors :func:`list_findings` minus pagination — the worklist passes the exact query params it is showing, so the download matches the screen. RBAC mirrors the list endpoint (``can_view_security``). Reuses ``findings_service.list_findings`` so any filter fix on the list applies here automatically. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response _require_view(user) result = await findings_service.list_findings( db, status=status, severity=severity, control_key=control_key, open_only=open_only, page=1, page_size=EXPORT_ROW_LIMIT, ) columns = [ ("title", "标题"), ("control_key", "控制项"), ("finding_type", "发现类型"), ("severity", "严重度"), ("status", "状态"), ("subject_type", "对象类型"), ("subject_display", "对象"), ("person_full_name", "关联人"), ("person_email", "人员邮箱"), ("source_table", "来源表"), ("first_seen_at", "首次发现"), ("last_seen_at", "最近发现"), ("id", "ID"), ] # ``subject_display`` is the human label but is nullable — fall # back to the always-present ``subject_ref`` so the 对象 cell is # never blank for machine-keyed subjects. rows = [ { **row, "subject_display": row.get("subject_display") or row.get("subject_ref"), } for row in result["items"] ] return await xlsx_export_response( db, request, user, columns=columns, rows=rows, total_matched=result["total"], filename_stem="auditor_findings", sheet_title="Findings", audit_action="auditor.findings.export", audit_resource_type="auditor_finding", filters={ "status": status, "severity": severity, "control_key": control_key, "open_only": open_only, }, ) @router.get("/findings/summary") async def get_findings_summary( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) summary = await findings_service.findings_summary(db) return success_response(data=summary, request=request) @router.post("/findings/{finding_id}/decide") async def decide_finding( finding_id: UUID, request: Request, body: audsch.FindingDecisionRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) row = await findings_service.decide_finding( db, finding_id=finding_id, decision=body.decision, note=body.note, accepted_days=body.accepted_days, actor_user_id=user.id, trace_id=_trace_id(request), ) await db.flush() view = await findings_service.view_finding(db, row) return success_response( data=audsch.FindingRow(**view).model_dump(mode="json"), request=request, ) # ───────────────────────────────────────────────────────────────────── # MFA # ───────────────────────────────────────────────────────────────────── # How far back a free-text sign-in search reaches when the caller did not # say. 30 days is what the shells display; an explicit `since_hours` still # wins, and a search-free listing is unchanged. _SEARCH_WINDOW_HOURS = 720 def _apply_text_search( q: Any, cq: Any, term: str | None, columns: Sequence[Any], *, person_fk: Any | None = None, ) -> tuple[Any, Any]: """Apply one case-insensitive OR-match across ``columns`` to both the row query and its COUNT twin. The six auditor list endpoints each had only enum filters, yet the mobile shells all render a SearchBar — so typing a term refetched page 1 and returned an identical list. A search box that does nothing is worse than no search box: the user concludes the record is not there. Added 2026-07-28 alongside the mobile infinite-scroll shell, which made the dead box more obvious by printing "showing 50 of 1200" next to it. Both queries must get the SAME predicate or the total lies about the filtered set (the exact bug shape that makes "showing 50 of 1200" wrong). ``person_fk`` opts the row's person into the searchable columns by outer-joining ``identity_persons``. It is not optional garnish: the MFA and sign-in shells label their search box "搜索员工 / Search employees", but neither model stores a name — the display name is stitched in at serialisation time from ``_persons_index``. Searching "张伟" therefore matched nothing and the shell rendered "没有匹配的结果", which reads as "this person has no MFA record" rather than "this box cannot search names". A search box that lies about WHAT it searches is the same class of defect as one that does not search at all. The term is escaped before it goes into the LIKE pattern. Unescaped, a lone ``_`` matches any single character and ``%`` matches everything — so searching for a literal underscore (common in resource ids and secret names) silently returns the whole table. """ term = (term or "").strip() if not term: return q, cq # `\` is the escape char declared on the ilike() calls below. escaped = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{escaped}%" cols = list(columns) if person_fk is not None: onclause = IdentityPerson.id == person_fk q = q.outerjoin(IdentityPerson, onclause) cq = cq.outerjoin(IdentityPerson, onclause) cols += [IdentityPerson.display_name, IdentityPerson.work_email] predicate = or_(*[c.ilike(like, escape="\\") for c in cols]) return q.where(predicate), cq.where(predicate) @router.get("/mfa-status") async def list_mfa_status( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=200), provider: str | None = Query(None), is_enrolled: bool | None = Query(None), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorMfaStatus) cq = select(func.count()).select_from(AuditorMfaStatus) if provider: q = q.where(AuditorMfaStatus.provider == provider) cq = cq.where(AuditorMfaStatus.provider == provider) if is_enrolled is not None: q = q.where(AuditorMfaStatus.is_enrolled.is_(is_enrolled)) cq = cq.where(AuditorMfaStatus.is_enrolled.is_(is_enrolled)) q, cq = _apply_text_search( q, cq, search, (AuditorMfaStatus.provider, AuditorMfaStatus.provider_email,), person_fk=AuditorMfaStatus.person_id, ) q = q.order_by( # Observed gaps first, unknowns last — a row whose provider could # not answer is not the auditor's most urgent item. AuditorMfaStatus.enrollment_observed.desc(), AuditorMfaStatus.is_enrolled, AuditorMfaStatus.provider, ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) persons = await _persons_index(db, [r.person_id for r in rows]) items = [] for r in rows: p = persons.get(r.person_id) items.append(audsch.MfaStatusRow( id=r.id, person_id=r.person_id, person_full_name=p.display_name if p else None, person_email=p.work_email if p else None, employment_status=p.employment_status if p else None, department_path=None, provider=r.provider, provider_email=r.provider_email, is_enrolled=r.is_enrolled, is_enforced=r.is_enforced, enrollment_observed=r.enrollment_observed, enforcement_observed=r.enforcement_observed, enrollment_methods=r.enrollment_methods, last_verified_at=r.last_verified_at, last_synced_at=r.last_synced_at, )) summary = { "providers": list(ALL_MFA_PROVIDERS), "total": total, } payload = audsch.MfaStatusListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary=summary, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/mfa-status/sync") async def trigger_mfa_sync( request: Request, body: audsch.MfaSyncTriggerRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await MfaSyncService.sync( db, providers=body.providers, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.MfaSyncResponse( triggered_providers=result["providers"], person_count=result["person_count"], synced_at=result["synced_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) # ───────────────────────────────────────────────────────────────────── # Public exposure # ───────────────────────────────────────────────────────────────────── @router.get("/public-exposure") async def list_public_exposure( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=200), status: str | None = Query(None), severity: str | None = Query(None), source: str | None = Query(None), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorPublicExposureFinding) cq = select(func.count()).select_from(AuditorPublicExposureFinding) if status: q = q.where(AuditorPublicExposureFinding.status == status) cq = cq.where(AuditorPublicExposureFinding.status == status) if severity: q = q.where(AuditorPublicExposureFinding.severity == severity) cq = cq.where(AuditorPublicExposureFinding.severity == severity) if source: q = q.where(AuditorPublicExposureFinding.source == source) cq = cq.where(AuditorPublicExposureFinding.source == source) q, cq = _apply_text_search( q, cq, search, (AuditorPublicExposureFinding.resource_name, AuditorPublicExposureFinding.resource_id, AuditorPublicExposureFinding.exposure_type, AuditorPublicExposureFinding.source,), ) q = q.order_by( AuditorPublicExposureFinding.severity.desc(), AuditorPublicExposureFinding.last_seen_at.desc(), ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) items = [ audsch.PublicExposureRow.model_validate(r) for r in rows ] summary = await _public_exposure_summary(db) payload = audsch.PublicExposureListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary=summary, ) return success_response(data=payload.model_dump(mode="json"), request=request) async def _public_exposure_summary(db: AsyncSession) -> dict[str, Any]: by_status = (await db.execute( select( AuditorPublicExposureFinding.status, func.count(), ).group_by(AuditorPublicExposureFinding.status) )).all() by_severity = (await db.execute( select( AuditorPublicExposureFinding.severity, func.count(), ).where( AuditorPublicExposureFinding.status == "open" ).group_by(AuditorPublicExposureFinding.severity) )).all() return { "by_status": {s: int(n) for s, n in by_status}, "open_by_severity": {s: int(n) for s, n in by_severity}, } @router.post("/public-exposure/scan") async def trigger_public_exposure_scan( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await PublicExposureScanService.scan( db, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.ExposureScanTriggerResponse( scan_run_id=result["scan_run_id"], sources_scanned=result["sources_scanned"], partial_sources=result.get("partial_sources") or {}, findings_count=result["findings_count"], started_at=result["started_at"], completed_at=result["completed_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/public-exposure/{finding_id}/resolve") async def resolve_public_exposure( finding_id: UUID, request: Request, body: audsch.ExposureResolveRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) f = await ExposureService.resolve( db, finding_id=finding_id, resolution=body.resolution, note=body.note, actor_user_id=user.id, trace_id=_trace_id(request), ) return success_response( data=audsch.PublicExposureRow.model_validate(f).model_dump( mode="json" ), request=request, ) # ───────────────────────────────────────────────────────────────────── # Secret rotation # ───────────────────────────────────────────────────────────────────── @router.get("/secret-rotation") async def list_secret_rotation( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=200), status: str | None = Query(None), secret_kind: str | None = Query(None), source: str | None = Query(None), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorSecretRotationRecord) cq = select(func.count()).select_from(AuditorSecretRotationRecord) if status: q = q.where(AuditorSecretRotationRecord.status == status) cq = cq.where(AuditorSecretRotationRecord.status == status) if secret_kind: q = q.where(AuditorSecretRotationRecord.secret_kind == secret_kind) cq = cq.where(AuditorSecretRotationRecord.secret_kind == secret_kind) if source: q = q.where(AuditorSecretRotationRecord.source == source) cq = cq.where(AuditorSecretRotationRecord.source == source) # MySQL/TiDB do not support ``NULLS LAST`` syntax; emulate via the # ``IS NULL`` sort key trick so NULL ``next_rotation_due_at`` sort last. q, cq = _apply_text_search( q, cq, search, (AuditorSecretRotationRecord.resource_name, AuditorSecretRotationRecord.resource_id, AuditorSecretRotationRecord.owner_email, AuditorSecretRotationRecord.secret_kind,), person_fk=AuditorSecretRotationRecord.owner_person_id, ) q = q.order_by( AuditorSecretRotationRecord.status.desc(), AuditorSecretRotationRecord.next_rotation_due_at.is_(None), AuditorSecretRotationRecord.next_rotation_due_at.asc(), ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) items = [ audsch.SecretRotationRow.model_validate(r) for r in rows ] by_status = (await db.execute( select( AuditorSecretRotationRecord.status, func.count(), ).group_by(AuditorSecretRotationRecord.status) )).all() payload = audsch.SecretRotationListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary={"by_status": {s: int(n) for s, n in by_status}}, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/secret-rotation/sync") async def trigger_secret_rotation_sync( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await SecretRotationSyncService.sync( db, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.SecretRotationSyncResponse( scanned_sources=result["scanned_sources"], record_count=result["record_count"], overdue_count=result["overdue_count"], synced_at=result["synced_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) # ───────────────────────────────────────────────────────────────────── # Signin events # ───────────────────────────────────────────────────────────────────── @router.get("/signin-events") async def list_signin_events( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=500), anomaly_level: str | None = Query(None), provider: str | None = Query(None), person_id: UUID | None = Query(None), since_hours: int | None = Query(None, ge=1, le=720), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorSigninEvent) cq = select(func.count()).select_from(AuditorSigninEvent) if anomaly_level: q = q.where(AuditorSigninEvent.anomaly_level == anomaly_level) cq = cq.where(AuditorSigninEvent.anomaly_level == anomaly_level) if provider: q = q.where(AuditorSigninEvent.provider == provider) cq = cq.where(AuditorSigninEvent.provider == provider) if person_id is not None: q = q.where(AuditorSigninEvent.person_id == person_id) cq = cq.where(AuditorSigninEvent.person_id == person_id) # A free-text search over four columns is a leading-wildcard LIKE, so # it cannot use an index; on `auditor_signin_events` — the one # high-volume table here — unbounded that is two full scans (rows + # COUNT) per keystroke-submitted query. Bound the SEARCH path only: # changing the endpoint's default outright would silently narrow every # existing caller's result set, which is its own kind of lie. effective_since = since_hours or (_SEARCH_WINDOW_HOURS if search else None) if effective_since: cutoff = datetime.now(UTC) - timedelta(hours=effective_since) q = q.where(AuditorSigninEvent.event_at >= cutoff) cq = cq.where(AuditorSigninEvent.event_at >= cutoff) q, cq = _apply_text_search( q, cq, search, (AuditorSigninEvent.provider_email, AuditorSigninEvent.source_ip, AuditorSigninEvent.source_city, AuditorSigninEvent.source_country,), person_fk=AuditorSigninEvent.person_id, ) q = q.order_by( AuditorSigninEvent.event_at.desc() ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) persons = await _persons_index( db, [r.person_id for r in rows if r.person_id], ) items = [] for r in rows: p = persons.get(r.person_id) if r.person_id else None items.append(audsch.SigninEventRow( id=r.id, person_id=r.person_id, person_full_name=p.display_name if p else None, provider=r.provider, provider_email=r.provider_email, event_at=r.event_at, event_kind=r.event_kind, status=r.status, source_ip=r.source_ip, source_country=r.source_country, source_city=r.source_city, user_agent=r.user_agent, anomaly_score=r.anomaly_score, anomaly_level=r.anomaly_level, anomaly_reasons=r.anomaly_reasons, acknowledged_at=r.acknowledged_at, acknowledged_by_person_id=r.acknowledged_by_person_id, )) by_level = (await db.execute( select(AuditorSigninEvent.anomaly_level, func.count()) .group_by(AuditorSigninEvent.anomaly_level) )).all() payload = audsch.SigninEventListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary={ "by_anomaly_level": {level: int(n) for level, n in by_level}, }, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/signin-events/pull") async def trigger_signin_pull( request: Request, lookback_minutes: int = Query(70, ge=10, le=1440), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await SigninAnomalyService.pull_and_score( db, lookback_minutes=lookback_minutes, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.SigninPullResponse( pulled_providers=result["pulled_providers"], new_event_count=result["new_event_count"], anomaly_count=result["anomaly_count"], pulled_at=result["pulled_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/signin-events/{event_id}/acknowledge") async def acknowledge_signin_event( event_id: UUID, request: Request, body: audsch.AcknowledgeSigninRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) ev = await SigninAcknowledgeService.acknowledge( db, event_id=event_id, note=body.note, actor_user_id=user.id, trace_id=_trace_id(request), ) return success_response( data={"id": str(ev.id), "acknowledged_at": ev.acknowledged_at.isoformat() if ev.acknowledged_at else None}, request=request, ) # ───────────────────────────────────────────────────────────────────── # PAT audit # ───────────────────────────────────────────────────────────────────── @router.get("/pat-audit") async def list_pat_audit( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=200), kind: str | None = Query(None), status: str | None = Query(None), has_risk: bool | None = Query(None), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorPatAudit) cq = select(func.count()).select_from(AuditorPatAudit) if kind: q = q.where(AuditorPatAudit.kind == kind) cq = cq.where(AuditorPatAudit.kind == kind) if status: q = q.where(AuditorPatAudit.status == status) cq = cq.where(AuditorPatAudit.status == status) if has_risk is True: q = q.where(AuditorPatAudit.risk_flags.is_not(None)) cq = cq.where(AuditorPatAudit.risk_flags.is_not(None)) elif has_risk is False: q = q.where(AuditorPatAudit.risk_flags.is_(None)) cq = cq.where(AuditorPatAudit.risk_flags.is_(None)) q, cq = _apply_text_search( q, cq, search, (AuditorPatAudit.token_name, AuditorPatAudit.owner_email, AuditorPatAudit.kind, AuditorPatAudit.source,), person_fk=AuditorPatAudit.person_id, ) q = q.order_by( AuditorPatAudit.kind, AuditorPatAudit.last_used_at, ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) items = [audsch.PatAuditRow.model_validate(r) for r in rows] by_kind = (await db.execute( select(AuditorPatAudit.kind, func.count()) .group_by(AuditorPatAudit.kind) )).all() payload = audsch.PatAuditListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary={"by_kind": {k: int(n) for k, n in by_kind}}, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/pat-audit/sync") async def trigger_pat_sync( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await PatAuditService.sync( db, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.PatAuditSyncResponse( pulled_sources=result["pulled_sources"], record_count=result["record_count"], risky_count=result["risky_count"], synced_at=result["synced_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) # ───────────────────────────────────────────────────────────────────── # Cert expiry # ───────────────────────────────────────────────────────────────────── @router.get("/cert-expiries") async def list_cert_expiries( request: Request, page: int = Query(1, ge=1), search: str | None = Query( None, # Upper bound only. The predicate is a leading-wildcard LIKE across # 4-6 TEXT columns, so it scans regardless of term length and a # 100 KB term is the shape worth refusing. # # Deliberately NO `min_length`: one character is a perfectly # ordinary query in Chinese and Japanese — an auditor looking for # 张伟 may well type 张 — and rejecting it with a 422 would be a # usability regression dressed up as a safeguard. The endpoints are # already behind `_require_view` (auditor / global_admin), roles # that can read the whole table anyway. max_length=100, description="Case-insensitive free-text match across the row's label columns.", ), page_size: int = Query(50, ge=1, le=500), asset_kind: str | None = Query(None), alert_level: str | None = Query(None), status: str | None = Query(None), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorCertExpiry) cq = select(func.count()).select_from(AuditorCertExpiry) if asset_kind: q = q.where(AuditorCertExpiry.asset_kind == asset_kind) cq = cq.where(AuditorCertExpiry.asset_kind == asset_kind) if alert_level: q = q.where(AuditorCertExpiry.alert_level == alert_level) cq = cq.where(AuditorCertExpiry.alert_level == alert_level) if status: q = q.where(AuditorCertExpiry.status == status) cq = cq.where(AuditorCertExpiry.status == status) q, cq = _apply_text_search( q, cq, search, (AuditorCertExpiry.name, AuditorCertExpiry.subject, AuditorCertExpiry.issuer, AuditorCertExpiry.owner_email,), person_fk=AuditorCertExpiry.owner_person_id, ) q = q.order_by( AuditorCertExpiry.expires_at.asc(), ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) items = [audsch.CertExpiryRow.model_validate(r) for r in rows] by_alert = (await db.execute( select(AuditorCertExpiry.alert_level, func.count()) .group_by(AuditorCertExpiry.alert_level) )).all() payload = audsch.CertExpiryListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary={"by_alert_level": {a: int(n) for a, n in by_alert}}, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/cert-expiries/sync") async def trigger_cert_sync( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await CertExpirySyncService.sync( db, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.CertExpirySyncResponse( pulled_sources=result["pulled_sources"], record_count=result["record_count"], alerting_count=result["alerting_count"], synced_at=result["synced_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/cert-expiries/manual") async def upsert_cert_manual( request: Request, body: audsch.CertManualUpsertRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) row = await CertManualService.upsert_manual( db, asset_kind=body.asset_kind, external_id=body.external_id, name=body.name, subject=body.subject, issuer=body.issuer, issued_at=body.issued_at, expires_at=body.expires_at, owner_person_id=body.owner_person_id, owner_email=body.owner_email, environment=body.environment, notes=body.notes, actor_user_id=user.id, trace_id=_trace_id(request), ) return success_response( data=audsch.CertExpiryRow.model_validate(row).model_dump( mode="json" ), request=request, ) # ───────────────────────────────────────────────────────────────────── # Risk score # ───────────────────────────────────────────────────────────────────── @router.get("/risk-scores") async def list_risk_scores( request: Request, page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=500), level: str | None = Query(None), min_score: int | None = Query(None, ge=0, le=100), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorRiskScore) cq = select(func.count()).select_from(AuditorRiskScore) if level: q = q.where(AuditorRiskScore.level == level) cq = cq.where(AuditorRiskScore.level == level) if min_score is not None: q = q.where(AuditorRiskScore.score >= min_score) cq = cq.where(AuditorRiskScore.score >= min_score) q = q.order_by( AuditorRiskScore.score.desc() ).offset((page - 1) * page_size).limit(page_size) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) persons = await _persons_index(db, [r.person_id for r in rows]) items = [] for r in rows: p = persons.get(r.person_id) items.append(audsch.RiskScoreRow( id=r.id, person_id=r.person_id, person_full_name=p.display_name if p else None, person_email=p.work_email if p else None, employment_status=p.employment_status if p else None, department_path=None, score=r.score, level=r.level, factors=r.factors, breakdown=r.breakdown, computed_at=r.computed_at, )) by_level = (await db.execute( select(AuditorRiskScore.level, func.count()) .group_by(AuditorRiskScore.level) )).all() # Full-population score histogram (NOT page-bounded) so the frontend # chart reflects every scored person, not just the current 50-row page. # Buckets are 0-9, 10-19, …, 90-100 (score==100 folds into the top # bucket). Computed via FLOOR(score/10) group-by — one extra aggregate # query, O(1) in fleet size. by_bucket_raw = (await db.execute( select(func.floor(AuditorRiskScore.score / 10), func.count()) .group_by(func.floor(AuditorRiskScore.score / 10)) )).all() by_score_bucket: dict[str, int] = {} for bucket_idx, n in by_bucket_raw: idx = min(int(bucket_idx), 9) lo = idx * 10 hi = 100 if idx == 9 else lo + 9 key = f"{lo}-{hi}" by_score_bucket[key] = by_score_bucket.get(key, 0) + int(n) payload = audsch.RiskScoreListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), summary={ "by_level": {lvl: int(n) for lvl, n in by_level}, "by_score_bucket": by_score_bucket, }, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/risk-scores/recompute") async def trigger_risk_recompute( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) result = await RiskScoreService.recompute_all( db, actor_user_id=user.id, trace_id=_trace_id(request), ) payload = audsch.RiskScoreRecomputeResponse( person_count=result["person_count"], high_count=result["high_count"], critical_count=result["critical_count"], computed_at=result["computed_at"], ) return success_response(data=payload.model_dump(mode="json"), request=request) # ───────────────────────────────────────────────────────────────────── # Evidence pack # ───────────────────────────────────────────────────────────────────── @router.get("/evidence-packs") async def list_evidence_packs( request: Request, page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_view(user) q = select(AuditorEvidencePack).order_by( AuditorEvidencePack.created_at.desc() ).offset((page - 1) * page_size).limit(page_size) cq = select(func.count()).select_from(AuditorEvidencePack) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) items = [audsch.EvidencePackRow.model_validate(r) for r in rows] payload = audsch.EvidencePackListResponse( items=items, meta=audsch.PaginatedMeta( page=page, page_size=page_size, total=total, ), ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/evidence-packs") async def request_evidence_pack( request: Request, body: audsch.EvidencePackBuildRequest, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_manage(user) pack = await EvidencePackService.request_build( db, title=body.title, period_start=body.period_start, period_end=body.period_end, modules=body.modules, actor_user_id=user.id, trace_id=_trace_id(request), ) return success_response( data=audsch.EvidencePackRow.model_validate(pack).model_dump( mode="json" ), request=request, ) @router.post("/evidence-packs/{pack_id}/build") async def build_evidence_pack_now( pack_id: UUID, request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Synchronous build (admin-trigger). Worker also picks pending rows from the queue. """ _require_manage(user) pack = await EvidencePackService.generate(db, pack_id=pack_id) return success_response( data=audsch.EvidencePackRow.model_validate(pack).model_dump( mode="json" ), request=request, ) @router.get("/evidence-packs/{pack_id}/download") async def download_evidence_pack( pack_id: UUID, request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Same tier as reading a bundle, and for the same reason: since the # ``access_evidence`` module landed, this ZIP carries frozen evidence # rows — named people's holdings with display names in the clear, # including the complete footprint archived at offboarding. Narrowing # the bundle API while the pack containing the identical rows stayed # open to the leadership read tier would have been a door that only # looks locked. _require_forensic_read(user) pack = await db.get(AuditorEvidencePack, pack_id) if pack is None: raise NotFoundError("AuditorEvidencePack", str(pack_id)) if pack.status != "completed" or not pack.file_path: raise HTTPException( status_code=409, detail=f"Pack {pack_id} is in status {pack.status!r}, " f"not yet downloadable", ) # Round-49: re-check size > 0 in addition to existence so a # truncated / mid-write file doesn't get an audit row claiming # "downloaded" when the user receives an empty stream. The audit # row used to commit via ``get_db``'s post-yield code AFTER # FileResponse started streaming — too late to know whether # the stream actually delivered bytes. try: st = os.stat(pack.file_path) except OSError: raise HTTPException( status_code=410, detail="Pack file not on disk (retention swept or " "evidence root remounted)", ) from None if st.st_size <= 0: raise HTTPException( status_code=410, detail="Pack file is empty (mid-write or corrupted)", ) pack.download_count = (pack.download_count or 0) + 1 pack.last_downloaded_at = datetime.now(UTC) # Evidence pack downloads expose every privileged-review snapshot, # MFA roster, signin event, etc. that fed the auditor module — i.e. # the entire compliance surface for the period. Each access must be # traceable in the audit trail. await AuditService(db).log( trace_id=_trace_id(request), actor_type="user", actor_user_id=user.id, action="auditor.evidence_pack.download", status="success", resource_type="auditor_evidence_pack", resource_id=str(pack_id), detail={ "pack_title": pack.title, "period_start": ( pack.period_start.isoformat() if pack.period_start else None ), "period_end": ( pack.period_end.isoformat() if pack.period_end else None ), "modules": list(pack.modules) if pack.modules else [], "download_count_after": pack.download_count, "file_size_bytes": st.st_size, }, ) # Round-49: commit before FileResponse streams. The previous # ``flush``+post-yield-commit pattern committed the audit row # AFTER streaming started — if FileResponse's path raced a # retention sweep that deleted the file between size check and # stream open, the audit row still landed claiming a success # the user never got. Commit synchronously so the audit is # durable before any stream-side I/O. await db.commit() return FileResponse( pack.file_path, media_type="application/zip", filename=f"helix-evidence-{pack_id}.zip", ) # ── Evidence queries ───────────────────────────────────────────────── # Registered last so the deterministic evidence surface shares this # router's prefix, RBAC helpers and trace-id extraction rather than # growing a second, quietly diverging definition of "who may read # security data". register_evidence_query_routes( router, require_view=_require_view, require_forensic_read=_require_forensic_read, require_manage=_require_manage, get_current_user=get_current_user, trace_id_of=_trace_id, )