"""Unified findings engine — the cross-control actionable worklist. Every detective control reconciles its current violating items into :class:`AuditorFinding` rows through :func:`reconcile_findings`. The finding carries the uniform lifecycle (open → acknowledged → accepted_risk / resolved) so a single inbox (and a single posture rollup) spans every control. Two flavours of producer: * **Native** controls (MFA, secret rotation, cert expiry, PAT audit) have no pre-existing per-item decision store, so the *finding* is the home of the decision. ``reconcile_findings(..., source_managed=False)`` preserves operator decisions across re-syncs and re-opens accepted-risk findings whose ``accepted_until`` has passed. * **Mirrored** controls (public exposure, sign-in anomaly) already keep a decision on their legacy row. There the *source row* stays the home and the finding is a projection of it (``source_managed=True``); the unified :func:`decide_finding` writes the source row and keeps the finding in lock-step, and the legacy endpoints call :func:`sync_mirror_finding` so both surfaces stay consistent on every write path. Datetime discipline: ``DateTime(timezone=True)`` columns round-trip as naive on fresh load, so this module computes in **naive UTC** throughout (matching ``compliance_service``) to avoid aware/naive subtraction 500s. """ from __future__ import annotations import hashlib import uuid from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, or_, select, true from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.auditor import constants as const from app.domains.auditor.models import ( AuditorCertExpiry, AuditorFinding, AuditorMfaStatus, AuditorPatAudit, AuditorPublicExposureFinding, AuditorSecretRotationRecord, AuditorSigninEvent, ) from app.identity.person_models import IdentityPerson # Source-table identifiers for mirrored controls. SOURCE_TABLE_EXPOSURE = AuditorPublicExposureFinding.__tablename__ SOURCE_TABLE_SIGNIN = AuditorSigninEvent.__tablename__ def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) def _to_naive(dt: datetime | None) -> datetime | None: if dt is None: return None if dt.tzinfo is not None: return dt.astimezone(UTC).replace(tzinfo=None) return dt def make_dedup_key(*parts: Any) -> str: """Stable per-control identity for upsert. Joins parts with ``|``; if the result would overflow the 255-char column it is replaced by a sha1 hex digest (still deterministic, so re-syncs land on the same row). """ raw = "|".join("" if p is None else str(p) for p in parts) if len(raw) <= 255: return raw # Stable database identity, not secret protection. Historical findings # persist this exact key; changing it would duplicate open controls. digest = hashlib.sha1( raw.encode("utf-8"), usedforsecurity=False, ).hexdigest() return "sha1:" + digest @dataclass class FindingSpec: """One desired finding emitted by a control producer.""" dedup_key: str severity: str title: str subject_type: str subject_ref: str | None = None subject_display: str | None = None person_id: uuid.UUID | None = None detail: dict[str, Any] | None = None source_table: str | None = None source_id: uuid.UUID | None = None # Only consulted when source_managed=True — the source row's lifecycle # status mapped into finding terms. status: str | None = None def _refresh_facts(row: AuditorFinding, spec: FindingSpec) -> None: """Update the mutable evidence fields (not the lifecycle status).""" row.severity = spec.severity row.title = spec.title row.subject_type = spec.subject_type row.subject_ref = spec.subject_ref row.subject_display = spec.subject_display row.person_id = spec.person_id row.detail = spec.detail if spec.source_table is not None: row.source_table = spec.source_table if spec.source_id is not None: row.source_id = spec.source_id async def reconcile_findings( db: AsyncSession, *, control_key: str, finding_type: str, specs: list[FindingSpec], source_managed: bool = False, ) -> dict[str, int]: """Upsert the desired findings for one (control, type); auto-resolve any prior open finding whose item has disappeared. Idempotent — safe to run on every sync. Does NOT commit (the caller's sync transaction owns the commit). """ now = _naive_now() existing = { r.dedup_key: r for r in (await db.execute( select(AuditorFinding).where( AuditorFinding.control_key == control_key, AuditorFinding.finding_type == finding_type, ) )).scalars().all() } desired: set[str] = set() counts = {"created": 0, "reopened": 0, "auto_resolved": 0, "present": 0} for spec in specs: desired.add(spec.dedup_key) row = existing.get(spec.dedup_key) if row is None: status = ( spec.status if (source_managed and spec.status) else const.FINDING_STATUS_OPEN ) db.add(AuditorFinding( control_key=control_key, finding_type=finding_type, dedup_key=spec.dedup_key, subject_type=spec.subject_type, subject_ref=spec.subject_ref, subject_display=spec.subject_display, person_id=spec.person_id, severity=spec.severity, title=spec.title, detail=spec.detail, status=status, source_table=spec.source_table, source_id=spec.source_id, first_seen_at=now, last_seen_at=now, resolved_at=( now if status == const.FINDING_STATUS_RESOLVED else None ), )) counts["created"] += 1 continue row.last_seen_at = now _refresh_facts(row, spec) counts["present"] += 1 if source_managed: # Source row owns the decision — mirror its status verbatim. if spec.status and row.status != spec.status: _set_status(row, spec.status, now) continue # Native lifecycle — preserve operator decisions, re-open stale ones. if row.status == const.FINDING_STATUS_RESOLVED: # It came back — it was never actually fixed. _reopen(row) counts["reopened"] += 1 elif row.status == const.FINDING_STATUS_ACCEPTED_RISK: au = _to_naive(row.accepted_until) if au is not None and now > au: _reopen(row) counts["reopened"] += 1 # open / acknowledged → leave as-is. # Auto-resolve findings whose item is no longer present. for key, row in existing.items(): if key in desired: continue if row.status == const.FINDING_STATUS_RESOLVED: continue row.status = const.FINDING_STATUS_RESOLVED row.auto_resolved = True row.resolved_at = now counts["auto_resolved"] += 1 return counts def _set_status(row: AuditorFinding, status: str, now: datetime) -> None: row.status = status if status == const.FINDING_STATUS_RESOLVED: row.resolved_at = now elif status == const.FINDING_STATUS_OPEN: row.resolved_at = None row.auto_resolved = False def _reopen(row: AuditorFinding) -> None: row.status = const.FINDING_STATUS_OPEN row.auto_resolved = False row.resolved_at = None row.decided_by_person_id = None row.decided_at = None row.accepted_until = None # ════════════════════════════════════════════════════════════════════ # Mirrored-control single-row sync (called by legacy endpoints) # ════════════════════════════════════════════════════════════════════ async def sync_mirror_finding( db: AsyncSession, *, control_key: str, finding_type: str, spec: FindingSpec, ) -> AuditorFinding | None: """Upsert the single finding mirroring one source row's current state. ``spec.status`` carries the source row's lifecycle status; pass it as resolved (with the underlying item still present is impossible here — the legacy endpoint just decided it) to keep the worklist in lock-step. """ now = _naive_now() row = (await db.execute( select(AuditorFinding).where( AuditorFinding.control_key == control_key, AuditorFinding.dedup_key == spec.dedup_key, ) )).scalar_one_or_none() if row is None: row = AuditorFinding( control_key=control_key, finding_type=finding_type, dedup_key=spec.dedup_key, first_seen_at=now, last_seen_at=now, ) db.add(row) row.last_seen_at = now _refresh_facts(row, spec) if spec.status: _set_status(row, spec.status, now) return row # ════════════════════════════════════════════════════════════════════ # Decision (the single human write path) # ════════════════════════════════════════════════════════════════════ # Which decisions are valid for a mirrored control whose source row has a # narrower lifecycle than the finding model. _EXPOSURE_DECISIONS = { const.FINDING_DECISION_ACCEPT_RISK, const.FINDING_DECISION_RESOLVE, const.FINDING_DECISION_REOPEN, } _SIGNIN_DECISIONS = { const.FINDING_DECISION_ACKNOWLEDGE, const.FINDING_DECISION_RESOLVE, const.FINDING_DECISION_REOPEN, } async def decide_finding( db: AsyncSession, *, finding_id: uuid.UUID, decision: str, note: str, accepted_days: int | None, actor_user_id: uuid.UUID, trace_id: uuid.UUID | None = None, ) -> AuditorFinding: """Apply an operator decision to a finding (and mirror to its source).""" if decision not in const.ALL_FINDING_DECISIONS: raise AppException( code="INVALID_DECISION", message=f"decision must be one of {const.ALL_FINDING_DECISIONS}", status_code=422, ) # Bound free-text + numeric inputs at the service layer so EVERY caller # is safe — the HTTP route validates via FindingDecisionRequest, but the # MCP twin (decide_security_finding) had no schema and could write an # unbounded note / absurd accepted_days straight through. note = (note or "").strip() if not note: raise AppException( code="NOTE_REQUIRED", message="a decision note is required", status_code=422, ) if len(note) > 2000: raise AppException( code="NOTE_TOO_LONG", message="decision note must be <= 2000 characters", status_code=422, ) if accepted_days is not None and not (1 <= accepted_days <= 365): raise AppException( code="INVALID_ACCEPTED_DAYS", message="accepted_days must be between 1 and 365", status_code=422, ) row = await db.get(AuditorFinding, finding_id) if row is None: raise NotFoundError("AuditorFinding", str(finding_id)) # Reject decisions the source's narrower lifecycle can't carry, so the # worklist never silently drops a decision a re-sync would overwrite. if row.source_table == SOURCE_TABLE_EXPOSURE and ( decision not in _EXPOSURE_DECISIONS ): raise AppException( code="UNSUPPORTED_DECISION", message=( "public-exposure findings support accept_risk / resolve / " "reopen only" ), status_code=422, ) if row.source_table == SOURCE_TABLE_SIGNIN and ( decision not in _SIGNIN_DECISIONS ): raise AppException( code="UNSUPPORTED_DECISION", message=( "sign-in findings support acknowledge / resolve / reopen only" ), status_code=422, ) # Re-opening only makes sense while the underlying item still exists; # an auto-resolved finding means the item disappeared from its source. if decision == const.FINDING_DECISION_REOPEN and row.auto_resolved: raise AppException( code="ITEM_GONE", message="cannot reopen — the underlying item is no longer present", status_code=422, ) now = _naive_now() if decision == const.FINDING_DECISION_ACKNOWLEDGE: # For sign-in anomalies, acknowledging IS the disposition (there is # no separate "fix"); the legacy ack endpoint + the projection both # treat an acknowledged event as handled, so acknowledge is terminal # here — keeping both write paths and the next projection consistent. if row.source_table == SOURCE_TABLE_SIGNIN: row.status = const.FINDING_STATUS_RESOLVED row.resolved_at = now else: row.status = const.FINDING_STATUS_ACKNOWLEDGED row.resolved_at = None row.accepted_until = None row.auto_resolved = False elif decision == const.FINDING_DECISION_ACCEPT_RISK: row.status = const.FINDING_STATUS_ACCEPTED_RISK # Auto-expiry is honored only for native findings; mirrored # (exposure / sign-in) source rows carry no expiry, so accepting # their risk is indefinite rather than a silently-ignored timer. row.accepted_until = ( now + timedelta(days=accepted_days) if accepted_days and row.source_table is None else None ) row.resolved_at = None row.auto_resolved = False elif decision == const.FINDING_DECISION_RESOLVE: row.status = const.FINDING_STATUS_RESOLVED row.resolved_at = now row.auto_resolved = False row.accepted_until = None else: # reopen row.status = const.FINDING_STATUS_OPEN row.resolved_at = None row.auto_resolved = False row.accepted_until = None row.decided_by_person_id = actor_user_id row.decided_at = now row.decision_note = note await _mirror_decision_to_source(db, row, decision, note, actor_user_id, now) await AuditService(db).log( trace_id=trace_id or uuid.uuid4(), actor_type="user", actor_user_id=actor_user_id, action=f"auditor.finding.{decision}", resource_type="auditor_findings", resource_id=str(row.id), status="success", detail={ "control_key": row.control_key, "finding_type": row.finding_type, "severity": row.severity, "subject_ref": row.subject_ref, "note": note, }, ) return row async def _mirror_decision_to_source( db: AsyncSession, row: AuditorFinding, decision: str, note: str, actor_user_id: uuid.UUID, now: datetime, ) -> None: """Reflect a finding decision back onto its legacy source row so the per-control list pages stay consistent until they migrate (PR-C).""" if row.source_table is None or row.source_id is None: return if row.source_table == SOURCE_TABLE_EXPOSURE: ex = await db.get(AuditorPublicExposureFinding, row.source_id) if ex is None: return if decision == const.FINDING_DECISION_RESOLVE: ex.status = const.EXPOSURE_STATUS_RESOLVED ex.resolved_at = now ex.resolved_by_person_id = actor_user_id ex.resolution_note = note elif decision == const.FINDING_DECISION_ACCEPT_RISK: ex.status = const.EXPOSURE_STATUS_ACCEPTED_RISK ex.resolved_at = now ex.resolved_by_person_id = actor_user_id ex.resolution_note = note else: # reopen ex.status = const.EXPOSURE_STATUS_OPEN ex.resolved_at = None ex.resolved_by_person_id = None ex.resolution_note = None elif row.source_table == SOURCE_TABLE_SIGNIN: ev = await db.get(AuditorSigninEvent, row.source_id) if ev is None: return if decision == const.FINDING_DECISION_REOPEN: ev.acknowledged_at = None ev.acknowledged_by_person_id = None ev.acknowledged_note = None else: # acknowledge / resolve both ack the underlying event ev.acknowledged_at = now ev.acknowledged_by_person_id = actor_user_id ev.acknowledged_note = note # ════════════════════════════════════════════════════════════════════ # Read queries # ════════════════════════════════════════════════════════════════════ def _severity_order(): """ORDER BY expression: critical → low, then most-recent first.""" return func.field( AuditorFinding.severity, const.FINDING_SEVERITY_CRITICAL, const.FINDING_SEVERITY_HIGH, const.FINDING_SEVERITY_MEDIUM, const.FINDING_SEVERITY_LOW, ) async def list_findings( db: AsyncSession, *, status: str | None = None, severity: str | None = None, control_key: str | None = None, open_only: bool = False, page: int = 1, page_size: int = 50, ) -> dict[str, Any]: q = select(AuditorFinding) cq = select(func.count()).select_from(AuditorFinding) def _filt(stmt): if status: stmt = stmt.where(AuditorFinding.status == status) elif open_only: stmt = stmt.where( AuditorFinding.status.in_(const.FINDING_OPEN_STATUSES) ) if severity: stmt = stmt.where(AuditorFinding.severity == severity) if control_key: stmt = stmt.where(AuditorFinding.control_key == control_key) return stmt q = _filt(q).order_by( _severity_order(), AuditorFinding.last_seen_at.desc(), ).offset((page - 1) * page_size).limit(page_size) cq = _filt(cq) rows = list((await db.execute(q)).scalars().all()) total = int((await db.execute(cq)).scalar_one()) person_ids = [r.person_id for r in rows if r.person_id] persons: dict[uuid.UUID, IdentityPerson] = {} if person_ids: persons = { p.id: p for p in (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(person_ids)) )).scalars().all() } items = [_finding_view(r, persons) for r in rows] return {"items": items, "total": total} async def view_finding(db: AsyncSession, r: AuditorFinding) -> dict[str, Any]: """Serialize a single finding (with person display) for an API reply.""" persons: dict[uuid.UUID, IdentityPerson] = {} if r.person_id: p = await db.get(IdentityPerson, r.person_id) if p is not None: persons[p.id] = p return _finding_view(r, persons) def _finding_view( r: AuditorFinding, persons: dict[uuid.UUID, IdentityPerson], ) -> dict[str, Any]: person = persons.get(r.person_id) if r.person_id else None return { "id": str(r.id), "control_key": r.control_key, "finding_type": r.finding_type, "severity": r.severity, "status": r.status, "title": r.title, "subject_type": r.subject_type, "subject_ref": r.subject_ref, "subject_display": r.subject_display, "person_id": str(r.person_id) if r.person_id else None, "person_full_name": person.display_name if person else None, "person_email": person.work_email if person else None, # 工号 + 在职状态:让前端 PersonTag 把离职人员标灰显示。 "person_employee_no": person.employee_no if person else None, "person_employment_status": person.employment_status if person else None, "detail": r.detail, "source_table": r.source_table, "first_seen_at": _iso(r.first_seen_at), "last_seen_at": _iso(r.last_seen_at), "decided_at": _iso(r.decided_at), "decision_note": r.decision_note, "accepted_until": _iso(r.accepted_until), } def _iso(dt: datetime | None) -> str | None: n = _to_naive(dt) return n.isoformat() if n else None async def findings_summary(db: AsyncSession) -> dict[str, Any]: """Worklist headline: needs-attention counts by severity and control.""" by_sev = (await db.execute( select(AuditorFinding.severity, func.count()).where( AuditorFinding.status.in_(const.FINDING_OPEN_STATUSES) ).group_by(AuditorFinding.severity) )).all() by_ctrl = (await db.execute( select(AuditorFinding.control_key, func.count()).where( AuditorFinding.status.in_(const.FINDING_OPEN_STATUSES) ).group_by(AuditorFinding.control_key) )).all() by_status = (await db.execute( select(AuditorFinding.status, func.count()) .group_by(AuditorFinding.status) )).all() sev_map = {s: int(n) for s, n in by_sev} return { "open_total": sum(sev_map.values()), "open_by_severity": sev_map, "open_by_control": {k: int(n) for k, n in by_ctrl}, "by_status": {s: int(n) for s, n in by_status}, } # ════════════════════════════════════════════════════════════════════ # Projection of mirrored controls (full re-sync) # ════════════════════════════════════════════════════════════════════ EXPOSURE_CONTROL_KEY = "public_exposure_scan" SIGNIN_CONTROL_KEY = "signin_anomaly_monitoring" _EXPOSURE_STATUS_TO_FINDING = { const.EXPOSURE_STATUS_OPEN: const.FINDING_STATUS_OPEN, const.EXPOSURE_STATUS_ACCEPTED_RISK: const.FINDING_STATUS_ACCEPTED_RISK, const.EXPOSURE_STATUS_RESOLVED: const.FINDING_STATUS_RESOLVED, } def _exposure_spec(r: AuditorPublicExposureFinding) -> FindingSpec: name = r.resource_name or r.resource_id return FindingSpec( dedup_key=str(r.id), severity=r.severity, title=f"{r.exposure_type} — {name}", subject_type=const.FINDING_SUBJECT_RESOURCE, subject_ref=r.resource_arn or r.resource_id, subject_display=name, detail={ "source": r.source, "region": r.region, "exposure_type": r.exposure_type, **(r.exposure_detail or {}), }, source_table=SOURCE_TABLE_EXPOSURE, source_id=r.id, status=_EXPOSURE_STATUS_TO_FINDING.get( r.status, const.FINDING_STATUS_OPEN), ) def _signin_spec(r: AuditorSigninEvent) -> FindingSpec: who = r.provider_email or "unknown principal" reasons = ", ".join(r.anomaly_reasons or []) or "anomalous sign-in" return FindingSpec( dedup_key=str(r.id), severity=( const.FINDING_SEVERITY_CRITICAL if r.anomaly_level == const.ANOMALY_LEVEL_CRITICAL else const.FINDING_SEVERITY_HIGH ), title=f"{r.provider}: {who} — {reasons}", subject_type=const.FINDING_SUBJECT_PERSON, subject_ref=r.provider_email, subject_display=r.provider_email, person_id=r.person_id, detail={ "provider": r.provider, "event_at": _iso(r.event_at), "source_ip": r.source_ip, "source_country": r.source_country, "anomaly_score": r.anomaly_score, "anomaly_reasons": r.anomaly_reasons, }, source_table=SOURCE_TABLE_SIGNIN, source_id=r.id, status=( const.FINDING_STATUS_RESOLVED if r.acknowledged_at is not None else const.FINDING_STATUS_OPEN ), ) async def project_exposure_findings(db: AsyncSession) -> dict[str, int]: """Mirror current public-exposure rows into findings.""" rows = (await db.execute( select(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.status.in_( [const.EXPOSURE_STATUS_OPEN, const.EXPOSURE_STATUS_ACCEPTED_RISK] ) ) )).scalars().all() return await reconcile_findings( db, control_key=EXPOSURE_CONTROL_KEY, finding_type=const.FINDING_TYPE_PUBLIC_EXPOSURE, specs=[_exposure_spec(r) for r in rows], source_managed=True, ) async def project_signin_findings( db: AsyncSession, *, lookback_hours: int = 24 * 30, ) -> dict[str, int]: """Mirror high/critical unacknowledged sign-in anomalies into findings. Only anomalies at/above 'high' become findings (the worklist is for things that need a decision); acknowledging the event resolves the finding. Bounded by ``lookback_hours`` so the worklist doesn't carry ancient events forever. """ cutoff = _naive_now() - timedelta(hours=lookback_hours) rows = (await db.execute( select(AuditorSigninEvent).where( AuditorSigninEvent.anomaly_level.in_( [const.ANOMALY_LEVEL_HIGH, const.ANOMALY_LEVEL_CRITICAL] ), AuditorSigninEvent.event_at >= cutoff, AuditorSigninEvent.acknowledged_at.is_(None), ) )).scalars().all() return await reconcile_findings( db, control_key=SIGNIN_CONTROL_KEY, finding_type=const.FINDING_TYPE_SIGNIN_ANOMALY, specs=[_signin_spec(r) for r in rows], source_managed=True, ) async def mirror_exposure_row( db: AsyncSession, r: AuditorPublicExposureFinding, ) -> None: """Keep the worklist finding in lock-step after a legacy exposure decision (called by ``ExposureService.resolve``).""" await sync_mirror_finding( db, control_key=EXPOSURE_CONTROL_KEY, finding_type=const.FINDING_TYPE_PUBLIC_EXPOSURE, spec=_exposure_spec(r), ) async def mirror_signin_row(db: AsyncSession, r: AuditorSigninEvent) -> None: """Keep the worklist finding in lock-step after a legacy sign-in acknowledgement (called by ``SigninAcknowledgeService.acknowledge``).""" await sync_mirror_finding( db, control_key=SIGNIN_CONTROL_KEY, finding_type=const.FINDING_TYPE_SIGNIN_ANOMALY, spec=_signin_spec(r), ) # ════════════════════════════════════════════════════════════════════ # Native producers (finding is the decision home — PR-B) # ════════════════════════════════════════════════════════════════════ # # The 4 controls whose raw table had no per-item decision store. The # finding owns the lifecycle; `reconcile_findings(source_managed=False)` # preserves operator decisions across re-syncs and re-opens expired # accept_risk. Each is called from its sync service after the raw upsert. MFA_CONTROL_KEY = "mfa_coverage" SECRET_CONTROL_KEY = "secret_rotation" CERT_CONTROL_KEY = "cert_expiry_monitoring" PAT_CONTROL_KEY = "pat_audit" _CERT_ALERT_SEVERITY = { const.CERT_ALERT_EXPIRED: const.FINDING_SEVERITY_CRITICAL, const.CERT_ALERT_30D: const.FINDING_SEVERITY_HIGH, const.CERT_ALERT_60D: const.FINDING_SEVERITY_MEDIUM, } async def reconcile_mfa_findings(db: AsyncSession) -> dict[str, int]: """One finding per active workforce member missing MFA on a provider. Posture for `mfa_coverage` stays coverage-%-based (it has a 95% threshold, not a zero-tolerance rule); these findings are the per-person worklist items, scoped to active employees so leavers pending deprovision don't generate noise. """ rows = (await db.execute( select(AuditorMfaStatus, IdentityPerson) .join(IdentityPerson, AuditorMfaStatus.person_id == IdentityPerson.id) .where( AuditorMfaStatus.is_enrolled.is_(False), # ``is_enrolled=False`` is only an ACCUSATION when the provider # actually said so. Atlassian's org-policy read has been # unavailable in production since day one, and treating that # unknown as "not enrolled" filled this worklist with 412 # items no one could act on — the auditor cannot enrol a user # in an MFA policy Helix cannot even see. AuditorMfaStatus.enrollment_observed.is_(True), IdentityPerson.employment_status == "active", # Share ONE population with the MFA-coverage dashboard # (DashboardService ``_mfa_active``), which also excludes # soft-deleted persons. Without this, an active-but-soft- # deleted person inflates the worklist while being absent # from the posture %, so the two numbers disagree. IdentityPerson.deleted_at.is_(None), ) )).all() specs = [ FindingSpec( dedup_key=make_dedup_key(m.person_id, m.provider), severity=( const.FINDING_SEVERITY_HIGH if m.is_enforced else const.FINDING_SEVERITY_MEDIUM ), title=f"MFA not enrolled — {m.provider} · {p.display_name}", subject_type=const.FINDING_SUBJECT_PERSON, subject_ref=m.provider_email or p.work_email, subject_display=p.display_name, person_id=p.id, detail={"provider": m.provider, "is_enforced": bool(m.is_enforced)}, ) for m, p in rows ] return await reconcile_findings( db, control_key=MFA_CONTROL_KEY, finding_type=const.FINDING_TYPE_MFA_NOT_ENROLLED, specs=specs, ) # ── Ledger staleness ───────────────────────────────────────────────── # # ``auditor_secret_rotation_records`` / ``auditor_pat_audit`` / # ``auditor_cert_expiry`` are upsert-only: nothing ever deletes or marks # a row whose upstream object has disappeared, so it FREEZES in its last # observed state (#2137). Three concrete ways that produces a finding # that can never be resolved: # # * a user rotates a key CORRECTLY — the old key was overdue, then it # is revoked, and the secret sync stops pulling it (it filters out # revoked keys). The row stays ``overdue`` for ever, so the HIGH # finding stays in ``specs`` for ever, and auto-resolve only fires # for findings that LEAVE specs; # * a GitHub PAT is deleted upstream — the row keeps its ``dormant`` / # ``no_expiry`` flags for ever; # * an ACM certificate is deleted — frozen as ``expired``. # # The exposure ledger in the same domain solved this with # ``_auto_resolve_unseen``; these three were never given an equivalent. # The staleness predicate is that equivalent, and it needs no migration: # every one of these tables already stamps ``last_synced_at`` on every # upsert, so "not re-stamped lately" IS "no longer observed upstream". # # The grace window is generous relative to the daily cadence of these # syncs. # # #2217 — the original safety argument was WRONG. It read: "safe because # a sync that stops running now FAILS LOUDLY (#2136)". It does not, for # the only failure mode that matters here: # # * ``_raise_if_systemic_failure`` raises only when there are errors # AND zero progress — a deliberate design choice ("a partial failure # is a real and normal state for a multi-provider sync"). All three # of these syncs are multi-provider with per-provider try/except, so # one dead provider (an expired Aliyun AK) is a quiet, ok-status run # with ``last_run_at`` advancing and the CCM freshness gate green; # * worse, the zero-progress branch is STRUCTURALLY unreachable for # two of the three: SecretRotation and PatAudit both start with the # internal ``HELIX_APIKEY`` segment (a pure DB query that ~cannot # fail) and CertExpiry unconditionally re-stamps manual rows, so # ``progress`` is always > 0 even in a total outage. # # Wall-clock staleness alone therefore auto-resolves every HIGH finding # owned by a provider that has been failing for 3 days, and announces it # as "resolved" on the compliance surface — exactly the failure #2123 # closed one layer up. The sweep is now gated PER SOURCE on the sources # the current round actually covered (``scanned_sources``), mirroring # what the exposure scanner does with ``_auto_resolve_unseen``: a source # that did not report this round keeps all of its findings, however old # its rows are. LEDGER_STALE_AFTER = timedelta(days=3) def _fresh(column: Any) -> Any: """Only rows re-observed by a recent sync describe reality.""" cutoff = datetime.now(UTC) - LEDGER_STALE_AFTER return or_(column.is_(None), column >= cutoff) def _fresh_for_scanned( ts_column: Any, source_column: Any, scanned_sources: Sequence[str] | None, ) -> Any: """Staleness sweep, but only for sources this round actually covered. ``scanned_sources=None`` means the caller cannot tell us (a manual reconcile, a test) — fall back to the plain wall-clock predicate so behaviour is unchanged for those. An EMPTY list means the round covered nothing, so nothing may be swept. """ if scanned_sources is None: return _fresh(ts_column) scanned = [s for s in scanned_sources if s] if not scanned: return true() return or_(source_column.not_in(scanned), _fresh(ts_column)) async def reconcile_secret_findings( db: AsyncSession, *, scanned_sources: Sequence[str] | None = None, ) -> dict[str, int]: """One finding per secret past its rotation due date. ``scanned_sources`` = the sources the calling sync round covered; the staleness sweep applies to those only (#2217). """ rows = (await db.execute( select(AuditorSecretRotationRecord).where( AuditorSecretRotationRecord.status == const.SECRET_STATUS_OVERDUE, _fresh_for_scanned( AuditorSecretRotationRecord.last_synced_at, AuditorSecretRotationRecord.source, scanned_sources, ), ) )).scalars().all() specs = [ FindingSpec( dedup_key=make_dedup_key(r.source, r.secret_kind, r.resource_id), severity=const.FINDING_SEVERITY_HIGH, title=f"Secret overdue for rotation — {r.resource_name or r.resource_id}", subject_type=const.FINDING_SUBJECT_SECRET, subject_ref=r.resource_id, subject_display=r.resource_name or r.resource_id, person_id=r.owner_person_id, detail={ "secret_kind": r.secret_kind, "source": r.source, "owner_email": r.owner_email, "next_rotation_due_at": _iso(r.next_rotation_due_at), "days_until_due": r.days_until_due, }, ) for r in rows ] return await reconcile_findings( db, control_key=SECRET_CONTROL_KEY, finding_type=const.FINDING_TYPE_SECRET_OVERDUE, specs=specs, ) async def reconcile_cert_findings( db: AsyncSession, *, scanned_sources: Sequence[str] | None = None, ) -> dict[str, int]: """One finding per certificate/domain/ICP expired or expiring ≤60d.""" rows = (await db.execute( select(AuditorCertExpiry).where( AuditorCertExpiry.alert_level.in_(list(_CERT_ALERT_SEVERITY)), _fresh_for_scanned( AuditorCertExpiry.last_synced_at, AuditorCertExpiry.source, scanned_sources, ), ) )).scalars().all() specs = [ FindingSpec( dedup_key=make_dedup_key(r.source, r.asset_kind, r.external_id), severity=_CERT_ALERT_SEVERITY[r.alert_level], title=( f"{r.asset_kind} expired — {r.name}" if r.alert_level == const.CERT_ALERT_EXPIRED else f"{r.asset_kind} expiring ({r.alert_level}) — {r.name}" ), subject_type=const.FINDING_SUBJECT_CERT, subject_ref=r.external_id, subject_display=r.name, person_id=r.owner_person_id, detail={ "asset_kind": r.asset_kind, "source": r.source, "owner_email": r.owner_email, "expires_at": _iso(r.expires_at), "alert_level": r.alert_level, "days_until_expiry": r.days_until_expiry, }, ) for r in rows ] return await reconcile_findings( db, control_key=CERT_CONTROL_KEY, finding_type=const.FINDING_TYPE_CERT_EXPIRING, specs=specs, ) _PAT_HIGH_FLAGS = frozenset({ const.PAT_RISK_OWNER_LEFT, const.PAT_RISK_BROAD_SCOPE_ADMIN, }) async def reconcile_pat_findings( db: AsyncSession, *, scanned_sources: Sequence[str] | None = None, ) -> dict[str, int]: """One finding per token carrying a risk flag.""" rows = (await db.execute( select(AuditorPatAudit).where( AuditorPatAudit.risk_flags.is_not(None), _fresh_for_scanned( AuditorPatAudit.last_synced_at, AuditorPatAudit.source, scanned_sources, ), ) )).scalars().all() specs = [] for r in rows: flags = r.risk_flags or [] if not flags: continue sev = ( const.FINDING_SEVERITY_HIGH if _PAT_HIGH_FLAGS.intersection(flags) else const.FINDING_SEVERITY_MEDIUM ) specs.append(FindingSpec( dedup_key=make_dedup_key(r.source, r.kind, r.token_external_id), severity=sev, title=f"Risky token ({', '.join(flags)}) — {r.token_name or r.token_external_id}", subject_type=const.FINDING_SUBJECT_TOKEN, subject_ref=r.token_external_id, subject_display=r.token_name or r.token_external_id, person_id=r.person_id, detail={ "kind": r.kind, "source": r.source, "owner_email": r.owner_email, "risk_flags": flags, "last_used_at": _iso(r.last_used_at), }, )) return await reconcile_findings( db, control_key=PAT_CONTROL_KEY, finding_type=const.FINDING_TYPE_PAT_RISK, specs=specs, ) async def count_open_by_control(db: AsyncSession) -> dict[str, int]: """{control_key: needs-attention count} (open + acknowledged). Consumed by the posture evaluator so a binary control's pass/fail derives from the unified findings — and therefore honors operator decisions (accepted_risk / resolved drop out of the count).""" rows = (await db.execute( select(AuditorFinding.control_key, func.count()).where( AuditorFinding.status.in_(const.FINDING_OPEN_STATUSES) ).group_by(AuditorFinding.control_key) )).all() return {k: int(n) for k, n in rows}