"""Offboarding leaver → cloud-principal assignment sweep. Root-cause fix for AWS Identity Center "blank-name" orphans ----------------------------------------------------------- An Identity Center account assignment is the triple ``(PrincipalId, PermissionSet, Account)``. The console renders the "username" by resolving ``PrincipalId`` against the Identity Store at read time. When a leaver is offboarded, the SCIM deprovision path DELETEs the directory user — but AWS does NOT cascade-delete their account assignments. Every assignment Helix never tracked as a grant row (created directly in the AWS console, pre-Helix, or by another tool) is left pointing at a now-dangling ``PrincipalId`` and shows up as a blank-name row on the account long after the person is gone. ``orphan_cleanup`` cleans these up *after the fact*, but only an admin running a scan surfaces them and they can no longer be attributed to a person. This module closes the gap at the source: while the leaver's Identity Store principal is still alive (i.e. BEFORE / independent of the SCIM delete), resolve every assignment they hold and delete it. The assignment never becomes an orphan. Scope guardrails (matches the user's "全帐号 · 只非生产" decision and the standing "prod cloud access is unmanaged" posture): * Only assignments on accounts Helix tracks as a non-prod :class:`CmdbCloudScope` are deleted. Prod scopes are skipped while the global prod-write lock is engaged (default) — same ``ensure_prod_scope_writable`` gate ``orphan_cleanup`` uses. * Assignments on accounts with NO Helix scope row are surfaced but NEVER deleted — without a scope row there is no ``environment`` to prod-gate against, so "don't touch what we can't prove is non-prod". * Assignments that already map to an active/pending Helix grant row are left to the standard offboarding grant-revoke path (``RES_CLOUD_ACCESS_GRANT``) — this sweep only owns the UNTRACKED residue, avoiding a double revoke + double history row. Idempotent: a re-run after the assignments are gone lists nothing and is a clean no-op. ``dry_run=True`` returns the plan without any cloud write — used by the read-only "would-sweep" preview and the MCP scan tool. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from datetime import UTC, datetime from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.database import async_session_factory from app.core.exceptions import ProdCloudLockedError from app.domains.cmdb.models import ( GRANT_REVOKE_REASON_LEAVER_PRINCIPAL_SWEPT, GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING, GRANT_STATUS_REVOKED, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudScope, ) from app.domains.cmdb.prod_write_lock import ensure_prod_scope_writable from app.identity.person_models import IdentityPerson from app.integrations.driver_base import ( CloudDriver, CloudScopeRef, OrphanAssignment, ) from app.integrations.models import Integration logger = logging.getLogger(__name__) # Per-assignment outcome codes (stable strings for logs / MCP payloads). OUTCOME_DELETED = "deleted" OUTCOME_ALREADY_ABSENT = "already_absent" OUTCOME_WOULD_DELETE = "would_delete" # dry_run OUTCOME_SKIPPED_UNMANAGED = "skipped_unmanaged_account" OUTCOME_SKIPPED_PROD_LOCKED = "skipped_prod_locked" OUTCOME_SKIPPED_TRACKED_GRANT = "skipped_tracked_grant" OUTCOME_ERROR = "error" @dataclass(slots=True) class SweepAssignmentResult: integration_id: str integration_name: str | None account_id: str scope_id: str | None role_key: str role_display_name: str | None principal_id: str outcome: str grant_id: str | None = None error_message: str | None = None @dataclass(slots=True) class LeaverSweepResult: person_id: str email: str | None dry_run: bool results: list[SweepAssignmentResult] = field(default_factory=list) # Integration-level failures (driver load / list_principal_assignments # raised) that prevented enumerating that integration at all. integration_errors: list[dict[str, str]] = field(default_factory=list) @property def deleted_count(self) -> int: """Assignments this run actively deleted cloud-side. Excludes ``already_absent`` (the driver reported the assignment was already gone — nothing was deleted this run), so re-runs and drivers that momentarily report ``already_absent`` don't inflate the figure the coordinator logs. Use :attr:`reconciled_count` for the combined "handled" total. """ return sum(1 for r in self.results if r.outcome == OUTCOME_DELETED) @property def already_absent_count(self) -> int: return sum( 1 for r in self.results if r.outcome == OUTCOME_ALREADY_ABSENT ) @property def reconciled_count(self) -> int: """Deleted this run + already-absent — total assignments confirmed cut cloud-side after the sweep.""" return self.deleted_count + self.already_absent_count @property def would_delete_count(self) -> int: return sum( 1 for r in self.results if r.outcome == OUTCOME_WOULD_DELETE ) @property def error_count(self) -> int: return ( sum(1 for r in self.results if r.outcome == OUTCOME_ERROR) + len(self.integration_errors) ) @property def skipped_prod_count(self) -> int: return sum( 1 for r in self.results if r.outcome == OUTCOME_SKIPPED_PROD_LOCKED ) def summary(self) -> dict[str, int]: return { "total": len(self.results), "deleted": self.deleted_count, "already_absent": self.already_absent_count, "would_delete": self.would_delete_count, "skipped_prod_locked": self.skipped_prod_count, "skipped_unmanaged": sum( 1 for r in self.results if r.outcome == OUTCOME_SKIPPED_UNMANAGED ), "skipped_tracked_grant": sum( 1 for r in self.results if r.outcome == OUTCOME_SKIPPED_TRACKED_GRANT ), "errors": self.error_count, } async def sweep_leaver_sso_assignments( person_id: str | uuid.UUID, *, actor_user_id: uuid.UUID | None, trace_id: str, dry_run: bool = False, ) -> LeaverSweepResult: """Delete every non-prod, untracked Identity Center account assignment the leaver ``person_id`` still holds. Read-only when ``dry_run=True`` (no cloud write, no history row) — the per-assignment outcomes come back as :data:`OUTCOME_WOULD_DELETE` so the caller can preview exactly what a real run would remove. """ from app.domains.cmdb.lifecycle_service import _load_cloud_driver pid = uuid.UUID(str(person_id)) async with async_session_factory() as session: person = await session.get(IdentityPerson, pid) if person is None: return LeaverSweepResult( person_id=str(pid), email=None, dry_run=dry_run, ) email = (person.work_email or "").strip() result = LeaverSweepResult( person_id=str(pid), email=email or None, dry_run=dry_run, ) if not email: logger.info( "leaver-sweep: person %s has no work_email — nothing to " "resolve against Identity Center", pid, ) return result integrations = (await session.execute( select(Integration).where( Integration.kind == "cloud", Integration.enabled.is_(True), ) )).scalars().all() audit = AuditService(session) for integration in integrations: try: driver: CloudDriver = await _load_cloud_driver( session, integration, ) except Exception as exc: # noqa: BLE001 logger.warning( "leaver-sweep: cannot load driver for integration " "%s (%s): %s", integration.id, integration.name, exc, ) result.integration_errors.append({ "integration_id": str(integration.id), "integration_name": integration.name or "", "error": str(exc), }) continue try: try: assignments = await driver.list_principal_assignments( email, ) except NotImplementedError: # Provider has no principal→assignment index (only # AWS Identity Center does today). Nothing to sweep. continue except Exception as exc: # noqa: BLE001 logger.warning( "leaver-sweep: list_principal_assignments failed " "for %s on integration %s: %s", email, integration.name, exc, ) result.integration_errors.append({ "integration_id": str(integration.id), "integration_name": integration.name or "", "error": str(exc), }) continue if not assignments: continue # Map every account_id this leaver holds → its Helix scope # (one query per integration instead of per assignment). account_ids = {a.account_id for a in assignments} scope_by_account: dict[str, CmdbCloudScope] = {} if account_ids: rows = (await session.execute( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == str( integration.id, ), CmdbCloudScope.scope_type == "aws_account", CmdbCloudScope.external_id.in_(account_ids), CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.status == SCOPE_STATUS_ACTIVE, ) )).scalars().all() scope_by_account = {r.external_id: r for r in rows} for asgn in assignments: res = await _sweep_one( session=session, audit=audit, driver=driver, integration=integration, person=person, assignment=asgn, scope=scope_by_account.get(asgn.account_id), actor_user_id=actor_user_id, trace_id=trace_id, dry_run=dry_run, ) result.results.append(res) finally: try: await driver.close() except Exception: # noqa: BLE001 pass return result async def _sweep_one( *, session: AsyncSession, audit: AuditService, driver: CloudDriver, integration: Integration, person: IdentityPerson, assignment, scope: CmdbCloudScope | None, actor_user_id: uuid.UUID | None, trace_id: str, dry_run: bool, ) -> SweepAssignmentResult: base = SweepAssignmentResult( integration_id=str(integration.id), integration_name=integration.name, account_id=assignment.account_id, scope_id=str(scope.id) if scope is not None else None, role_key=assignment.role_key, role_display_name=assignment.role_display_name, principal_id=assignment.principal_id, outcome=OUTCOME_ERROR, ) # (1) No Helix scope row → we can't classify the account's # environment, so we refuse to touch it. Surfaced so an admin can # register the account (and then re-run) or clean it manually. if scope is None: base.outcome = OUTCOME_SKIPPED_UNMANAGED return base # (2) Prod kill-switch — identical gate to orphan_cleanup. Skipped # (not failed) so the leaver's non-prod assignments in the same # integration still get swept. try: await ensure_prod_scope_writable(session, scope) except ProdCloudLockedError: base.outcome = OUTCOME_SKIPPED_PROD_LOCKED return base # (3) Already-tracked grant → owned by the standard offboarding # RES_CLOUD_ACCESS_GRANT revoke path. Leaving it here avoids a # double cloud delete + a duplicate revoked-history row. tracked = (await session.execute( select(CmdbCloudAccessGrant.id).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(person.id), CmdbCloudAccessGrant.role_key == assignment.role_key, CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ).limit(1) )).first() if tracked is not None: base.outcome = OUTCOME_SKIPPED_TRACKED_GRANT return base external_ref = ( assignment.external_ref or f"{scope.external_id}::{assignment.role_key}::" f"{assignment.principal_id}" ) if dry_run: base.outcome = OUTCOME_WOULD_DELETE return base ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) orphan = OrphanAssignment( principal_id=assignment.principal_id, role_key=assignment.role_key, scope=ref, role_display_name=assignment.role_display_name, external_ref=external_ref, ) try: del_result = await driver.delete_orphan_assignment(ref, orphan) except Exception as exc: # noqa: BLE001 base.outcome = OUTCOME_ERROR base.error_message = str(exc) return base if not del_result.ok: base.outcome = OUTCOME_ERROR base.error_message = ( del_result.error_message or "driver returned ok=False" ) return base # Cloud-side delete succeeded — record the revoked-grant history row # (with the REAL person_id, unlike the after-the-fact orphan sweep) # + audit, then commit per-assignment so one late DB failure can't # roll back the assignments already deleted on the cloud. now = datetime.now(UTC).replace(tzinfo=None) grant_id: str | None = None # Capture the scope's non-PK column into a plain local BEFORE the # commit/rollback below. ``session.rollback()`` in the except branch # expires every persistent instance, so reading ``scope.external_id`` # afterwards would trigger SQLAlchemy's sync lazy-refresh path and # raise ``MissingGreenlet`` FROM INSIDE the error handler — aborting # the whole per-person sweep instead of soft-continuing. (Same class # as the sync.py:569 remedy.) scope_external_id = scope.external_id try: # Idempotent history write. ``CmdbCloudAccessGrant`` carries TWO # unique keys: # uq_cmdb_cloud_grants_scope_person_role (scope_id, person_id, role_key) # uq_cmdb_cloud_grants_scope_role_extprincipal (scope_id, role_key, external_principal_id) # A blind INSERT with the REAL person_id collides with EITHER a prior # grant-lifecycle row for this (scope, person, role) — e.g. a # previously-REVOKED Helix grant, the PR's own "re-added via console # after a prior grant" case — OR an ``orphan_cleanup`` history row # (person_id=NULL, same principal_id). Both are exactly the residue # this sweep targets, so UPDATE the existing row in place (upgrading a # NULL person_id to the real, attributable one) instead of inserting a # duplicate. Without this the flush raises IntegrityError → the caught # branch below logs a false CRITICAL "manual reconciliation required" # and the module's documented re-run idempotency breaks. existing = (await session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.role_key == assignment.role_key, or_( CmdbCloudAccessGrant.person_id == str(person.id), CmdbCloudAccessGrant.external_principal_id == assignment.principal_id, ), ).limit(1) )).scalar_one_or_none() if existing is not None: existing.person_id = str(person.id) existing.role_display_name = assignment.role_display_name existing.status = GRANT_STATUS_REVOKED existing.revoked_at = now existing.revoked_by_person_id = ( str(actor_user_id) if actor_user_id else None ) existing.revoke_reason = GRANT_REVOKE_REASON_LEAVER_PRINCIPAL_SWEPT existing.external_ref = external_ref existing.external_principal_id = assignment.principal_id existing.external_principal_email = person.work_email # Preserve provenance: the ``existing`` row may be a # previously-REVOKED *Helix-issued* grant (source='helix') — # the step-3 guard only skips active/pending tracked grants, # not revoked ones. Overwriting to "discovered" would erase # the fact this access originated from a Helix grant. Only # rows we truly discovered cloud-side get "discovered". if existing.source != "helix": existing.source = "discovered" existing.detail = { "swept_from": "leaver_sso_sweep", "trace_id": trace_id, "already_absent": bool(del_result.already_absent), } grant = existing else: grant = CmdbCloudAccessGrant( scope_id=str(scope.id), person_id=str(person.id), role_key=assignment.role_key, role_display_name=assignment.role_display_name, status=GRANT_STATUS_REVOKED, granted_at=now, revoked_at=now, revoked_by_person_id=( str(actor_user_id) if actor_user_id else None ), revoke_reason=GRANT_REVOKE_REASON_LEAVER_PRINCIPAL_SWEPT, external_ref=external_ref, external_principal_id=assignment.principal_id, external_principal_email=person.work_email, source="discovered", detail={ "swept_from": "leaver_sso_sweep", "trace_id": trace_id, "already_absent": bool(del_result.already_absent), }, ) session.add(grant) await session.flush() grant_id = str(grant.id) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="cloud.grant.leaver_principal_swept", status="success", resource_type="cmdb_cloud_access_grant", resource_id=grant_id, detail={ "integration_id": str(integration.id), "provider": integration.provider, "scope_id": str(scope.id), "scope_external_id": scope.external_id, "role_key": assignment.role_key, "principal_id": assignment.principal_id, "person_id": str(person.id), "email": person.work_email, "already_absent": bool(del_result.already_absent), }, ) await session.commit() except Exception as exc: # noqa: BLE001 # Cloud-side delete already happened; Helix bookkeeping failed. # Loud log for manual reconciliation, then keep going (same # contract as orphan_cleanup). try: await session.rollback() except Exception: # noqa: BLE001 pass logger.critical( "leaver-sweep: cloud-side delete SUCCEEDED but Helix history " "write FAILED — manual reconciliation required. integration=%s " "scope_external_id=%s role_key=%s principal_id=%s person=%s " "trace_id=%s err=%s", integration.id, scope_external_id, assignment.role_key, assignment.principal_id, person.id, trace_id, exc, ) base.outcome = OUTCOME_ERROR base.error_message = ( "cloud delete succeeded but Helix history write failed — " f"see logs for trace {trace_id}" ) return base base.outcome = ( OUTCOME_ALREADY_ABSENT if del_result.already_absent else OUTCOME_DELETED ) base.grant_id = grant_id return base