"""Cross-vendor external-account registry — write + directory logic. The seat-sync handlers stay pure categorizers; this service is the admin-facing layer that annotates a non-employee email ONCE and serves the unified "外部 / 共用账号" directory (union of every vendor's external holders + the Google orphan inventory). ``google_orphan_accounts`` remains the source of truth for Google account presence + the offboarding ``RES_GOOGLE_ORPHAN_OWNER`` blocker (owner stays there); this registry adds the cross-vendor classification + note that every panel consults. See ``app/domains/asset/models.py::ExternalAccountRegistry``. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.domains.asset.models import ( EXTERNAL_ACCOUNT_CLASSIFICATIONS, EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, ExternalAccountRegistry, SoftwareExternalHolder, ) from app.domains.google_workspace.models import GoogleOrphanAccount from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) def _norm(email: str) -> str: return (email or "").strip().lower() class ExternalAccountRegistryService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) # ── read: cross-vendor annotation lookup (used by reconciliation) ── async def get_annotations_for_emails( self, emails: list[str], ) -> dict[str, ExternalAccountRegistry]: """Return ``{lower_email: registry_row}`` for the given emails. Used by the seat-reconciliation router to annotate holder rows at display time without touching any sync handler. """ norm = {_norm(e) for e in emails if e and e.strip()} if not norm: return {} rows = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email).in_(norm), ) )).scalars().all() return {_norm(r.email): r for r in rows} # ── read: the unified directory ── async def list_directory( self, *, classification: str | None = None, provider: str | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> dict[str, Any]: """One row per distinct external identity across ALL vendors. Presence is derived live (NOT stored): union of ``software_external_holder`` (category external / no_email, unresolved) and ``google_orphan_accounts`` (not soft-deleted). Each row carries its registry classification/note and — for Google emails — the orphan owner/purpose. Identity key: the lowercased email when the holder has one, else the holder_key (``no_email`` rows — e.g. ``feishu-user:``). The registry annotates by the same key either way. """ # 1) vendor presence from software_external_holder. no_email # rows have email=NULL and dedup on holder_key, so the grouping # key coalesces email → holder_key. _identity = func.lower(func.coalesce( func.nullif(SoftwareExternalHolder.email, ""), SoftwareExternalHolder.holder_key, )) holder_rows = (await self.db.execute( select( _identity.label("identity"), SoftwareExternalHolder.provider, func.max(SoftwareExternalHolder.display_name).label("dn"), func.max(SoftwareExternalHolder.last_observed_at).label("seen"), # An identity is "no_email" iff it has no email anywhere # (min over the group: any row with an email wins). func.min( func.coalesce( func.nullif(SoftwareExternalHolder.email, ""), "", ) ).label("any_email"), func.max( SoftwareExternalHolder.vendor_account_id, ).label("vendor_account_id"), ) .where( SoftwareExternalHolder.category.in_( ("external", "no_email"), ), SoftwareExternalHolder.resolved == False, # noqa: E712 ) .group_by(_identity, SoftwareExternalHolder.provider) )).all() directory: dict[str, dict[str, Any]] = {} for identity, prov, dn, seen, any_email, vendor_account_id in ( holder_rows ): if not identity: continue d = directory.setdefault( identity, {"email": any_email or None, "display_name": None, "providers": set(), "last_seen": None, "google_owner_id": None, "google_purpose": None, "google_orphan_id": None, "vendor_account_id": None}, ) if prov: d["providers"].add(prov) if dn and not d["display_name"]: d["display_name"] = dn if vendor_account_id and not d["vendor_account_id"]: d["vendor_account_id"] = vendor_account_id if seen and (d["last_seen"] is None or seen > d["last_seen"]): d["last_seen"] = seen # 2) Google orphan inventory (presence='google' + owner/purpose) orphan_rows = (await self.db.execute( select(GoogleOrphanAccount).where( GoogleOrphanAccount.deleted_at.is_(None), ) )).scalars().all() for o in orphan_rows: email = _norm(o.primary_email) d = directory.setdefault( email, {"email": email, "display_name": None, "providers": set(), "last_seen": None, "google_owner_id": None, "google_purpose": None, "google_orphan_id": None, "vendor_account_id": None}, ) d["providers"].add("google") if o.display_name and not d["display_name"]: d["display_name"] = o.display_name if o.last_seen_at and ( d["last_seen"] is None or o.last_seen_at > d["last_seen"] ): d["last_seen"] = o.last_seen_at d["google_owner_id"] = str(o.owner_user_id) if o.owner_user_id else None d["google_purpose"] = o.purpose d["google_orphan_id"] = str(o.id) # 3) attach registry annotations annos = await self.get_annotations_for_emails(list(directory.keys())) # 4) resolve owner display names (registry owner + google owner) owner_ids: set[uuid.UUID] = set() for email, d in directory.items(): a = annos.get(email) if a and a.owner_person_id: owner_ids.add(a.owner_person_id) if d["google_owner_id"]: owner_ids.add(uuid.UUID(d["google_owner_id"])) owner_names: dict[str, str] = {} if owner_ids: for p in (await self.db.execute( select(IdentityPerson.id, IdentityPerson.display_name) .where(IdentityPerson.id.in_(owner_ids)) )).all(): owner_names[str(p[0])] = p[1] items: list[dict[str, Any]] = [] for identity, d in directory.items(): a = annos.get(identity) cls = a.classification if a else "unknown" reg_owner_id = ( str(a.owner_person_id) if a and a.owner_person_id else None ) items.append({ # Stable annotation/dedup key — the email when the # account has one, else the holder_key (no-email # accounts, e.g. ``feishu-user:``). classify / # delete take this key. "key": identity, "email": d["email"] or None, "no_email": not d["email"], "vendor_account_id": d["vendor_account_id"], "display_name": d["display_name"], "providers": sorted(d["providers"]), "classification": cls, "is_known": cls in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, "note": a.note if a else None, # Google owner stays authoritative for Google emails; # registry owner covers the other vendors. "owner_person_id": d["google_owner_id"] or reg_owner_id, "owner_name": owner_names.get( d["google_owner_id"] or reg_owner_id or "", ), "google_purpose": d["google_purpose"], "google_orphan_id": d["google_orphan_id"], "has_google": "google" in d["providers"], "last_seen_at": ( d["last_seen"].isoformat() if d["last_seen"] else None ), }) # filters (python-side; dataset is the few-hundred external emails) if classification: items = [i for i in items if i["classification"] == classification] if provider: items = [i for i in items if provider in i["providers"]] if search: s = search.strip().lower() items = [ i for i in items if s in i["key"] or (i["display_name"] and s in i["display_name"].lower()) ] items.sort(key=lambda i: (-len(i["providers"]), i["key"])) total = len(items) start = (page - 1) * page_size return { "items": items[start:start + page_size], "total": total, "page": page, "page_size": page_size, } # ── write: classify / annotate (upsert by email) ── async def classify( self, email: str, *, classification: str | None, note: str | None, owner_person_id: uuid.UUID | None, set_owner: bool, trace_id: str, actor_user_id: uuid.UUID, ) -> dict[str, Any]: norm = _norm(email) if not norm: raise AppException("EMAIL_REQUIRED", "email is required", status_code=400) if classification is not None and ( classification not in EXTERNAL_ACCOUNT_CLASSIFICATIONS ): raise AppException( "INVALID_CLASSIFICATION", f"classification must be one of " f"{EXTERNAL_ACCOUNT_CLASSIFICATIONS}", status_code=400, ) row = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == norm, ) )).scalars().first() created = row is None if row is None: # Check-then-insert races the unique email index when two # admin tabs / UI+MCP classify the same key concurrently — # SAVEPOINT the insert and fall back to the winner's row # instead of surfacing a 1062 → 500. from sqlalchemy.exc import IntegrityError row = ExternalAccountRegistry( id=uuid.uuid4(), email=norm, classification="unknown", created_by=actor_user_id, ) try: async with self.db.begin_nested(): self.db.add(row) except IntegrityError: created = False row = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == norm, ) )).scalars().one() if classification is not None: row.classification = classification if note is not None: row.note = note.strip() or None # ``set_owner`` distinguishes "not provided" from "clear to NULL". if set_owner: row.owner_person_id = owner_person_id row.updated_by = actor_user_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="external_account.classified", status="success", resource_type="external_account_registry", resource_id=str(row.id), detail={ "email": norm, "classification": row.classification, "created": created, }, ) return { "email": row.email, "classification": row.classification, "is_known": ( row.classification in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS ), "note": row.note, "owner_person_id": ( str(row.owner_person_id) if row.owner_person_id else None ), } async def ensure_retained_classification( self, email: str, *, classification: str = "external_guest", note: str | None = None, trace_id: str, actor_user_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Idempotently ensure a KEEP/retention row exists for ``email``, WITHOUT clobbering an admin's explicit decision. Used by offboarding when a leaver is retained via the live Feishu CBG-department check (``retained_cbg_department``): that path has no registry row yet, so #1983's leaver-assurance (which exempts leavers whose ``work_email`` carries a KNOWN registry classification) would otherwise page IT daily for a documented keep. Writing the row makes the registry the shared contract both sides read. Non-clobber rules (mirrors ``classify``'s upsert + race-safety): * absent → insert with ``classification`` + ``note`` * already KNOWN → leave untouched (never downgrade an admin's ``shared_service`` / ``contractor`` to ``external_guest``) * ``unknown`` / NULL → upgrade to ``classification`` (a documented retention now applies to an untriaged row) * other (e.g. ``former_employee``) → leave untouched (an explicit ghost-account decision is not silently overridden here) Never raises on a benign classification value. Returns ``{"email", "classification", "created", "updated"}``. """ norm = _norm(email) if not norm: raise AppException("EMAIL_REQUIRED", "email is required", status_code=400) if classification not in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS: raise AppException( "INVALID_CLASSIFICATION", f"retention classification must be one of " f"{EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS}", status_code=400, ) row = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == norm, ) )).scalars().first() created = False updated = False if row is None: from sqlalchemy.exc import IntegrityError row = ExternalAccountRegistry( id=uuid.uuid4(), email=norm, classification=classification, note=(note.strip() or None) if note else None, created_by=actor_user_id, updated_by=actor_user_id, ) try: # SAVEPOINT the insert so a concurrent writer racing the # unique email index (another offboarding tick / an admin # classify) surfaces as the winner's row, not a 1062 → 500. async with self.db.begin_nested(): self.db.add(row) created = True except IntegrityError: row = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == norm, ) )).scalars().one() if not created: # Only upgrade an untriaged row; never clobber a KNOWN keep or an # explicit action classification (e.g. former_employee). if row.classification in (None, "unknown"): row.classification = classification if note and not row.note: row.note = note.strip() or None row.updated_by = actor_user_id updated = True await self.db.flush() if created or updated: await self.audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="external_account.retention_recorded", status="success", resource_type="external_account_registry", resource_id=str(row.id), detail={ "email": norm, "classification": row.classification, "created": created, "updated": updated, }, ) return { "email": row.email, "classification": row.classification, "created": created, "updated": updated, } async def delete( self, email: str, *, trace_id: str, actor_user_id: uuid.UUID, ) -> None: norm = _norm(email) row = (await self.db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == norm, ) )).scalars().first() if row is None: return await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="external_account.classification_cleared", status="success", resource_type="external_account_registry", resource_id=str(row.id), detail={"email": norm}, )