"""LeaverBlockGuard — the public service every code path must use to flip a person to ``employment_status='left'``. Why a guard instead of just letting callers write the column: Direct ``person.employment_status = 'left'`` writes are tolerated as the *backstop* (caught by the SQLAlchemy event listener in ``listener.py``), but the GUARD is the canonical entry point. Going through the guard: * Performs the blocker check inline so the caller sees a ``LeaverBlockedError`` immediately (with the list of pending blocker IDs) rather than a flush-time exception that requires transaction rollback. * Writes ``leaver_block_audit`` rows on REJECTION even when the caller wrapped in try/except — the listener fires too late for that. * Marks the in-flight write as authorized so the listener doesn't double-check (via ``GUARD_AUTHORIZED`` contextvar). * Threads dual-sign override IDs through cleanly: a caller executing an approved override passes ``override_id``, the guard validates the override covers exactly the pending blockers, and authorizes the flip. """ from __future__ import annotations import contextvars import logging import uuid from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import LeaverBlockedError, format_exception_for_log from app.core.txn import best_effort from app.domains.leaver_blocked.models import ( DENIED_PENDING_BLOCKERS, DENIED_PROCESS_CANCELLED, DENIED_PROCESS_NOT_FOUND, DUAL_SIGN_APPROVED, DUAL_SIGN_EXECUTED, LEAVER_VIA_ADMIN_API, LEAVER_VIA_CLI, LEAVER_VIA_DUAL_SIGN_OVERRIDE, LEAVER_VIA_HRIS_SYNC, LEAVER_VIA_PROCESS_ADVANCE, LeaverBlockAudit, LeaverDualSignOverride, ) logger = logging.getLogger(__name__) # ── Contextvars used by the SQLAlchemy listener ──────────────────── # # The guard sets these inside an ``authorized_flip()`` block while it # performs the actual ORM mutation. The listener reads them on # ``before_flush`` to decide whether to second-guess the write. # # We use contextvars (not module globals) so concurrent async tasks # don't pollute each other's authorization state. GUARD_AUTHORIZED: contextvars.ContextVar[bool] = contextvars.ContextVar( "leaver_block_guard_authorized", default=False, ) GUARD_AUTHORIZED_FOR_PERSON: contextvars.ContextVar[uuid.UUID | None] = ( contextvars.ContextVar("leaver_block_guard_authorized_for", default=None) ) GUARD_OVERRIDE_ID: contextvars.ContextVar[uuid.UUID | None] = ( contextvars.ContextVar("leaver_block_guard_override_id", default=None) ) GUARD_VIA: contextvars.ContextVar[str | None] = contextvars.ContextVar( "leaver_block_guard_via", default=None, ) GUARD_ACTOR: contextvars.ContextVar[uuid.UUID | None] = contextvars.ContextVar( "leaver_block_guard_actor", default=None, ) @dataclass class LeaverBlockResult: """Outcome of a guard check / flip. Always returned from ``assert_can_flip_to_left``; ``flip_to_left`` returns the same shape with ``flipped=True`` on success.""" allowed: bool pending_blocker_ids: list[str] pending_blockers_snapshot: list[dict[str, Any]] override_id: uuid.UUID | None audit_id: uuid.UUID | None flipped: bool = False class LeaverBlockGuard: """Hard-gate enforcer for ``employment_status='left'`` flips.""" def __init__(self, db: AsyncSession) -> None: self.db = db # ── Public API ────────────────────────────────────────────── async def assert_can_flip_to_left( self, person, *, via: str, trace_id: uuid.UUID | str, actor_user_id: uuid.UUID | None = None, client_ip: str | None = None, user_agent: str | None = None, override_id: uuid.UUID | str | None = None, ) -> LeaverBlockResult: """Raise ``LeaverBlockedError`` if a flip-to-left is forbidden. ``override_id``: when provided, it must reference an approved ``leaver_dual_sign_overrides`` row that covers all currently pending blockers for the person. The override flag is consumed on success — the override row gets its status updated to ``executed``. """ # Lazy-import the offboarding model to dodge the # leaver_blocked → offboarding → asset → ... cycle from app.domains.offboarding.models import ( PROC_CANCELLED, RESO_PENDING, OffboardingBlocker, OffboardingProcess, ) # Lock the process row for the duration of the check-then-flip # so a concurrent blocker insertion (e.g. a parallel rescan) # cannot sneak in between the pending-blocker read and the # actual employment_status write. Without FOR UPDATE the window # between lines ~188 and ~286 is a TOCTOU gap. proc = await self.db.scalar( select(OffboardingProcess) .where(OffboardingProcess.person_id == person.id) .with_for_update() ) # No process / cancelled process: there cannot be pending # blockers (no scanner has run for this person), so the # historical platform behaviour was to allow the flip and # rely on ``post_leaver_verify`` to catch any orphans T+1. # We preserve that behaviour but write a soft audit row so # operators can see the unusual path was taken. if proc is None or proc.status == PROC_CANCELLED: soft_reason = ( DENIED_PROCESS_NOT_FOUND if proc is None else DENIED_PROCESS_CANCELLED ) soft_detail = { "soft": True, "note": ( "no offboarding process — flip allowed; " "post_leaver_verify will catch orphans" if proc is None else "process cancelled — flip allowed; " "post_leaver_verify will catch orphans" ), } if proc is not None: soft_detail["process_id"] = str(proc.id) await self._write_soft_audit( person=person, via=via, denied_reason=soft_reason, detail=soft_detail, trace_id=_coerce_uuid(trace_id), actor_user_id=actor_user_id, client_ip=client_ip, user_agent=user_agent, ) return LeaverBlockResult( allowed=True, pending_blocker_ids=[], pending_blockers_snapshot=[], override_id=None, audit_id=None, flipped=False, ) # Load all pending blockers for the snapshot. # # ``with_for_update()`` is load-bearing, not cosmetic: under # MySQL/TiDB REPEATABLE READ a plain SELECT is a *snapshot read* # pinned to this transaction's first consistent read. The # ``with_for_update()`` on ``proc`` above is a *current read* # (sees latest committed), but a current read does NOT advance # the snapshot for subsequent plain SELECTs. So a flip-to-left # whose transaction snapshot predates a concurrent # ``start_process`` scan's blocker INSERT commit would lock the # freshly-committed process row yet still read ZERO pending # blockers from the stale snapshot → allow the flip → leave the # person ``left`` with orphaned pending blockers and a process # wedged at BLOCKED forever (prod incident 2026-06-01: lifu.wu / # chujie). Making this a locking/current read closes the TOCTOU # window: it sees the latest committed blockers and serializes # against the inserting transaction. pending = list(( await self.db.execute( select(OffboardingBlocker).where( OffboardingBlocker.process_id == proc.id, OffboardingBlocker.resolution == RESO_PENDING, ).with_for_update() ) ).scalars().all()) # Build snapshot used for both audit + override coverage check snapshot = [ { "id": str(b.id), "resource_type": b.resource_type, "resource_id": str(b.resource_id) if b.resource_id else None, "label": b.label, "software_id": str(b.software_id) if b.software_id else None, } for b in pending ] pending_ids = [str(b.id) for b in pending] # Override path if override_id is not None: override = await self._validate_override( override_id=_coerce_uuid(override_id), person_id=person.id, process_id=proc.id, pending_blocker_ids=pending_ids, ) return LeaverBlockResult( allowed=True, pending_blocker_ids=pending_ids, pending_blockers_snapshot=snapshot, override_id=override.id, audit_id=None, flipped=False, ) # No override — must have zero pending blockers to pass if pending: audit_id = await self._write_audit( person=person, via=via, target_status="left", denied_reason=DENIED_PENDING_BLOCKERS, pending_blockers=snapshot, detail={ "process_id": str(proc.id), "process_status": proc.status, }, trace_id=_coerce_uuid(trace_id), actor_user_id=actor_user_id, client_ip=client_ip, user_agent=user_agent, ) raise LeaverBlockedError( person_id=str(person.id), pending_blocker_ids=pending_ids, message=( f"Cannot transition {person.work_email} to 'left': " f"{len(pending_ids)} unresolved blocker(s). " f"Resolve via force-transfer / revoke / dual-sign " f"override before retrying." ), # The denial audit row we just wrote — the id was # previously computed and dropped, so a 409 gave support # no handle on its own forensic record. audit_id=str(audit_id) if audit_id else None, ) return LeaverBlockResult( allowed=True, pending_blocker_ids=[], pending_blockers_snapshot=[], override_id=None, audit_id=None, flipped=False, ) async def flip_to_left( self, person, *, via: str, trace_id: uuid.UUID | str, actor_user_id: uuid.UUID | None = None, client_ip: str | None = None, user_agent: str | None = None, override_id: uuid.UUID | str | None = None, last_synced_at: datetime | None = None, ) -> LeaverBlockResult: """Asserts + actually mutates ``person.employment_status = 'left'``. Wraps the write in an authorized contextvar block so the SQLAlchemy event listener at flush time recognizes this as an authorized write and does not re-check / re-audit. """ result = await self.assert_can_flip_to_left( person, via=via, trace_id=trace_id, actor_user_id=actor_user_id, client_ip=client_ip, user_agent=user_agent, override_id=override_id, ) with self._authorized( person_id=person.id, override_id=result.override_id, via=via, actor_user_id=actor_user_id, ): person.employment_status = "left" person.last_synced_at = last_synced_at or datetime.now(UTC) await self.db.flush() # Mark the override consumed (idempotent — already validated as approved) if result.override_id is not None: override = await self.db.get(LeaverDualSignOverride, result.override_id) if override is not None and override.status == DUAL_SIGN_APPROVED: override.status = DUAL_SIGN_EXECUTED override.executed_at = datetime.now(UTC) await self.db.flush() # Orphan-seat sweep — defense-in-depth for the soft-allow path # (no offboarding process / cancelled process). Without this, # an HRIS sync that flips a person to ``left`` directly leaves # any active SoftwareSeatAllocation rows unattended; daily # billing keeps charging until ``post_leaver_verify`` raises # an incident T+1, AND seat-sync handlers may re-import a # revoked vendor-side row before the incident is resolved. # For the rigorous path (process completed all blockers) the # seats are already revoked, so this is an idempotent no-op. # Round-2 audit P1: pass actor_user_id explicitly because # the contextvar is no longer set here (we exit `_authorized` # at line 288 above) — without this, the orphan-revoke audit # rows would record actor=None instead of the real caller. await self._revoke_orphan_seats( person, via=via, trace_id=trace_id, actor_user_id=actor_user_id, ) # Cloud-billing alert cc-list sweep — same hygiene story as the # orphan-seat sweep above, applied to # ``cmdb_cloud_scopes.alert_rules[*].cc_person_ids``. Without # this, the leaver's UUID lingers in cc lists indefinitely, # rendering as "未知用户 · <8-char-uuid>" in the admin alert # panel because ``/scopes/{id}/pickable-people`` filters out # ``status='left'`` and the frontend can't resolve the id to # a name. Same root-cause class as the AWS Identity Center # "User not found" residue: the platform mutated the # principal's lifecycle but didn't sweep the references that # pointed at it. Failure is logged + swallowed: cc hygiene is # not a correctness invariant — runtime alert routing already # filters out missing IdentityPerson rows in # ``_resolve_billing_alert_recipients`` — but a clean cc list # keeps the admin UI honest. Idempotent (a re-run finds zero # references to strip and writes zero audit rows). # Savepoint discipline for every best-effort sweep below: a DB # lock-wait-timeout (1205) or deadlock (1213) inside a sweep puts # the shared session into a rollback-required state. Without a # SAVEPOINT the poison propagates — the caller's final commit # fails and rolls back the already-flushed ``employment_status= # 'left'`` flip, EVEN THOUGH ``result.flipped`` was set True → # the leaver is reported cut but is not durably cut. Wrapping each # sweep's DB writes in ``begin_nested()`` confines a failure to # its own savepoint, so the flip stays durable and the remaining # sweeps + the final commit still succeed. try: from app.domains.cmdb.lifecycle_service import ( strip_person_from_alert_cc, ) async with self.db.begin_nested(): await strip_person_from_alert_cc( self.db, str(person.id), actor_user_id=actor_user_id, trace_id=str(trace_id), via=via, ) except Exception as exc: # noqa: BLE001 logger.exception( "alert-cc strip failed during flip_to_left for %s — " "non-fatal; runtime routing still filters via " "IdentityPerson lookup", person.id, ) # Audit the silent-failure so an investigator looking at # "why does leaver X's UUID still appear as 未知用户 in # this scope's cc" has a row pointing at the trace_id + # exception class. App logs alone are easy to miss; an # audit row is the durable signal. try: from app.audit.service import AuditService audit = AuditService(self.db) async with self.db.begin_nested(): await audit.log( trace_id=str(trace_id), actor_type=( "user" if actor_user_id else "system" ), actor_user_id=actor_user_id, action="cloud.alert.cc_leaver_strip_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={ "via": via, "error_class": type(exc).__name__, "error_message": format_exception_for_log( exc, max_len=500, include_type=False, ), }, ) except Exception: # noqa: BLE001 logger.exception( "audit-write for cc-strip failure also failed " "(person=%s) — only the prior log line records this", person.id, ) # API Key shared-access roster scrub (migration 200): when a # person flips to ``left``, every row in # ``apikey_assignment_members`` that references them carries no # meaningful state — a leaver should not be able to read their # former teammates' API Key spend / reveal / detail surfaces # via a stolen PAT or a not-yet-rotated JWT. The # ``is_owner_or_member`` helper additionally filters leavers at # read time as defence-in-depth, but cleaning the rows here # closes the access surface immediately and keeps the Owner's # panel honest (no "leaver remains a member" residue). Mirrors # the existing ``strip_person_from_alert_cc`` sweep in posture # — best-effort, audited per-row, idempotent. try: from app.domains.apikey.service import ApiKeyService async with self.db.begin_nested(): removed = await ApiKeyService( self.db, ).cleanup_leaver_memberships( person_id=person.id, trace_id=str(trace_id), actor_user_id=actor_user_id, via=via, ) if removed: logger.info( "apikey-member leaver-strip removed %d rows for %s", removed, person.id, ) except Exception: # noqa: BLE001 logger.exception( "apikey-member leaver-strip failed during " "flip_to_left for %s — non-fatal; read-time leaver " "filter still gates access.", person.id, ) # Sweep the leaver's admin rows: ``asset_location_admins`` # (asset authority) and ``asset_office_local_admins`` # (当地 Admin, a contact list that grants nothing). # # Without this the join rows persist indefinitely — the FK # cascade only fires on a hard-delete of the location/office or # the person. The role auto-derive already revokes # ``asset_admin`` because the person is no longer # ``employment_status='active'``, but a surviving row shows up # as a ghost admin on the settings page and would silently # re-grant the role on rehire without operator confirmation. # Failures are logged and swallowed: hygiene only — the role # filter is the real gate. try: from sqlalchemy import delete as _delete # noqa: PLC0415 from app.auth.models import ( # noqa: PLC0415 AuthRole, AuthUserRole, ) from app.core.rbac import ( # noqa: PLC0415 ASSET_GLOBAL_ADMIN_ROLE, ) from app.domains.asset.location_admin_sync import ( # noqa: PLC0415 sync_asset_admin_role_for_person, ) from app.domains.asset.models import ( # noqa: PLC0415 AssetLocationAdmin, AssetOfficeLocalAdmin, ) async with self.db.begin_nested(): # Core DELETE is correct for these two: neither table # has a cycle ledger. Only ``AuthUserRole`` must go # through the ORM (see below). removed_location_admin = await self.db.execute( _delete(AssetLocationAdmin).where( AssetLocationAdmin.person_id == person.id, ) ) removed_location_count = getattr( removed_location_admin, "rowcount", 0, ) if removed_location_count: logger.info( "asset_location_admins leaver-strip removed %d " "rows for %s", removed_location_count, person.id, ) removed_local_admin = await self.db.execute( _delete(AssetOfficeLocalAdmin).where( AssetOfficeLocalAdmin.person_id == person.id, ) ) removed_local_count = getattr(removed_local_admin, "rowcount", 0) if removed_local_count: logger.info( "asset_office_local_admins leaver-strip removed " "%d rows for %s", removed_local_count, person.id, ) # Synchronously revoke the auto-derived ``asset_admin`` # role grant — without this the auth_user_roles row # survives until the daily 03:15 reconciler cron, and the # just-departed person keeps the role bit on their JWT for # up to ~24h. ``sync_asset_admin_role_for_person`` is a # no-op when the person administers no locations (which # the DELETE above guarantees) AND when the person is # not active — both conditions hold here, so the grant # is correctly revoked in this transaction. await sync_asset_admin_role_for_person(self.db, person.id) # Also strip a manually-granted ``asset_global_admin`` # (全局资产管理员) row. Same rationale as the admin-list # sweep above: the grant itself is inert while the person # is ``left`` (is_active_employee blocks mutations), but a # surviving row would silently restore GLOBAL asset # authority on rehire without operator confirmation — # worse than the regional ghost-admin case. Runs AFTER the # sync call deliberately: sync revokes asset_admin off the # person's inactive status regardless, so a failure here # can't strand the co-derived role. global_role_id = await self.db.scalar( select(AuthRole.id).where( AuthRole.name == ASSET_GLOBAL_ADMIN_ROLE, ) ) if global_role_id is not None: # ORM delete rather than Core DELETE so the # access-cycle ledger listener closes the holding # interval. A leaver whose global-admin cycle stays # open is the single worst point-in-time answer this # ledger can give, so this path especially must not # bypass the unit of work. stale_global = ( await self.db.execute( select(AuthUserRole).where( AuthUserRole.user_id == person.id, AuthUserRole.role_id == global_role_id, ) ) ).scalars().all() for stale in stale_global: await self.db.delete(stale) if stale_global: logger.info( "asset_global_admin leaver-strip removed the " "grant for %s", person.id, ) except Exception: # noqa: BLE001 logger.exception( "asset admin-list leaver-strip failed during " "flip_to_left for %s — non-fatal; role-derive filter " "still excludes leavers.", person.id, ) # Pre-launch P1 fix: cascade-cancel any still-pending request # rows in the 5 self-service request domains # (sso / vpn / ai_tool / announcement / domain_registration). # The Feishu webhooks for these flows already re-validate # employment_status before flipping APPROVED, but a row that # never gets a callback (Feishu instance deleted, network # outage, future replay queue) sits forever pending — # polluting admin queues and leaving a leaver in the request # graph as if they had open requests. Sweep them at flip-time. # Failures are logged + swallowed: this is hygiene, not a # correctness invariant (the webhook re-validate is the real # gate). Idempotent. try: async with self.db.begin_nested(): await self._cancel_pending_requests_for_leaver( person, via=via, trace_id=trace_id, actor_user_id=actor_user_id, ) except Exception: # noqa: BLE001 logger.exception( "cancel-pending-requests sweep failed during " "flip_to_left for %s — non-fatal; webhook re-validate " "still gates approve.", person.id, ) # Login-binding deactivation sweep: flip the leaver's Google / # Feishu ``identity_external_identities`` rows to # ``is_login_identity=False`` + ``identity_status='inactive'``. # The auth layer already rejects every ``is_effectively_departed`` # request, so this is data-hygiene / defence-in-depth, not the # security gate — but without it the binding row stays # login-enabled + active forever and the identity data-quality # scanner keeps reporting a false "已离职但仍可登录" finding # (prod 2026-06-09: 5 such stale findings). The ``successfactors`` # provider row is intentionally untouched (HR sync owns it; it is # never a login binding). Same posture as the sweeps above — # best-effort, audited, idempotent. try: from app.audit.service import AuditService # noqa: PLC0415 from app.identity.identity_reconcile import ( # noqa: PLC0415 deactivate_login_identities, ) async with self.db.begin_nested(): deactivated = await deactivate_login_identities( self.db, person.id, ) if deactivated: logger.info( "login-binding leaver-deactivate flipped %d " "row(s) off for %s", deactivated, person.id, ) await AuditService(self.db).log( trace_id=str(trace_id), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action=( "identity.login_binding_deactivated_on_leave" ), status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "via": via, "bindings_deactivated": deactivated, }, ) except Exception: # noqa: BLE001 logger.exception( "login-binding leaver-deactivate failed during " "flip_to_left for %s — non-fatal; auth's " "is_effectively_departed gate still blocks login.", person.id, ) result.flipped = True return result async def _cancel_pending_requests_for_leaver( self, person, *, via: str, trace_id: uuid.UUID, actor_user_id: uuid.UUID | None, ) -> None: """Flip every PENDING row across SSO grant / VPN / AI-tool / announcement / domain registration request domains to CANCELLED for a person who just flipped to ``left``. Pre-launch P1 fix. Writes one ``leaver_pending_request_cancelled`` audit row per domain summarising the count + ids. Each row update is its own service-less ORM mutation — these domains intentionally share the same minimal shape (``person_id`` + ``status``). """ from sqlalchemy import update as _update # noqa: PLC0415 from app.audit.service import AuditService # noqa: PLC0415 from app.domains.ai_tool_application.models import ( # noqa: PLC0415 AI_TOOL_STATUS_CANCELLED, AI_TOOL_STATUS_PENDING, AiToolApplication, ) from app.domains.announcement_request.models import ( # noqa: PLC0415 ANNOUNCEMENT_REQ_STATUS_CANCELLED, ANNOUNCEMENT_REQ_STATUS_PENDING, AnnouncementRequest, ) from app.domains.domain_registration.models import ( # noqa: PLC0415 DOMAIN_REG_STATUS_CANCELLED, DOMAIN_REG_STATUS_PENDING, DomainRegistration, ) from app.domains.sso.models import ( # noqa: PLC0415 SSO_GRANT_STATUS_CANCELLED, SSO_GRANT_STATUS_PENDING, SsoGrant, ) from app.domains.vpn_application.models import ( # noqa: PLC0415 VPN_APP_STATUS_CANCELLED, VPN_APP_STATUS_PENDING, VpnApplication, ) person_id_str = str(person.id) domains = [ ( "sso_grant_request", SsoGrant, SSO_GRANT_STATUS_PENDING, SSO_GRANT_STATUS_CANCELLED, ), ( "vpn_application_request", VpnApplication, VPN_APP_STATUS_PENDING, VPN_APP_STATUS_CANCELLED, ), ( "ai_tool_application_request", AiToolApplication, AI_TOOL_STATUS_PENDING, AI_TOOL_STATUS_CANCELLED, ), ( "announcement_request", AnnouncementRequest, ANNOUNCEMENT_REQ_STATUS_PENDING, ANNOUNCEMENT_REQ_STATUS_CANCELLED, ), ( "domain_registration_request", DomainRegistration, DOMAIN_REG_STATUS_PENDING, DOMAIN_REG_STATUS_CANCELLED, ), ] audit = AuditService(self.db) total = 0 for action_key, Model, pending_v, cancelled_v in domains: # Two-phase: snapshot ids first (for audit detail), then # bulk-update. The flush listener doesn't react to these # tables — only ``identity_persons.employment_status``. ids_q = await self.db.execute( select(Model.id).where( # type: ignore[arg-type] Model.person_id == person_id_str, Model.status == pending_v, ) ) ids = [str(r) for r in ids_q.scalars().all()] if not ids: continue await self.db.execute( _update(Model) .where( Model.person_id == person_id_str, Model.status == pending_v, ) .values(status=cancelled_v) ) total += len(ids) await audit.log( trace_id=str(trace_id), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action=f"leaver.{action_key}.cancelled_on_flip", status="success", resource_type=Model.__tablename__, resource_id=person_id_str, detail={ "via": via, "person_id": person_id_str, "cancelled_count": len(ids), "cancelled_ids": ids[:50], }, ) # Pre-launch P1 fix (Round 45, CR-407): also cancel pending # rows in employee_requests (asset request / laptop replacement # / computer borrow / asset return / buyback / api_key_request). # That table uses ``requester_id`` not ``person_id``, so it # doesn't fit the generic loop. Without this sweep, a leaver's # pending asset requests pile up in the admin queue forever. # SAVEPOINT-isolate (2026-08-06): this sweep writes request rows; # poisoning here used to take the CMDB sweep right below AND the # caller's ``employment_status='left'`` flush down with it. async with best_effort( self.db, label=f"leaver flip: cancel employee_requests for {person.work_email}", log=logger, ) as er_step: from app.domains.asset.models import EmployeeRequest # noqa: PLC0415 # Pre-launch P1 fix (Round 45, CR-407 follow-up): exclude # `offboarding_buyback` and `asset_return` — these flows # are submitted by leavers AS PART OF offboarding (buyback # is by definition the leaver buying their own device; # asset_return is the leaver handing back hardware on the # way out). Cancelling them at the leaver-flip moment # destroys exactly the requests that need to survive into # the offboarding process. exempt_request_types = ("offboarding_buyback", "asset_return") er_ids_q = await self.db.execute( select(EmployeeRequest.id).where( EmployeeRequest.requester_id == person.id, EmployeeRequest.status == "pending", EmployeeRequest.request_type.notin_(exempt_request_types), ) ) er_ids = [str(r) for r in er_ids_q.scalars().all()] if er_ids: await self.db.execute( _update(EmployeeRequest) .where( EmployeeRequest.requester_id == person.id, EmployeeRequest.status == "pending", EmployeeRequest.request_type.notin_(exempt_request_types), ) .values(status="cancelled") ) total += len(er_ids) await audit.log( trace_id=str(trace_id), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="leaver.employee_request.cancelled_on_flip", status="success", resource_type="employee_request", resource_id=person_id_str, detail={ "via": via, "person_id": person_id_str, "cancelled_count": len(er_ids), "cancelled_ids": er_ids[:50], "exempted_request_types": list(exempt_request_types), }, ) if not er_step.ok: logger.warning( "leaver flip: failed to cancel employee_requests for " "%s: %s", person.work_email, er_step.reason(), ) # Pre-launch P1 fix (Round 45 follow-up, CR-315): also cancel # pending CMDB self-service request rows. These two tables use # ``requester_person_id`` (not ``person_id``) so they can't # share the generic loop above. Without this sweep, a leaver's # pending DNS records / cloud-scope-onboarding requests pile # up under their now-inactive name in admin queues. # SAVEPOINT-isolate — same rationale as the employee_requests sweep. async with best_effort( self.db, label=f"leaver flip: cancel CMDB requests for {person.work_email}", log=logger, ) as cmdb_step: from app.domains.cmdb.models import ( # noqa: PLC0415 DNS_REQ_CANCELLED, DNS_REQ_PENDING, SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_WITHDRAWN, CmdbCloudScopeRequest, CmdbDnsRecordRequest, ) for action_key, Model, pending_v, cancelled_v in ( ( "cmdb_dns_record_request", CmdbDnsRecordRequest, DNS_REQ_PENDING, DNS_REQ_CANCELLED, ), ( "cmdb_cloud_scope_request", CmdbCloudScopeRequest, SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_WITHDRAWN, ), ): cmdb_ids_q = await self.db.execute( select(Model.id).where( # type: ignore[arg-type] Model.requester_person_id == person_id_str, Model.status == pending_v, ) ) cmdb_ids = [str(r) for r in cmdb_ids_q.scalars().all()] if not cmdb_ids: continue await self.db.execute( _update(Model) .where( Model.requester_person_id == person_id_str, Model.status == pending_v, ) .values(status=cancelled_v) ) total += len(cmdb_ids) await audit.log( trace_id=str(trace_id), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action=f"leaver.{action_key}.cancelled_on_flip", status="success", resource_type=Model.__tablename__, resource_id=person_id_str, detail={ "via": via, "person_id": person_id_str, "cancelled_count": len(cmdb_ids), "cancelled_ids": cmdb_ids[:50], }, ) if not cmdb_step.ok: logger.warning( "leaver flip: failed to cancel CMDB requests for " "%s: %s", person.work_email, cmdb_step.reason(), ) if total: logger.info( "leaver flip cascade-cancelled %d pending requests " "across 8 domains for %s (via=%s)", total, person.work_email, via, ) # ── Internal: audit + override validation ─────────────────── async def _write_soft_audit( self, *, person, via: str, denied_reason: str, detail: dict[str, Any], trace_id: uuid.UUID, actor_user_id: uuid.UUID | None, client_ip: str | None, user_agent: str | None, ) -> None: """Audit a non-rejection path that took an unusual route. Writes a row with ``target_status='left'`` and the unusual ``denied_reason`` as a tag, but does NOT raise. Used when the guard allows the flip on a degenerate path (no process / cancelled process) so operators can later distinguish historical-loose flips from rigorously-cleared flips. """ row = LeaverBlockAudit( id=uuid.uuid4(), person_id=person.id, attempted_at=datetime.now(UTC), attempted_by_user_id=actor_user_id, attempted_via=via, target_status="left", denied_reason=f"soft:{denied_reason}", pending_blocker_count=0, pending_blockers=[], detail=detail, trace_id=trace_id, client_ip=client_ip, user_agent=user_agent, ) self.db.add(row) await self.db.flush() logger.info( "LeaverBlocked: soft-allow flip-to-left for %s via %s " "(reason=%s) — post_leaver_verify will check orphans", person.work_email, via, denied_reason, ) async def _write_audit( self, *, person, via: str, target_status: str, denied_reason: str, pending_blockers: list[dict[str, Any]], detail: dict[str, Any] | None, trace_id: uuid.UUID, actor_user_id: uuid.UUID | None, client_ip: str | None, user_agent: str | None, ) -> uuid.UUID: row = LeaverBlockAudit( id=uuid.uuid4(), person_id=person.id, attempted_at=datetime.now(UTC), attempted_by_user_id=actor_user_id, attempted_via=via, target_status=target_status, denied_reason=denied_reason, pending_blocker_count=len(pending_blockers), pending_blockers=pending_blockers, detail=detail, trace_id=trace_id, client_ip=client_ip, user_agent=user_agent, ) self.db.add(row) await self.db.flush() logger.warning( "LeaverBlocked: rejected status='%s' for %s via %s " "reason=%s pending=%d trace=%s", target_status, person.work_email, via, denied_reason, len(pending_blockers), trace_id, ) return row.id async def _validate_override( self, *, override_id: uuid.UUID, person_id: uuid.UUID, process_id: uuid.UUID, pending_blocker_ids: list[str], ) -> LeaverDualSignOverride: override = await self.db.get(LeaverDualSignOverride, override_id) if override is None: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=f"Dual-sign override {override_id} not found", ) if override.status != DUAL_SIGN_APPROVED: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} is in status " f"{override.status!r}; only 'approved' overrides " f"can be executed" ), ) # Defense-in-depth expiry check: override_service.execute() # also checks this, but guard._validate_override is the gate # for ALL flip_to_left callers — some bypass the service layer. # Treat NULL expires_at as expired (should never occur given # initiate() always sets it, but a direct DB edit could set it). now_utc = datetime.now(UTC) exp = override.expires_at if exp is None: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} has no expiry set; " "refusing as a safety measure. Re-initiate the override." ), ) if exp.tzinfo is None: exp = exp.replace(tzinfo=UTC) if exp < now_utc: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} expired at " f"{exp.isoformat()}; re-initiate the override." ), ) if override.process_id != process_id: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} targets a " f"different OffboardingProcess " f"({override.process_id})" ), ) if override.person_id != person_id: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} targets a " f"different person ({override.person_id})" ), ) # Coverage check: every currently-pending blocker must be # listed in the override's overridden_blocker_ids. We don't # require the lists to MATCH (the override may name blockers # that have since been resolved by other means) — only that # everything still pending is in the override's set. listed = set(override.overridden_blocker_ids or []) uncovered = [bid for bid in pending_blocker_ids if bid not in listed] if uncovered: raise LeaverBlockedError( person_id=str(person_id), pending_blocker_ids=pending_blocker_ids, message=( f"Dual-sign override {override_id} does not cover " f"{len(uncovered)} new pending blocker(s) that " f"appeared after the override was approved: " f"{uncovered}. Re-initiate the override." ), ) return override async def _revoke_orphan_seats( self, person, *, via: str, trace_id: uuid.UUID | str, actor_user_id: uuid.UUID | None = None, ) -> None: """Revoke any still-active SoftwareSeatAllocation rows for the person — but ONLY the ones the offboarding pipeline is not already handling. Was designed as defense-in-depth for the soft-allow path (no offboarding process / cancelled process). The original implementation iterated every ``status='active'`` seat and re-enqueued vendor revokes regardless of whether the rigorous path had already done so — because ``OffboardingService. _apply_software_policy`` writes its revoke task with ``not_before = leaver_date + 1 day`` (deferring vendor execution to T+1 in the leaver's local TZ), the seat stays ``status=active`` until the task actually fires, so this sweep saw it as an orphan even when offboarding was already on it. Production effect: every leaver going through start_process followed by flip_to_left in the same HRIS sync produced TWO revoke ``ProvisioningTask`` rows per seat — the offboarding Path A row (with ``offboarding_blocker_id`` + auto-resolved successor) and the orphan-sweep Path B row (bare context, no successor). The second row crashed the Google ``user_suspend`` precheck because successors are mandatory for Drive/Calendar handoff; the failure DM then double-paged the admin. Fix: * Per-seat dedup against ``ProvisioningTask`` (PENDING / IN_PROGRESS / COMPLETED) via the shared ``find_in_flight_seat_revoke_task`` helper — if the pipeline already owns this seat, skip every side effect (no flip, no enqueue, no audit-revoke). Write a single ``leaver_block_guard.orphan_sweep_skipped`` audit row per skip so the dedup is traceable. * For true orphans, resolve a successor when the bound driver requires one (Zoom ``action_on_offboard=delete``, Google ``operation=user_suspend``). Manager → IT-admin pool, gated on candidate holding an active seat on the same software (Google Data Transfer rejects recipients who don't exist in the directory). If unresolvable, flip the seat (billing-correctness invariant stands) but DO NOT enqueue a doomed vendor revoke; raise a ``leaver_orphan`` compliance incident so the admin sees it and can pick a successor manually. Cross-domain note: this domain (leaver_blocked) avoids compile-time imports from asset / provisioning to keep the domain-boundary rule intact. Lazy imports are the established pattern in this guard (see assert_can_flip_to_left's lazy import of offboarding models). """ from app.audit.service import AuditService from app.compliance.service import ComplianceIncidentService from app.domains.asset.models import ( SoftwareLicensePlan, SoftwareSeatAllocation, ) rows = list((await self.db.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan) .join( SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id, ) .where( SoftwareSeatAllocation.user_id == person.id, SoftwareSeatAllocation.status == "active", ), )).all()) if not rows: return audit = AuditService(self.db) incidents = ComplianceIncidentService(self.db) trace_str = str(trace_id) now = datetime.now(UTC) total_seats = len(rows) processed_with_errors = 0 for seat, plan in rows: # Per-seat best-effort isolation: a DB error or unexpected # exception while processing seat N (dedup query timeout, # successor-resolution SELECT failure, audit insert # collision) must NOT abandon seats N+1..M. The whole sweep # is "defense-in-depth for orphans" — losing one seat to # an exception just means ``post_leaver_verify`` (T+1) # picks it up later. Letting an exception escape would # also propagate up into ``flip_to_left``'s caller and # roll back the person's ``employment_status='left'`` # flush, which is exactly the wrong outcome (the flip is # the billing-correctness invariant and must stand). # SAVEPOINT-isolate (2026-08-06) — without it the guarantee # above is not actually delivered: a DB-level error in # ``_sweep_one_seat`` deactivates the transaction, the swallow # succeeds but the flush below raises, and that escapes # ``flip_to_left``. Every other hygiene sweep in this class # already wraps in ``begin_nested``; this one was the omission. async with best_effort( self.db, label=f"leaver orphan seat sweep {seat.id}", log=logger, ) as sweep_step: await self._sweep_one_seat( person=person, seat=seat, plan=plan, now=now, via=via, trace_str=trace_str, actor_user_id=actor_user_id, audit=audit, incidents=incidents, ) if not sweep_step.ok: processed_with_errors += 1 logger.error( "leaver_block_guard: orphan sweep raised on seat " "%s for person %s (%s) — continuing with remaining " "seats; post_leaver_verify will retry this one", seat.id, person.id, sweep_step.reason(), ) await self.db.flush() logger.warning( "leaver_block_guard: orphan sweep for %s via %s — " "%d seat(s) processed, %d raised (recovered)", person.work_email, via, total_seats, processed_with_errors, ) async def _sweep_one_seat( self, *, person, seat, plan, now: datetime, via: str, trace_str: str, actor_user_id: uuid.UUID | None, audit, incidents, ) -> None: """Process a single orphan seat. Extracted from ``_revoke_orphan_seats`` so the caller can wrap each iteration in its own try/except. Per-seat outcomes are attributed via audit rows (``orphan_sweep_skipped`` / ``provisioning.task_enqueued`` from the coordinator / ``compliance.incident_created`` when applicable) rather than a function-local counter, so the audit log is the source of truth for "what happened" — the parent's summary log is purely advisory. """ from app.compliance.models import ( SEVERITY_CRITICAL, TYPE_LEAVER_ORPHAN, ) from app.core.coordinators.provisioning_coordinator import ( enqueue_revoke_for_seat, find_in_flight_seat_revoke_task, resolve_revoke_requires_successor, ) software_id = plan.software_id # ── Dedup gate: if the offboarding pipeline already owns # this seat (PENDING / IN_PROGRESS / COMPLETED revoke # task exists), the orphan sweep MUST short-circuit. Flip # + enqueue here would compete with the pipeline's task, # the second one always lacks the auto-resolved successor # context, and the resulting failure double-DMs IT admins. existing_task = await find_in_flight_seat_revoke_task( self.db, user_id=person.id, software_id=software_id, seat_id=seat.id, ) if existing_task is not None: await audit.log( trace_id=trace_str, actor_type=( "user" if actor_user_id is not None else "system" ), actor_user_id=actor_user_id, action="leaver_block_guard.orphan_sweep_skipped", status="success", resource_type="software_seat_allocation", resource_id=str(seat.id), detail={ "via": via, "reason": "offboarding_pipeline_owns_seat", "existing_task_id": str(existing_task.id), "existing_task_status": existing_task.status, "person_id": str(person.id), "software_id": ( str(software_id) if software_id else None ), }, ) return # True orphan path. Decide if the driver needs a successor # BEFORE we mutate Helix state, so a doomed-vendor-revoke # doesn't leave a confusing audit trail. successor_user_id: uuid.UUID | None = None successor_email: str | None = None needs_successor = False successor_resolution_reason = "" if software_id is not None: needs_successor, _provider = await resolve_revoke_requires_successor( self.db, software_id=software_id, ) if needs_successor: ( successor_user_id, successor_email, successor_resolution_reason, ) = await self._resolve_orphan_seat_successor( leaver=person, software_id=software_id, ) # Flip Helix-side state regardless of vendor outcome — the # billing-correctness invariant. The vendor revoke is # best-effort; missing-successor branch below still flips # the seat but skips the enqueue and opens an incident. seat.status = "revoked" seat.revoked_at = now seat.reason = ( (seat.reason or "") + f"\n[auto-revoked by leaver_block_guard ({via}): " "person flipped to 'left' on the soft-allow path " "with no offboarding process to clean up seats]" ).strip() if needs_successor and successor_user_id is None: # Cannot safely call the driver — Google ``user_suspend`` / # Zoom ``delete`` will refuse without a successor (data # hand-off). Open a compliance incident so the admin sees # it and can pick someone in the offboarding UI. Skip # enqueue. # SAVEPOINT-isolate — the incident write is best-effort but must # not poison the per-seat sweep it runs inside. async with best_effort( self.db, label=f"leaver orphan-seat incident {seat.id}", log=logger, ): # ``resource_id`` is ``CHAR(36)`` on # ``compliance_incidents``; pass ``str(seat.id)`` so # the dedup SELECT inside create_or_upsert binds against # the same string representation asyncmy stores # (memory ref: reference_asyncmy_uuid_bind + the four # offboarding callsites pinned by # test_incident_per_process_dedup_2026_05_31). await incidents.create_or_upsert_incident( incident_type=TYPE_LEAVER_ORPHAN, severity=SEVERITY_CRITICAL, person_id=person.id, resource_type="software_seat_allocation", resource_id=str(seat.id), resource_label=( f"{plan.name} (seat {str(seat.id)[:8]})" ), title=( "Orphan seat revoke needs manual successor: " f"{person.work_email}" ), detail={ "via": via, "software_id": str(software_id), "plan_id": str(plan.id), "plan_name": plan.name, "successor_resolution_reason": ( successor_resolution_reason or "(no reason recorded)" ), "note": ( "Helix-side seat revoked for billing " "correctness; vendor-side revoke held " "until admin picks a successor." ), }, ) return # Driver doesn't need successor OR we resolved one: best- # effort vendor revoke. Failures here must NOT roll back # the flip — the Helix-side revoke is the billing- # correctness invariant; vendor-side cleanup is retried by # ``post_leaver_verify`` and the next provisioning dispatch # tick. async with best_effort( self.db, label=f"leaver: enqueue revoke for seat {seat.id}", log=logger, ) as revoke_step: await enqueue_revoke_for_seat( self.db, seat=seat, trace_id=trace_str, actor_user_id=actor_user_id, enforce_offboarding_policy=False, successor_user_id=successor_user_id, successor_email=successor_email, ) if not revoke_step.ok: logger.error( "leaver_block_guard: enqueue_revoke_for_seat failed " "for seat %s person %s (%s) — Helix-side revoke stands; " "vendor cleanup will retry via post_leaver_verify", seat.id, person.id, revoke_step.reason(), ) async def _resolve_orphan_seat_successor( self, *, leaver, software_id: uuid.UUID, ) -> tuple[uuid.UUID | None, str | None, str]: """Pick a successor for a true-orphan seat revoke. Returns ``(successor_user_id, successor_email, reason)``. Cheap echo of ``OffboardingService._auto_successor_for_seat`` intentionally NOT shared via cross-domain import (this guard avoids leaver_blocked → offboarding compile-time deps — importing the offboarding service here would pull asset + provisioning + integrations transitively, fattening the import graph for every endpoint that touches employment status). The two implementations must agree on what counts as a valid successor; a regression test in ``test_orphan_sweep_dedup.py`` pins the contract. Candidate order: leaver's manager → IT-admin pool (active global_admin / it_admin role-holders). A candidate must hold an ACTIVE seat on the SAME software so the vendor data transfer lands on an account that actually exists in the provider (Google Data Transfer returns 404 on recipient if the new owner has no Google Workspace presence — see google_workspace.py:_initiate_data_transfer). """ from app.auth.models import AuthRole, AuthUserRole from app.domains.asset.models import ( SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.identity.person_models import IdentityPerson async def _candidate_holds_seat(candidate_id: uuid.UUID) -> bool: # ``status='active'`` is intentional: Google Data Transfer # (and Zoom delete handoff) require the recipient to have # an EXISTING vendor account. A ``pending`` Helix-side seat # means the allocation row was created but the driver's # grant call hasn't run yet — the vendor doesn't know the # candidate exists yet, so passing them as the successor # would 404 on the recipient lookup. ``revoked`` candidates # are obviously wrong. Future drivers that don't need a # pre-existing vendor account could relax this — flag at # that point, do not relax pre-emptively. row = await self.db.scalar( select(SoftwareSeatAllocation.id) .join( SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id, ) .where( SoftwareSeatAllocation.user_id == candidate_id, SoftwareLicensePlan.software_id == software_id, SoftwareSeatAllocation.status == "active", ), ) return row is not None # Tier 1: manager. ``leaver`` is always an IdentityPerson (caller # passes ``leaver=person``); ``manager_person_id`` is a real column, # so use direct access (project rule: no getattr-default on ORM cols). manager_id = leaver.manager_person_id if manager_id is not None: manager = await self.db.get(IdentityPerson, manager_id) if ( manager is not None and manager.employment_status == "active" and manager.deleted_at is None ): if await _candidate_holds_seat(manager.id): return manager.id, manager.work_email, "" # Manager exists but has no seat — fall through to # IT-admin pool. Record why so the incident detail # is honest when nobody qualifies. reason_so_far = ( "leaver_manager_has_no_seat_on_this_software" ) else: reason_so_far = "leaver_manager_inactive_or_missing" else: reason_so_far = "leaver_has_no_manager_record" # Tier 2: IT-admin pool — first active admin who holds a seat admin_ids = list(( await self.db.execute( select(AuthUserRole.user_id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthRole.name.in_(("global_admin", "it_admin"))) ) ).scalars().all()) for admin_id in admin_ids: admin = await self.db.get(IdentityPerson, admin_id) if ( admin is None or admin.employment_status != "active" or admin.deleted_at is not None ): continue if await _candidate_holds_seat(admin.id): return admin.id, admin.work_email, "" return ( None, None, f"{reason_so_far}; no IT-admin holds an active seat on this software", ) @contextmanager def _authorized( self, *, person_id: uuid.UUID, override_id: uuid.UUID | None, via: str, actor_user_id: uuid.UUID | None, ) -> Iterator[None]: tok_a = GUARD_AUTHORIZED.set(True) tok_p = GUARD_AUTHORIZED_FOR_PERSON.set(person_id) tok_o = GUARD_OVERRIDE_ID.set(override_id) tok_v = GUARD_VIA.set(via) tok_actor = GUARD_ACTOR.set(actor_user_id) try: yield finally: GUARD_ACTOR.reset(tok_actor) GUARD_VIA.reset(tok_v) GUARD_OVERRIDE_ID.reset(tok_o) GUARD_AUTHORIZED_FOR_PERSON.reset(tok_p) GUARD_AUTHORIZED.reset(tok_a) def _coerce_uuid(value: uuid.UUID | str | None) -> uuid.UUID: """Accept a UUID, a UUID-shape string, or a free-form tag. Tag strings (e.g. ``"process_advance"`` from scheduled-job handlers that don't have an HTTP request's real trace_id) are hashed deterministically into a UUID v5 so the row's UUID column stays a real UUID. Mirrors AuditService._coerce_trace_id. """ if value is None: raise ValueError("trace_id / override_id cannot be None") if isinstance(value, uuid.UUID): return value s = str(value) try: return uuid.UUID(s) except (ValueError, TypeError): return uuid.uuid5(uuid.NAMESPACE_OID, s) __all__ = [ "LeaverBlockGuard", "LeaverBlockResult", "GUARD_AUTHORIZED", "GUARD_AUTHORIZED_FOR_PERSON", "GUARD_OVERRIDE_ID", "GUARD_VIA", "GUARD_ACTOR", "LEAVER_VIA_ADMIN_API", "LEAVER_VIA_HRIS_SYNC", "LEAVER_VIA_PROCESS_ADVANCE", "LEAVER_VIA_DUAL_SIGN_OVERRIDE", "LEAVER_VIA_CLI", ]