"""SSO Apps catalog: admin-managed CRUD + grant-approval executor. Pivoted 2026-05-07 (see migration 127 + memory `feedback_verify_third_party_api_direction.md`): the original design assumed a Google API existed to sync the "Apps → Web and mobile apps" SAML catalog. It doesn't. So this module no longer talks to Google at all — the catalog is admin-managed via the Helix UI / CSV import, and the SSO grant approval flow's "auto-onboard new app" step writes a LOCAL row + notifies IT, who then manually creates the app in Google admin and (optionally) back-fills the ``external_id`` here. Three operations live here: 1. **``import_apps_from_csv``** — admin-triggered bulk import from a CSV blob. Each row becomes one ``sso_apps`` entry tagged ``source='admin_added'``. Idempotent on ``(provider, display_name)`` — re-running with the same CSV updates rather than duplicates. 2. **``execute_pending_google_grant``** — invoked by the 5-minute cron AND optionally inline from admin UI. For each approved ``sso_grants`` row whose ``created_app_id`` is NULL: parses the SAML metadata (XML or three discrete fields), inserts the catalog row tagged ``helix_created``, links it back via ``grant.created_app_id``, and sends a notification to IT admins asking them to manually create the matching SAML app in Google admin. NEVER calls Google. 3. **``set_app_idp_config``** (2026-08-18) — the RETURN leg. After IT creates the app in Google admin, Google has generated the IdP-side SSO URL / Entity ID / certificate, and the requester cannot configure their vendor side without all three. This records them and (on ``deliver=True``) notifies the requester, stamping ``config_delivered_at`` only once a dispatch has actually succeeded. Before it existed the hop had no storage and no notification, so it survived purely in IT's memory while a stalled request rendered as done. """ from __future__ import annotations import csv import io import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, format_exception_for_log from app.core.txn import best_effort from app.domains.sso.models import ( SSO_APP_SOURCE_ADMIN_ADDED, SSO_APP_SOURCE_HELIX_CREATED, SSO_GRANT_STATUS_APPROVED, SsoApp, SsoGrant, ) from app.domains.sso.saml_metadata import ( SamlMetadataParseError, parse_saml_metadata, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # ── Admin CRUD helpers ───────────────────────────────────────────── # Sentinel used by ``upsert_app`` to discriminate "caller did not # specify this field" from "caller specified None to clear". Python's # default-argument-of-None fails this discrimination — the existing # row's value would be silently preserved when the admin meant to # clear it. CSV import always passes positional kwargs, so it's # unaffected; the HTTP PATCH path bypasses this function via # direct setattr() so it can clear too. Kept here for the executor # path which uses upsert_app's update branch. class _Unset: pass _UNSET: Any = _Unset() async def upsert_app( db: AsyncSession, *, display_name: str, platform: str | None = None, # ``None`` for platform / auth_type means "caller didn't specify": # the update branch preserves the existing value (``x or # existing.x``), the insert branch falls back to the model # defaults. Callers that DO pass a value keep full overwrite # semantics. (Launch audit 2026-06-10 — the MCP twin's omitted # args were resetting platform/auth_type on update-in-place.) auth_type: str | None = None, user_access_state: str | None | Any = _UNSET, certificate_expires_at: datetime | None | Any = _UNSET, acs_url: str | None | Any = _UNSET, entity_id: str | None | Any = _UNSET, x509_cert: str | None | Any = _UNSET, owner_person_id: str | None | Any = _UNSET, external_id: str | None | Any = _UNSET, source: str = SSO_APP_SOURCE_ADMIN_ADDED, ) -> SsoApp: """Insert-or-update keyed on ``(provider='google_workspace', display_name)``. Used by the manual "Add app" modal, the CSV bulk import, and the SSO grant approval executor. Field semantics: any field whose value is ``_UNSET`` is left unchanged on update (and lands NULL on insert). Pass ``None`` explicitly to CLEAR an existing value. This matters for the HTTP PATCH path's "clear owner" UX — without the sentinel the function couldn't tell "leave alone" from "set to NULL". Caller is responsible for the audit row + commit. """ clean_name = (display_name or "").strip() if not clean_name: raise AppException( code="SSO_APP_NAME_REQUIRED", message="display_name is required.", status_code=400, ) existing = await db.scalar( select(SsoApp).where( SsoApp.provider == "google_workspace", SsoApp.display_name == clean_name, ) ) now = datetime.now(UTC) if existing is not None: existing.platform = platform or existing.platform existing.auth_type = auth_type or existing.auth_type if user_access_state is not _UNSET: existing.user_access_state = user_access_state if certificate_expires_at is not _UNSET: existing.certificate_expires_at = certificate_expires_at if acs_url is not _UNSET: existing.acs_url = acs_url if entity_id is not _UNSET: existing.entity_id = entity_id if x509_cert is not _UNSET: existing.x509_cert = x509_cert if owner_person_id is not _UNSET: existing.owner_person_id = owner_person_id if external_id is not _UNSET: existing.external_id = external_id existing.updated_at = now return existing # Insert path: collapse any _UNSET sentinels to None — new rows # are NULL by default for fields the caller didn't supply. def _v(x: Any) -> Any: return None if x is _UNSET else x row = SsoApp( id=str(uuid.uuid4()), provider="google_workspace", external_id=_v(external_id), display_name=clean_name, # ``None`` / empty platform (MCP omitted-arg path) falls back # to the model default on insert; the update branch above # preserves the existing value instead. platform=platform or "Web", auth_type=auth_type or "SAML", user_access_state=_v(user_access_state), certificate_expires_at=_v(certificate_expires_at), acs_url=_v(acs_url), entity_id=_v(entity_id), x509_cert=_v(x509_cert), owner_person_id=_v(owner_person_id), source=source, ) # Attempt the INSERT inside a SAVEPOINT so a unique-collision rolls back # ONLY this row. The old ``db.rollback()`` on IntegrityError rolled back the # WHOLE session — discarding every row a CSV import had already staged in # the same (single-commit) transaction, i.e. silent data loss reported as # success. Pre-flush prior pending state first so the savepoint captures # only this row (the begin_nested "add-inside" contract; mirrors # IdempotencyService's first-insert-race recovery). await db.flush() try: async with db.begin_nested(): db.add(row) await db.flush() except IntegrityError: # Concurrency: another worker (parallel CSV import / admin # double-click / cron + manual trigger overlap) inserted a # row with the same ``(provider, display_name)`` between our # SELECT above and this INSERT. The DB UNIQUE constraint # added in migration 128 stops the dual-insert; the SAVEPOINT # auto-rolled-back on exit (only THIS row undone — rows staged # earlier in this transaction survive), so we re-SELECT the # winner and treat this call as an UPDATE-in-place against it. # Net effect: still idempotent, just under contention. winner = await db.scalar( select(SsoApp).where( SsoApp.provider == "google_workspace", SsoApp.display_name == clean_name, ) ) if winner is None: # Lost both the INSERT race AND the SELECT — caller can # retry. Returning a fresh AppException so the HTTP / # CSV path surfaces a clean error instead of a generic # 500. raise AppException( code="SSO_APP_CONCURRENT_INSERT_LOST", message=( "Lost a race against another concurrent writer; " "retry the request." ), status_code=409, ) # Apply the same set-if-supplied logic as the UPDATE branch # above so the caller's values still land. platform/auth_type # were previously dropped on this race path (launch audit # 2026-06-10) — mirror the UPDATE branch exactly. winner.platform = platform or winner.platform winner.auth_type = auth_type or winner.auth_type if user_access_state is not _UNSET: winner.user_access_state = user_access_state if certificate_expires_at is not _UNSET: winner.certificate_expires_at = certificate_expires_at if acs_url is not _UNSET: winner.acs_url = acs_url if entity_id is not _UNSET: winner.entity_id = entity_id if x509_cert is not _UNSET: winner.x509_cert = x509_cert if owner_person_id is not _UNSET: winner.owner_person_id = owner_person_id if external_id is not _UNSET: winner.external_id = external_id winner.updated_at = now await db.flush() return winner return row async def import_apps_from_csv( db: AsyncSession, *, csv_text: str, actor_user_id: str | None = None, ) -> dict[str, Any]: """Bulk-import a CSV of SAML apps. Required header columns (case-insensitive, whitespace-trimmed): ``display_name`` (string, required) ``platform`` (Web | iOS | Android, default Web) ``user_access_state`` (ON for everyone | OFF for everyone | Mixed) ``acs_url`` (URL) ``entity_id`` (URL or URN) ``certificate_expires_at`` (ISO-8601 date or datetime) ``owner_email`` (email — resolved to person via identity_persons.work_email) Unknown columns are ignored. Empty cells leave the field NULL. The "owner_email" column triggers a per-row lookup; missing persons land the row with NULL owner + an entry in the ``unresolved_owners`` summary list (the row still imports — admin can fix the owner via the UI later). Returns a summary dict with counts + the list of unresolved owner emails so the UI can surface them in the import-result toast. """ if not csv_text or not csv_text.strip(): raise AppException( code="SSO_APP_CSV_EMPTY", message="CSV is empty.", status_code=400, ) # Strip the UTF-8 BOM () that Excel + some text editors # prepend when saving CSV. Without this, the first column name # lands as "display_name" which fails the case-insensitive # header match below — admin's whole import silently falls # through to "missing display_name" failures. if csv_text.startswith(""): csv_text = csv_text.lstrip("") if len(csv_text) > 1_000_000: # 1 MB cap — same as the SAML metadata XML cap. A real # 34-app CSV is < 5 KB; anything larger is either the wrong # file or someone trying to OOM the importer. raise AppException( code="SSO_APP_CSV_TOO_LARGE", message="CSV must be ≤ 1 MB.", status_code=400, ) reader = csv.DictReader(io.StringIO(csv_text)) if not reader.fieldnames: raise AppException( code="SSO_APP_CSV_NO_HEADER", message="CSV missing header row.", status_code=400, ) # Normalise headers — admins typing in Excel often produce # leading/trailing whitespace + odd casing. header_map = {(h or "").strip().lower(): h for h in reader.fieldnames} def _cell(row: dict[str, str], key: str) -> str | None: actual_header = header_map.get(key) if actual_header is None: return None v = (row.get(actual_header) or "").strip() return v or None # Pre-launch P1 fix (Round 45, CR-503): row-count cap. The 1 MB # body cap allows ~170k rows of minimal "a,Web\n" lines, each # doing 2 DB roundtrips. Without a row cap, a global admin (or # compromised PAT) can DoS the DB inside a single long # transaction. Real CSVs are <50 apps. rows = list(reader) if len(rows) > 1000: raise AppException( code="SSO_APP_CSV_TOO_MANY_ROWS", message=( f"CSV has {len(rows)} rows; cap is 1000. Split the " "import or fix the source file." ), status_code=400, ) inserted = 0 updated = 0 failed: list[dict[str, Any]] = [] unresolved_owners: list[str] = [] audit = AuditService(db) for line_no, raw_row in enumerate(rows, start=2): # line 1 = header # SAVEPOINT-isolate (2026-08-06): one bad row must not poison the # session — without it row N's DB failure made every later row AND # the summary audit write below raise, so a single malformed line # aborted the whole import with a misleading error. async with best_effort( db, label=f"sso app-catalog CSV row {line_no}", log=logger, ) as row_step: display_name = _cell(raw_row, "display_name") if not display_name: failed.append({ "line": line_no, "error": "missing display_name", }) continue platform = _cell(raw_row, "platform") or "Web" user_access = _cell(raw_row, "user_access_state") acs_url = _cell(raw_row, "acs_url") entity_id = _cell(raw_row, "entity_id") cert_expiry_raw = _cell(raw_row, "certificate_expires_at") cert_expiry: datetime | None = None if cert_expiry_raw: cert_expiry = _parse_iso_date(cert_expiry_raw) if cert_expiry is None: failed.append({ "line": line_no, "error": ( "invalid certificate_expires_at " f"({cert_expiry_raw!r}); expected ISO-8601" ), }) continue owner_email = _cell(raw_row, "owner_email") owner_person_id: str | None = None if owner_email: # Exclude soft-deleted persons — work_email is unique and # survives soft-delete, so without this a CSV naming a # departed employee stamps owner_person_id to the tombstone # and the app is never re-owned on the next leaver sweep. person = await db.scalar( select(IdentityPerson).where( IdentityPerson.work_email == owner_email, IdentityPerson.deleted_at.is_(None), ) ) if person is None: unresolved_owners.append(owner_email) else: owner_person_id = str(person.id) existed_before = await db.scalar( select(SsoApp.id).where( SsoApp.provider == "google_workspace", SsoApp.display_name == display_name, ) ) # #1787: honour the ``_UNSET`` sentinel on the CSV path. ``_cell`` # returns None for BOTH an absent column and an empty cell, and # passing None to ``upsert_app`` CLEARS the stored value. So a # follow-up CSV carrying only the ``display_name`` column (to fix # a label) wiped every listed app's acs_url / entity_id / cert # expiry / owner — while reporting "updated: N". A bulk import # only ever SETS provided values; it must never clear on omission # (explicit clearing stays a PATCH-path concern). Pass ``_UNSET`` # for any field the row didn't provide so it's left unchanged. await upsert_app( db, display_name=display_name, platform=platform, user_access_state=user_access if user_access is not None else _UNSET, certificate_expires_at=( cert_expiry if cert_expiry is not None else _UNSET ), acs_url=acs_url if acs_url is not None else _UNSET, entity_id=entity_id if entity_id is not None else _UNSET, owner_person_id=( owner_person_id if owner_person_id is not None else _UNSET ), source=SSO_APP_SOURCE_ADMIN_ADDED, ) if existed_before: updated += 1 else: inserted += 1 if not row_step.ok: logger.error("CSV row %s import failed: %s", line_no, row_step.reason()) failed.append({ "line": line_no, "error": row_step.reason() or "unknown", }) await audit.log( trace_id=f"sso.app_catalog.csv_import.{datetime.now(UTC).isoformat()}", actor_type="user", actor_user_id=uuid.UUID(actor_user_id) if actor_user_id else None, action="sso.app_catalog.csv_import", status="success" if not failed else "partial", resource_type="sso_app", resource_id=None, detail={ "inserted": inserted, "updated": updated, "failed": len(failed), "failures": failed[:20], # cap audit blob — full list in response "unresolved_owners": unresolved_owners, }, ) await db.commit() return { "inserted": inserted, "updated": updated, "failed_count": len(failed), "failures": failed, "unresolved_owners": unresolved_owners, } def _parse_iso_date(s: str) -> datetime | None: """Tolerant ISO-8601 parser — accepts ``2027-09-07``, ``2027-09-07T12:00:00Z``, ``2027-09-07 12:00:00``. Returns timezone-aware UTC datetime or None on parse failure. """ s = s.strip() # Try date-only first (most common case for cert expiries). for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): try: dt = datetime.strptime(s.rstrip("Z"), fmt) return dt.replace(tzinfo=UTC) except ValueError: continue # Last resort — fromisoformat handles offset suffixes. try: dt = datetime.fromisoformat(s.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) return dt except ValueError: return None # ── Grant executor (5-minute cron + admin one-shot trigger) ──────── async def execute_pending_google_grants( db: AsyncSession, ) -> dict[str, Any]: """Walk approved Google SSO grants whose catalog row hasn't landed yet; create the local row + notify IT. Idempotent — see ``execute_grant_for_google``. Cap of 50/run so a one-time backlog doesn't tie up the cron worker. """ rows = list((await db.execute( select(SsoGrant).where( SsoGrant.status == SSO_GRANT_STATUS_APPROVED, SsoGrant.sso_provider == "Google", SsoGrant.created_app_id.is_(None), ).order_by(SsoGrant.granted_at.asc()).limit(50) )).scalars()) summary = { "considered": len(rows), "executed": 0, "failed": 0, "errors": [], } for grant in rows: try: outcome = await execute_grant_for_google(db, grant) if outcome.get("ok"): summary["executed"] += 1 else: summary["failed"] += 1 summary["errors"].append({ "grant_id": grant.id, "error": outcome.get("error"), }) except Exception as e: # noqa: BLE001 logger.exception("execute_grant_for_google %s crashed", grant.id) summary["failed"] += 1 summary["errors"].append({"grant_id": grant.id, "error": format_exception_for_log(e)}) return summary async def execute_grant_for_google( db: AsyncSession, grant: SsoGrant, ) -> dict[str, Any]: """Single-grant executor — creates a LOCAL ``sso_apps`` catalog row (NOT a Google API call; see module docstring) + sends IT a notification asking them to create the matching SAML app in Google admin. Idempotent: ``created_app_id`` already set → skip. Catalog dedup: if a row with the same display_name already exists (admin pre-added the app, or another grant for the same app already executed), reuse it instead of duplicating. Concurrency: the cron + admin manual-trigger endpoint can both invoke this on the same grant. ``SELECT … FOR UPDATE`` on the grant row serialises the two paths so only one execution actually creates the catalog row; the second sees the ``created_app_id`` set and short-circuits to "skipped". Without this lock the dedup-check → INSERT window allowed dual rows with the same display_name (the UNIQUE on ``(provider, external_id)`` doesn't catch helix_created rows because both have ``external_id=NULL``). """ audit = AuditService(db) # Quick-path skip BEFORE acquiring the lock — avoids holding a # row lock for grants that already executed (vast majority on # any given cron tick). if grant.created_app_id is not None: return {"ok": True, "app_id": grant.created_app_id, "skipped": True} if grant.sso_provider != "Google": return { "ok": False, "error": f"non-Google grant (provider={grant.sso_provider!r})", } # Lock + re-fetch. The caller holds a stale snapshot from # ``execute_pending_google_grants``'s SELECT; another worker # may have advanced it between then and now. locked = await db.scalar( select(SsoGrant).where(SsoGrant.id == grant.id).with_for_update() ) if locked is None: return {"ok": False, "error": "grant disappeared"} if locked.created_app_id is not None: # Lost the race; the other worker already created the row. return { "ok": True, "app_id": locked.created_app_id, "skipped": True, "lost_race": True, } grant = locked # Approved-only invariant. The cron caller # (``execute_pending_google_grants``) filters ``status == approved`` # in its query, but the admin manual-trigger endpoint # (``run_grant_executor``) passes an arbitrary grant_id straight # here — without this guard an admin could mint a catalog row and # page IT for a pending / rejected / cancelled grant, bypassing the # approval gate. Checked under the row lock so a concurrent # cancel/reject can't slip through. Fixed 2026-06-19 (self-service # authz review). if grant.status != SSO_GRANT_STATUS_APPROVED: return { "ok": False, "error": f"grant not approved (status={grant.status!r})", } # Materialise SAML fields — XML wins when both are present. acs = grant.acs_url entity = grant.entity_id cert = grant.x509_cert if grant.saml_metadata_xml: try: parsed = parse_saml_metadata(grant.saml_metadata_xml) except SamlMetadataParseError as e: await audit.log( trace_id=f"sso.execute_pending_google_grants.{grant.id}", actor_type="system", action="sso.grant.execute_google.metadata_parse_failed", status="failure", resource_type="sso_grant", resource_id=grant.id, detail={"error": format_exception_for_log(e)}, ) await db.commit() return {"ok": False, "error": f"XML parse failed: {e}"} acs = parsed.acs_url entity = parsed.entity_id if parsed.x509_cert_pem: cert = parsed.x509_cert_pem if not acs or not entity: await audit.log( trace_id=f"sso.execute_pending_google_grants.{grant.id}", actor_type="system", action="sso.grant.execute_google.missing_fields", status="failure", resource_type="sso_grant", resource_id=grant.id, detail={ "has_acs": bool(acs), "has_entity": bool(entity), "has_cert": bool(cert), }, ) await db.commit() return {"ok": False, "error": "missing acs_url / entity_id"} # Catalog-side dedup. existing_app = await db.scalar( select(SsoApp).where( SsoApp.provider == "google_workspace", SsoApp.display_name == grant.system_name, ) ) if existing_app is not None: grant.created_app_id = existing_app.id grant.updated_at = datetime.now(UTC) await audit.log( trace_id=f"sso.execute_pending_google_grants.{grant.id}", actor_type="system", action="sso.grant.execute_google.linked_existing", status="success", resource_type="sso_grant", resource_id=grant.id, detail={ "app_id": existing_app.id, "external_id": existing_app.external_id, "display_name": existing_app.display_name, }, ) await db.commit() return {"ok": True, "app_id": existing_app.id, "linked_existing": True} # Create local catalog row tagged helix_created. app_row = SsoApp( id=str(uuid.uuid4()), provider="google_workspace", external_id=None, # admin back-fills after creating in Google display_name=grant.system_name, platform="Web", auth_type="SAML", user_access_state=None, acs_url=acs, entity_id=entity, x509_cert=cert, owner_person_id=grant.system_owner_person_id, source=SSO_APP_SOURCE_HELIX_CREATED, raw={ "created_via": "sso_grant_request", "grant_id": grant.id, "requester_person_id": grant.person_id, }, ) db.add(app_row) await db.flush() grant.created_app_id = app_row.id grant.updated_at = datetime.now(UTC) # Notify IT — best-effort. A notify failure must not roll back # the catalog write (otherwise the grant would never get its # created_app_id back-filled and the cron would retry forever). try: await _notify_it_to_create_in_google( db, app=app_row, grant=grant, acs=acs, entity=entity, ) notify_ok = True except Exception: # noqa: BLE001 logger.exception( "sso.app.create_pending notify failed for app %s", app_row.id, ) notify_ok = False await audit.log( trace_id=f"sso.execute_pending_google_grants.{grant.id}", actor_type="system", action="sso.grant.execute_google.catalog_row_created", status="success", resource_type="sso_grant", resource_id=grant.id, detail={ "app_id": app_row.id, "display_name": grant.system_name, "owner_person_id": grant.system_owner_person_id, "notify_ok": notify_ok, }, ) await db.commit() return {"ok": True, "app_id": app_row.id, "notify_ok": notify_ok} async def resend_it_notification_for_app( db: AsyncSession, app_id: str, *, actor_user_id: str | None = None, ) -> dict[str, Any]: """Manual resend of the ``sso.app_create_pending`` notification. Use case: the post-grant-approval executor's automatic notify failed (Feishu webhook flap, all global_admins missing IdentityPerson rows, etc.) and IT never got the heads-up. Without this manual path the catalog row's ``created_app_id`` is set, the cron skips it, and IT is permanently in the dark. Wired up via ``POST /sso/apps/{id}/resend-notification`` and the ``resend_sso_app_notification`` MCP tool. Resolves the linked grant from ``sso_grants.created_app_id`` — if no grant links here (admin-added rows), there's no requester / owner context to render, so return an explicit refusal. """ audit = AuditService(db) app = await db.get(SsoApp, app_id) if app is None: raise AppException( code="SSO_APP_NOT_FOUND", message=f"SsoApp {app_id} not found", status_code=404, ) grant = await db.scalar( select(SsoGrant).where(SsoGrant.created_app_id == app.id).limit(1) ) if grant is None: raise AppException( code="SSO_APP_NO_LINKED_GRANT", message=( "This catalog row has no linked sso_grants entry — " "the IT-creation notification only applies to " "rows created by an SSO grant approval. Manually-" "added rows (source=admin_added) don't need this." ), status_code=409, ) acs = app.acs_url or grant.acs_url or "" entity = app.entity_id or grant.entity_id or "" if not acs or not entity: raise AppException( code="SSO_APP_MISSING_FIELDS", message=( "Cannot send notification — ACS URL or Entity ID is " "missing on both the catalog row and the linked grant." ), status_code=409, ) await _notify_it_to_create_in_google( db, app=app, grant=grant, acs=acs, entity=entity, ) await audit.log( trace_id=f"sso.app.resend_notification.{app.id}", actor_type="user" if actor_user_id else "system", actor_user_id=uuid.UUID(actor_user_id) if actor_user_id else None, action="sso.app.resend_notification", status="success", resource_type="sso_app", resource_id=app.id, detail={ "app_display_name": app.display_name, "linked_grant_id": grant.id, }, ) await db.commit() return { "ok": True, "app_id": app.id, "linked_grant_id": grant.id, "notify_summary": (app.raw or {}).get("notify_summary"), } async def _notify_it_to_create_in_google( db: AsyncSession, *, app: SsoApp, grant: SsoGrant, acs: str, entity: str, ) -> None: """Page IT admins to create the matching SAML app in Google admin. Recipients: every active person with the global_admin role — they're the ones who can log into Google admin. Resolved per-call (rather than cached) so a role grant landing right before this fires is honoured. """ from app.auth.models import AuthRole, AuthUser, AuthUserRole from app.core.rbac import GLOBAL_ADMIN_ROLE from app.notifications.service import ( NotificationEnvelope, NotificationService, ) role_row = await db.scalar( select(AuthRole).where(AuthRole.name == GLOBAL_ADMIN_ROLE) ) if role_row is None: logger.warning( "sso.app.create_pending: no %s role configured, skipping notify", GLOBAL_ADMIN_ROLE, ) return admin_users = list((await db.execute( select(AuthUser) .join(AuthUserRole, AuthUserRole.user_id == AuthUser.id) .where(AuthUserRole.role_id == role_row.id) )).scalars()) if not admin_users: logger.warning( "sso.app.create_pending: no global_admin users, skipping notify", ) return notify = NotificationService(db) requester = await db.get(IdentityPerson, uuid.UUID(grant.person_id)) requester_label = ( (requester.display_name or requester.work_email) if requester else grant.person_id ) ctx = { "app_name": app.display_name, "requester_label": requester_label, "acs_url": acs, "entity_id": entity, "has_cert": bool(app.x509_cert), "owner_label": ( grant.system_owner_label or "—" ), "reason": grant.reason or "—", } skipped_admin_emails: list[str] = [] delivered_count = 0 for admin in admin_users: # AuthUser carries the email; the recipient resolver needs # an IdentityPerson for the inbox dispatch. Look up by email. person = await db.scalar( select(IdentityPerson).where( IdentityPerson.work_email == admin.email, IdentityPerson.deleted_at.is_(None), ) ) if person is None or not person.work_email: # Service-account admins (e.g. ``ops-bot@``) typically # have no IdentityPerson row; flagging them so an admin # ops review can decide whether to reassign the role to # a real human or accept silent skip. logger.warning( "sso.app_create_pending: global_admin %s has no " "IdentityPerson row, skipping notify for app %s", admin.email, app.id, ) skipped_admin_emails.append(admin.email) continue try: await notify.dispatch( NotificationEnvelope( recipient=person, title=f"[SSO] 请在 Google admin 创建 SAML 应用: {app.display_name}", body_text="(rendered from sso.app_create_pending template)", channels=["feishu", "email", "inbox"], template_key="sso.app_create_pending", context=ctx, url=f"/admin/sso-apps?highlight={app.id}", related_resource_type="sso_app", related_resource_id=app.id, ), trace_id=f"sso.app_create_pending.{app.id}", ) delivered_count += 1 except Exception: # noqa: BLE001 logger.exception( "sso.app.create_pending dispatch to %s failed", admin.email, ) # Stash recipient-resolution telemetry on the app row's raw blob # so the audit row + UI can surface it. Useful when an admin # asks "why didn't anyone get the notification?" — `raw` carries # the answer without a separate audit query. if app.raw is None: app.raw = {} app.raw = dict(app.raw, notify_summary={ "delivered": delivered_count, "skipped_admin_emails": skipped_admin_emails, "total_admins": len(admin_users), }) async def set_app_idp_config( db: AsyncSession, app_id: str, *, idp_sso_url: str | None, idp_entity_id: str | None, idp_certificate: str | None, deliver: bool, actor_user_id: str | None = None, actor_person_id: str | None = None, trace_id: str | None = None, ) -> dict[str, Any]: """Record the Google-generated IdP config and optionally hand it to the requester. This is the return leg of the Google SSO grant flow. IT creates the custom SAML app in Google admin by hand (Helix has no API for it); Google mints an SSO URL, an Entity ID and a signing certificate; the requester cannot finish their side without all three. Before this existed there was nowhere to put those values and no record of whether anyone had passed them on — the grant read as ``approved`` with a complete catalog row while the integration sat dead. ``deliver=False`` saves the values only (IT pasting them in as they go, before they're ready to notify). ``deliver=True`` additionally notifies the requester and stamps ``config_delivered_at``, which is what makes "IT still owes this person their config" queryable. Delivery requires all three values AND a linked grant — the grant is the only thing that names a requester, so an ``admin_added`` row has nobody to deliver to and is refused explicitly rather than silently marked delivered. Caller owns the commit. """ audit = AuditService(db) app = await db.get(SsoApp, app_id) if app is None: raise AppException( code="SSO_APP_NOT_FOUND", message=f"SsoApp {app_id} not found", status_code=404, ) # Only overwrite what the caller actually supplied — IT commonly # pastes the URL and Entity ID first, then the certificate. if idp_sso_url is not None: app.idp_sso_url = idp_sso_url.strip() or None if idp_entity_id is not None: app.idp_entity_id = idp_entity_id.strip() or None if idp_certificate is not None: app.idp_certificate = idp_certificate.strip() or None app.updated_at = datetime.now(UTC) result: dict[str, Any] = { "ok": True, "app_id": app.id, "delivered": False, "idp_sso_url": app.idp_sso_url, "idp_entity_id": app.idp_entity_id, "has_idp_certificate": bool(app.idp_certificate), "config_delivered_at": ( app.config_delivered_at.isoformat() if app.config_delivered_at else None ), } if not deliver: await audit.log( trace_id=trace_id or f"sso.app.idp_config_set.{app.id}", actor_type="user", actor_user_id=uuid.UUID(actor_user_id) if actor_user_id else None, action="sso.app.idp_config_set", status="success", resource_type="sso_app", resource_id=app.id, detail={ "has_sso_url": bool(app.idp_sso_url), "has_entity_id": bool(app.idp_entity_id), "has_certificate": bool(app.idp_certificate), }, ) return result missing = [ name for name, value in ( ("idp_sso_url", app.idp_sso_url), ("idp_entity_id", app.idp_entity_id), ("idp_certificate", app.idp_certificate), ) if not value ] if missing: raise AppException( code="SSO_APP_IDP_CONFIG_INCOMPLETE", message=( "Cannot deliver an incomplete IdP config — missing " f"{', '.join(missing)}. Google supplies all three on the " "app's setup screen; sending a partial config just makes " "the requester come back." ), status_code=409, ) grant = await db.scalar( select(SsoGrant).where(SsoGrant.created_app_id == app.id).limit(1) ) if grant is None: raise AppException( code="SSO_APP_NO_LINKED_GRANT", message=( "This catalog row has no linked sso_grants entry, so there " "is no requester to deliver the config to. Manually-added " "rows (source=admin_added) have no requester by design — " "save the config without delivering." ), status_code=409, ) requester = await db.get(IdentityPerson, uuid.UUID(grant.person_id)) if requester is None or not requester.work_email: raise AppException( code="SSO_APP_REQUESTER_UNREACHABLE", message=( "The requester has no reachable IdentityPerson / work " "email — deliver the config out of band and record it " "with the save-only path." ), status_code=409, ) delivered = await _notify_requester_of_idp_config( db, app=app, grant=grant, requester=requester, ) if not delivered: raise AppException( code="SSO_APP_DELIVERY_FAILED", message=( "Notification dispatch to the requester failed; the config " "was saved but NOT marked delivered. Retry, or deliver out " "of band." ), status_code=502, ) # Stamp only after a delivery actually succeeded. Stamping first and # letting the notify fail would turn this field into the same kind of # lie it was added to remove. app.config_delivered_at = datetime.now(UTC) app.config_delivered_by_person_id = actor_person_id result["delivered"] = True result["config_delivered_at"] = app.config_delivered_at.isoformat() result["requester_label"] = ( requester.display_name or requester.work_email ) await audit.log( trace_id=trace_id or f"sso.app.idp_config_delivered.{app.id}", actor_type="user", actor_user_id=uuid.UUID(actor_user_id) if actor_user_id else None, action="sso.app.idp_config_delivered", status="success", resource_type="sso_app", resource_id=app.id, detail={ "grant_id": grant.id, "requester_person_id": grant.person_id, "system_name": grant.system_name, }, ) return result async def _notify_requester_of_idp_config( db: AsyncSession, *, app: SsoApp, grant: SsoGrant, requester: IdentityPerson, ) -> bool: """Send the IdP config to the requester. Returns delivery success. The certificate goes in the body verbatim: the requester has to paste it into their SP config, and a fingerprint or a "[attached]" placeholder would just force another round trip. It is a PUBLIC signing certificate, not a secret. """ from app.notifications.service import ( NotificationEnvelope, NotificationService, ) notify = NotificationService(db) try: await notify.dispatch( NotificationEnvelope( recipient=requester, title=( f"[SSO] {grant.system_name} 的 Google SSO 配置信息已就绪" ), body_text="(rendered from sso.idp_config_ready template)", channels=["feishu", "email", "inbox"], template_key="sso.idp_config_ready", context={ "system_name": grant.system_name, "app_name": app.display_name, "idp_sso_url": app.idp_sso_url or "", "idp_entity_id": app.idp_entity_id or "", "idp_certificate": app.idp_certificate or "", }, url="/my/requests", related_resource_type="sso_app", # ``SsoApp.id`` is a String(36) UUID; the envelope field is # typed ``UUID``. SQLAlchemy would coerce the str (which is # why the sibling IT notification gets away with passing one) # but converting here keeps the declared contract honest. related_resource_id=uuid.UUID(app.id), ), trace_id=f"sso.idp_config_ready.{app.id}", ) except Exception: # noqa: BLE001 logger.exception( "sso.idp_config_ready dispatch to requester %s failed for app %s", grant.person_id, app.id, ) return False return True __all__ = [ "execute_grant_for_google", "execute_pending_google_grants", "import_apps_from_csv", "resend_it_notification_for_app", "set_app_idp_config", "upsert_app", ]