"""Access-drift detectors — the evidence queries, run as controls. The same deterministic queries that answer an auditor's questions also detect the conditions those questions exist to find. Running them on a schedule turns "we will discover this next audit season" into "a ticket appeared the week it happened". Four detectors, each for a gap the platform can currently only surface by someone going looking: * ``leaver_access_not_revoked`` — a departed person still holding access. * ``offboarding_without_evidence`` — a completed offboarding with no frozen archive, i.e. the archival path silently failed. * ``multiple_open_holding_cycles`` — two open intervals for the same key in the point-in-time ledgers, which means a transition was double-recorded and every roster crossing that window is inflated. * ``privileged_role_without_grant_record`` — a platform role held with no ledger interval at all, so no point-in-time query can see it. **Observe-only by default.** The detectors compute and report; whether they open findings is a deployment decision, because turning a new detector loose on years of accumulated state can produce a first-day ticket storm that trains people to ignore the queue. The job returns its would-be finding count in observe mode so the real number can be measured against production before anything is opened. Notification fan-out, when enabled, reaches auditors and global admins only — never employees. A drift ticket is IT's work, and the platform's standing rule is that new pushes do not reach employees. """ from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.rbac import ( CANONICAL_ROLE_NAMES, EMPLOYEE_ROLE, LEAVER_REVOCABLE_ROLES, ) from app.domains.auditor.findings_service import FindingSpec, make_dedup_key CONTROL_KEY = "access_drift" # Every canonical role except ``employee``, which is self-scope only and # therefore confers no privilege to lose track of. Derived from the RBAC # canon rather than hardcoded so a tenth role cannot be added to the # platform and silently escape this control. Legacy aliases are included # because ``auth_roles.name`` still stores them. PRIVILEGED_ROLE_NAMES: frozenset[str] = frozenset( set(CANONICAL_ROLE_NAMES) - {EMPLOYEE_ROLE} | {"it_admin", "security_auditor"} ) # What must actually be revoked when someone departs. ``LEAVER_REVOCABLE_ROLES`` # is the platform's existing answer (see rbac.py for why ``employee`` and # ``asset_admin`` are excluded: the former is self-scope only and harmless on a # disabled account, the latter is reconciler-derived from office-admin rows and # would just be re-added). Reusing it keeps one definition of the question # instead of a second, drifting copy. LEAVER_HOLDING_ROLE_NAMES: frozenset[str] = frozenset( set(LEAVER_REVOCABLE_ROLES) | {"it_admin", "security_auditor"} ) FINDING_LEAVER_ACCESS = "leaver_access_not_revoked" FINDING_MISSING_EVIDENCE = "offboarding_without_evidence" FINDING_DOUBLE_CYCLE = "multiple_open_holding_cycles" FINDING_ROLE_WITHOUT_LEDGER = "privileged_role_without_grant_record" # A leaver's access is not drift the instant they leave — provisioning # and the offboarding queue need a window to run. Beyond it, "still # holding" is a finding rather than work in progress. LEAVER_GRACE_DAYS = 7 # Both statuses count. ``LeaverBlockGuard`` REFUSES the flip to ``left`` # while any offboarding blocker is unresolved — and the blockers ARE the # un-revoked seats, permissions and SSO grants. Scoping to ``left`` alone # therefore excluded, by construction, exactly the population this # detector exists to find: a departed person still holding access is # parked at ``leaver_pending`` and was invisible. LEAVER_STATUSES = ("left", "leaver_pending") # A control cannot judge what happened before it existed. Both keys are # seeded by migration 511 at the instant the paths they observe went live. SETTING_ROLE_LEDGER_EPOCH = "access_drift.role_ledger_epoch" SETTING_OFFBOARDING_ARCHIVE_EPOCH = "access_drift.offboarding_archive_epoch" async def _control_epoch(db: AsyncSession, key: str) -> datetime | None: """When this control became measurable, or ``None`` if unknown. ``None`` means judge NOTHING. That is the whole point: the previous behaviour was to judge everything, which on production data produced 617 findings for role grants that predate the ledger and 70 for offboardings that predate the archive — every one of them false by construction. A detector that cannot establish its own start date has no basis for calling anything drift. """ from app.integrations.models import PlatformSetting # noqa: PLC0415 raw = (await db.execute( select(PlatformSetting.value).where(PlatformSetting.key == key) )).scalar_one_or_none() if raw is None: return None text = raw if isinstance(raw, str) else str(raw) try: parsed = datetime.fromisoformat(text) except (TypeError, ValueError): return None # MySQL DATETIME carries no offset, so a stored value reads back # naive. Compare in UTC rather than raising on a naive/aware mix. return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) async def detect_leaver_access( db: AsyncSession, *, grace_days: int = LEAVER_GRACE_DAYS, ) -> list[FindingSpec]: """Departed people who still hold access past the grace window.""" from app.auth.models import AuthRole, AuthUserRole # noqa: PLC0415 from app.domains.asset.models import SoftwareSeatAllocation # noqa: PLC0415 from app.domains.cmdb.models import CmdbCloudAccessGrant # noqa: PLC0415 from app.domains.permission.models import UserPermissionCycle # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 cutoff = (datetime.now(UTC) - timedelta(days=grace_days)).date() leavers = (await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.termination_date, ).where( IdentityPerson.employment_status.in_(LEAVER_STATUSES), IdentityPerson.termination_date.is_not(None), IdentityPerson.termination_date <= cutoff, IdentityPerson.deleted_at.is_(None), ) )).all() if not leavers: return [] by_id = {str(row[0]): row for row in leavers} ids = {row[0] for row in leavers} specs: list[FindingSpec] = [] async def _emit(kind: str, person_key: str, label: str) -> None: row = by_id.get(person_key) if row is None: return person_id, display_name, work_email, terminated = row specs.append(FindingSpec( dedup_key=make_dedup_key(FINDING_LEAVER_ACCESS, person_key, kind, label), severity="high", title=( f"Leaver {display_name or work_email} still holds {kind}: " f"{label}" ), subject_type="person", subject_ref=person_key, subject_display=display_name or work_email, person_id=person_id, detail={ "access_kind": kind, "access_label": label, "termination_date": ( terminated.isoformat() if terminated else None ), "grace_days": grace_days, }, )) seats = (await db.execute( select(SoftwareSeatAllocation.user_id, SoftwareSeatAllocation.plan_id) .where( SoftwareSeatAllocation.user_id.in_(ids), SoftwareSeatAllocation.revoked_at.is_(None), SoftwareSeatAllocation.status == "active", ) )).all() for user_id, plan_id in seats: await _emit("software seat", str(user_id), f"plan {plan_id}") grants = (await db.execute( select( CmdbCloudAccessGrant.person_id, CmdbCloudAccessGrant.scope_id, CmdbCloudAccessGrant.role_key, ).where( CmdbCloudAccessGrant.person_id.in_({str(i) for i in ids}), CmdbCloudAccessGrant.revoked_at.is_(None), CmdbCloudAccessGrant.status == "active", ) )).all() for person_id, scope_id, role_key in grants: await _emit("cloud grant", str(person_id), f"{role_key} on {scope_id}") perms = (await db.execute( select(UserPermissionCycle.user_id, UserPermissionCycle.permission_id) .where( UserPermissionCycle.user_id.in_(ids), UserPermissionCycle.granted_at.is_not(None), UserPermissionCycle.closed_at.is_(None), ) )).all() for user_id, permission_id in perms: await _emit( "software permission", str(user_id), f"permission {permission_id}", ) # ``auth_user_roles``, NOT ``auth_user_role_cycles``. The ledger answers # "did they hold it on date D"; this control asks "do they hold it NOW", # and the live table is the only authority on that. Reading the ledger # made the control structurally silent: 484 deliberately does not # backfill, so on production every holding predates it and this query # matched nothing — 120 departed people held a role the detector could # not see. All 120 are plain ``employee`` (harmless, and excluded below), # so nothing serious was actually missed, but the day somebody leaves # holding ``global_admin`` this control has to fire. # # No epoch gate here, unlike the other two detectors: a question about # the present needs no start date. roles = (await db.execute( select(AuthUserRole.user_id, AuthRole.name) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( AuthUserRole.user_id.in_(ids), AuthRole.name.in_(LEAVER_HOLDING_ROLE_NAMES), ) )).all() for user_id, role_name in roles: await _emit("platform role", str(user_id), role_name) return specs async def detect_missing_offboarding_evidence( db: AsyncSession, *, lookback_days: int = 90, ) -> list[FindingSpec]: """Completed offboardings with no frozen evidence archive. The archival path deliberately swallows its own failures so a broken evidence query can never block a real offboarding. This is the other half of that bargain: the gap has to become visible somewhere, or "never blocks" quietly becomes "never happens". """ from app.core.coordinators.offboarding_evidence_coordinator import ( # noqa: PLC0415 ARCHIVED_QUERIES, GENERATED_VIA_OFFBOARDING, ) from app.domains.auditor.models import AuditorEvidenceBundle # noqa: PLC0415 from app.domains.offboarding.models import OffboardingProcess # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 epoch = await _control_epoch(db, SETTING_OFFBOARDING_ARCHIVE_EPOCH) if epoch is None: # Not measurable yet. Deliberately NOT derived from "are there any # bundles" — zero bundles is precisely the broken archive this # detector exists to catch, so a data-derived epoch would silence # it forever. return [] since = max(datetime.now(UTC) - timedelta(days=lookback_days), epoch) completed = (await db.execute( select( OffboardingProcess.id, OffboardingProcess.person_id, OffboardingProcess.completed_at, OffboardingProcess.leaver_date, ).where( OffboardingProcess.status == "completed", OffboardingProcess.completed_at.is_not(None), OffboardingProcess.completed_at >= since, ) )).all() if not completed: return [] archived_params = (await db.execute( select( AuditorEvidenceBundle.params, AuditorEvidenceBundle.query_key, ) .where( AuditorEvidenceBundle.generated_via == GENERATED_VIA_OFFBOARDING, ) )).all() # Keyed on (process, cycle), matching what the archive stamps. # Neither half identifies a departure alone: keying on the person # made a rehire's second exit look archived because the first was, # and keying on the process does the same, because ``person_id`` is # UNIQUE on ``offboarding_processes`` and a rehire therefore reopens # the existing row instead of inserting a second one. # # No fallback for older key shapes. Two review rounds went into a # person-keyed fallback meant to prevent a "ticket storm on the day # the key changes" — and there is no such day: this feature has # never been on main, so production holds zero bundles with # ``generated_via='offboarding'``. The fallback protected against a # migration that never happens, could not be exercised by any real # row, and its first version reopened the very blind spot the key # exists to close. Deleted rather than kept as an unreachable # branch with a test locking it in place. archived: dict[tuple[str, str | None], set[str]] = {} for params, query_key in archived_params: payload = params or {} process_key = payload.get("offboarding_process_id") if process_key: archived.setdefault( (str(process_key), payload.get("offboarding_cycle")), set(), ).add(query_key) people = { str(pid): (name, email) for pid, name, email in (await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where( IdentityPerson.id.in_({row[1] for row in completed}), ) )).all() } specs: list[FindingSpec] = [] for process_id, person_id, completed_at, leaver_date in completed: cycle = leaver_date.isoformat() if leaver_date else None have = archived.get((str(process_id), cycle), set()) missing = sorted(set(ARCHIVED_QUERIES) - have) if not missing: continue name, email = people.get(str(person_id), (None, None)) specs.append(FindingSpec( # The cycle belongs in the dedup key for the same reason it # belongs in the archive key. ``RESOLVED`` findings reopen on # recurrence, but an ``ACCEPTED_RISK`` one with no expiry # does not — so an auditor who accepted the risk on a 2024 # departure would silently suppress the identical finding for # that person's 2026 departure. The blind spot would have # moved from the archive layer to the ticket layer. dedup_key=make_dedup_key( FINDING_MISSING_EVIDENCE, str(process_id), cycle or "-", ), severity="medium", title=( f"Offboarding completed with no evidence archive: " f"{name or email or person_id}" ), subject_type="offboarding_process", subject_ref=str(process_id), subject_display=name or email, person_id=person_id, detail={ "offboarding_cycle": cycle, "missing_queries": missing, "completed_at": ( completed_at.isoformat() if completed_at else None ), }, source_table="offboarding_processes", source_id=process_id, )) return specs async def detect_double_open_cycles(db: AsyncSession) -> list[FindingSpec]: """Two open holding intervals for one key. The ledgers' invariant is at most one open cycle per key, held by the listener closing before it opens. Deliberately not enforced with a UNIQUE constraint — an evidence ledger must never be able to fail the business transaction it observes — so it is enforced here instead, where a violation becomes a ticket rather than a 500 on someone's grant. """ from app.auth.models import AuthUserRoleCycle # noqa: PLC0415 from app.domains.permission.models import UserPermissionCycle # noqa: PLC0415 specs: list[FindingSpec] = [] from app.identity.person_models import IdentityPerson # noqa: PLC0415 role_dupes = (await db.execute( select( AuthUserRoleCycle.user_id, AuthUserRoleCycle.role_name, func.count().label("n"), ) .join(IdentityPerson, IdentityPerson.id == AuthUserRoleCycle.user_id) .where( AuthUserRoleCycle.revoked_at.is_(None), # Soft-deleted people would otherwise generate findings # nobody can act on. The platform's standing rule is that a # diagnostic query carries ``deleted_at IS NULL``. IdentityPerson.deleted_at.is_(None), ) .group_by(AuthUserRoleCycle.user_id, AuthUserRoleCycle.role_name) .having(func.count() > 1) )).all() for user_id, role_name, count in role_dupes: specs.append(FindingSpec( dedup_key=make_dedup_key( FINDING_DOUBLE_CYCLE, "role", str(user_id), role_name, ), severity="medium", title=( f"{count} open holding intervals for role {role_name} — " "point-in-time rosters over this window are inflated" ), subject_type="person", subject_ref=str(user_id), person_id=user_id, detail={"ledger": "auth_user_role_cycles", "open_cycles": count}, )) perm_dupes = (await db.execute( select( UserPermissionCycle.user_permission_id, UserPermissionCycle.user_id, func.count().label("n"), ) .join(IdentityPerson, IdentityPerson.id == UserPermissionCycle.user_id) .where( UserPermissionCycle.closed_at.is_(None), IdentityPerson.deleted_at.is_(None), ) .group_by( UserPermissionCycle.user_permission_id, UserPermissionCycle.user_id, ) .having(func.count() > 1) )).all() for row_id, user_id, count in perm_dupes: specs.append(FindingSpec( dedup_key=make_dedup_key( FINDING_DOUBLE_CYCLE, "permission", str(row_id), ), severity="medium", title=( f"{count} open holding intervals for one permission row — " "point-in-time rosters over this window are inflated" ), subject_type="person", subject_ref=str(user_id), person_id=user_id, detail={ "ledger": "user_permission_cycles", "user_permission_id": str(row_id), "open_cycles": count, }, )) return specs async def detect_roles_without_ledger(db: AsyncSession) -> list[FindingSpec]: """Platform roles held with no ledger interval at all. Either the grant predates migration 484 (expected, and harmless once understood) or something wrote ``auth_user_roles`` outside the ORM and the listener never saw it — in which case no point-in-time query can see the role either, which is the failure this whole feature exists to prevent. """ from app.auth.models import ( # noqa: PLC0415 AuthRole, AuthUserRole, AuthUserRoleCycle, ) from app.identity.person_models import IdentityPerson # noqa: PLC0415 epoch = await _control_epoch(db, SETTING_ROLE_LEDGER_EPOCH) if epoch is None: return [] held = (await db.execute( select( AuthUserRole.user_id, AuthUserRole.role_id, AuthRole.name, IdentityPerson.display_name, IdentityPerson.work_email, ) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .join(IdentityPerson, IdentityPerson.id == AuthUserRole.user_id) .where( IdentityPerson.deleted_at.is_(None), # The finding is called *privileged*_role_without_grant_record, # and the predicate did not filter on privilege at all: 566 of # the 617 it would have opened were somebody's plain # ``employee`` role, which confers self scope only. A control # whose name promises "privileged" and whose query means # "everyone" trains people to close the queue unread. AuthRole.name.in_(PRIVILEGED_ROLE_NAMES), # Grants that predate the ledger have no interval BY DESIGN # (migration 484 does not backfill). Only a grant made while # the listener was watching can be evidence that it missed one. AuthUserRole.granted_at.is_not(None), AuthUserRole.granted_at >= epoch, ) )).all() if not held: return [] open_cycles = { (str(user_id), role_name) for user_id, role_name in (await db.execute( select(AuthUserRoleCycle.user_id, AuthUserRoleCycle.role_name) .where(AuthUserRoleCycle.revoked_at.is_(None)) )).all() } specs: list[FindingSpec] = [] for user_id, _role_id, role_name, display_name, work_email in held: if (str(user_id), role_name) in open_cycles: continue specs.append(FindingSpec( dedup_key=make_dedup_key( FINDING_ROLE_WITHOUT_LEDGER, str(user_id), role_name, ), severity="low", title=( f"Role {role_name} held with no ledger interval — " "invisible to point-in-time queries" ), subject_type="person", subject_ref=str(user_id), subject_display=display_name or work_email, person_id=user_id, detail={ "role": role_name, "note": ( "Either the grant predates migration 484, or it was " "written outside the ORM and the access-cycle " "listener never observed it." ), }, )) return specs DETECTORS: tuple[tuple[str, Any], ...] = ( (FINDING_LEAVER_ACCESS, detect_leaver_access), (FINDING_MISSING_EVIDENCE, detect_missing_offboarding_evidence), (FINDING_DOUBLE_CYCLE, detect_double_open_cycles), (FINDING_ROLE_WITHOUT_LEDGER, detect_roles_without_ledger), ) async def run_access_drift( db: AsyncSession, *, observe_only: bool = True, grace_days: int = LEAVER_GRACE_DAYS, ) -> dict[str, Any]: """Run every detector; open findings only when not observing. Observe mode exists so the first-day volume can be measured against real production state before any ticket is created. A detector let loose on years of accumulated drift can produce a queue nobody reads, and a queue nobody reads is worse than no queue. """ from app.domains.auditor.findings_service import ( # noqa: PLC0415 reconcile_findings, ) summary: dict[str, Any] = { "observe_only": observe_only, "by_type": {}, "would_open_total": 0, } for finding_type, detector in DETECTORS: # Only the leaver detector has a grace window; threading it # through generically keeps the documented config knob honest — # it was previously advertised in the job description and # silently ignored, so an operator who widened the window still # got findings at seven days. specs = ( await detector(db, grace_days=grace_days) if detector is detect_leaver_access else await detector(db) ) entry: dict[str, Any] = {"detected": len(specs)} summary["would_open_total"] += len(specs) if not observe_only: entry["reconciled"] = await reconcile_findings( db, control_key=CONTROL_KEY, finding_type=finding_type, specs=specs, ) summary["by_type"][finding_type] = entry return summary __all__ = [ "CONTROL_KEY", "DETECTORS", "FINDING_DOUBLE_CYCLE", "FINDING_LEAVER_ACCESS", "FINDING_MISSING_EVIDENCE", "FINDING_ROLE_WITHOUT_LEDGER", "LEAVER_HOLDING_ROLE_NAMES", "PRIVILEGED_ROLE_NAMES", "SETTING_OFFBOARDING_ARCHIVE_EPOCH", "SETTING_ROLE_LEDGER_EPOCH", "detect_double_open_cycles", "detect_leaver_access", "detect_missing_offboarding_evidence", "detect_roles_without_ledger", "run_access_drift", ]