"""HTTP surface for the SSO Apps catalog (admin-only). Endpoints: GET /api/v1/sso/apps — list catalog POST /api/v1/sso/apps — create one app PATCH /api/v1/sso/apps/{id} — update fields DELETE /api/v1/sso/apps/{id} — delete one app POST /api/v1/sso/apps/import-csv — bulk import GET /api/v1/sso/apps/{id}/grants — list linked grants POST /api/v1/sso/apps/{id}/idp-config — record + deliver IdP config POST /api/v1/sso/apps/grants/{id}/run — re-run grant exec The catalog is fully admin-managed (see migration 127 docstring + memory `feedback_verify_third_party_api_direction.md` — Google has no public API for the outbound SAML apps catalog, so admin enters the 34 existing apps either via the create endpoint, the bulk-CSV endpoint, or by letting the SSO grant approval flow auto-create new ones via `helix_created`). """ from __future__ import annotations import uuid from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Request from pydantic import BaseModel, Field from sqlalchemy import func as _func from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import require_admin, require_admin_read from app.core.response import success_response from app.domains.sso.app_catalog_service import ( execute_grant_for_google, import_apps_from_csv, resend_it_notification_for_app, set_app_idp_config, upsert_app, ) from app.domains.sso.models import ( SSO_APP_SOURCE_ADMIN_ADDED, SSO_GRANT_STATUS_APPROVED, SsoApp, SsoGrant, ) from app.identity.person_models import IdentityPerson router = APIRouter( prefix="/api/v1/sso/apps", tags=["sso", "sso-apps"], ) # ── Schemas ──────────────────────────────────────────────────────── class SsoAppCreateBody(BaseModel): display_name: str = Field(min_length=1, max_length=255) platform: str = Field(default="Web", max_length=64) auth_type: str = Field(default="SAML", max_length=32) user_access_state: str | None = Field(default=None, max_length=32) certificate_expires_at: datetime | None = None acs_url: str | None = Field(default=None, max_length=1024) entity_id: str | None = Field(default=None, max_length=1024) x509_cert: str | None = Field(default=None, max_length=20_000) owner_person_id: str | None = Field(default=None, max_length=36) external_id: str | None = Field(default=None, max_length=255) class SsoAppUpdateBody(BaseModel): display_name: str | None = Field(default=None, max_length=255) platform: str | None = Field(default=None, max_length=64) auth_type: str | None = Field(default=None, max_length=32) user_access_state: str | None = Field(default=None, max_length=32) certificate_expires_at: datetime | None = None acs_url: str | None = Field(default=None, max_length=1024) entity_id: str | None = Field(default=None, max_length=1024) x509_cert: str | None = Field(default=None, max_length=20_000) # Nullable so admins can clear an owner. owner_person_id: str | None = Field(default=None, max_length=36) external_id: str | None = Field(default=None, max_length=255) class SsoAppCsvImportBody(BaseModel): csv_text: str = Field(min_length=1, max_length=1_000_000) class SsoAppIdpConfigBody(BaseModel): """The values Google generated when IT created the SAML app. Deliberately NOT part of ``SsoAppUpdateBody``: delivery has to stamp ``config_delivered_at`` only when a notification actually went out, and folding that into the generic PATCH would let a field edit claim a delivery that never happened. """ idp_sso_url: str | None = Field(default=None, max_length=1024) idp_entity_id: str | None = Field(default=None, max_length=1024) # Same 20 KB ceiling as the SP-side cert on the grant. idp_certificate: str | None = Field(default=None, max_length=20_000) # ``False`` saves the values only; ``True`` also notifies the # requester and stamps the delivery. deliver: bool = False def _serialize( app: SsoApp, owner: IdentityPerson | None, grant_count: int, ) -> dict[str, Any]: return { "id": app.id, "provider": app.provider, "external_id": app.external_id, "display_name": app.display_name, "platform": app.platform, "auth_type": app.auth_type, "user_access_state": app.user_access_state, "certificate_expires_at": ( app.certificate_expires_at.isoformat() if app.certificate_expires_at else None ), "acs_url": app.acs_url, "entity_id": app.entity_id, "x509_cert": app.x509_cert, # IdP side — what Google generated, and whether it reached the # requester. The certificate is a public signing cert, so it is # returned in full (the admin has to copy it out to hand over). "idp_sso_url": app.idp_sso_url, "idp_entity_id": app.idp_entity_id, "idp_certificate": app.idp_certificate, "config_delivered_at": ( app.config_delivered_at.isoformat() if app.config_delivered_at else None ), "config_delivered_by_person_id": app.config_delivered_by_person_id, "owner_person_id": app.owner_person_id, "owner_label": ( (owner.display_name or owner.work_email) if owner else None ), "source": app.source, "last_synced_at": ( app.last_synced_at.isoformat() if app.last_synced_at else None ), "created_at": app.created_at.isoformat(), "updated_at": app.updated_at.isoformat(), "grant_count": grant_count, } async def _hydrate_owner( db: AsyncSession, owner_id: str | None, ) -> IdentityPerson | None: if not owner_id: return None try: return await db.get(IdentityPerson, uuid.UUID(owner_id)) except (ValueError, TypeError): return None # ── Endpoints ────────────────────────────────────────────────────── @router.get("") async def list_sso_apps( request: Request, _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), source: str | None = Query(None, pattern="^(admin_added|helix_created)$"), expiring_within_days: int | None = Query(None, ge=1, le=3650), ): """List the SSO Apps catalog. Filters: ``source`` (provenance — admin_added vs helix_created), ``expiring_within_days`` (cert expiry highlight).""" q = select(SsoApp).where(SsoApp.provider == "google_workspace") if source: q = q.where(SsoApp.source == source) if expiring_within_days: from datetime import timedelta cutoff = datetime.now(UTC) + timedelta(days=expiring_within_days) q = q.where(SsoApp.certificate_expires_at <= cutoff) count_q = select(_func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = ( q.order_by(SsoApp.display_name.asc()) .offset((page - 1) * page_size) .limit(page_size) ) rows = list((await db.execute(q)).scalars()) owner_ids = {r.owner_person_id for r in rows if r.owner_person_id} owners: dict[str, IdentityPerson] = {} if owner_ids: person_rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in owner_ids]), ) )).scalars().all() owners = {str(p.id): p for p in person_rows} grant_counts: dict[str, int] = {} if rows: grant_count_rows = (await db.execute( select(SsoGrant.created_app_id, _func.count()) .where(SsoGrant.created_app_id.in_([r.id for r in rows])) .group_by(SsoGrant.created_app_id) )).all() grant_counts = {row[0]: int(row[1]) for row in grant_count_rows} payload = [ _serialize( r, owners.get(r.owner_person_id) if r.owner_person_id else None, grant_counts.get(r.id, 0), ) for r in rows ] return success_response( data=payload, request=request, meta={"total": total, "page": page, "page_size": page_size}, ) @router.post("", status_code=201) async def create_sso_app( body: SsoAppCreateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Admin manually adds one SAML app to the catalog. Idempotent on display_name — re-posting with the same name updates the existing row in place.""" owner_id_str: str | None = None if body.owner_person_id: try: owner_uuid = uuid.UUID(body.owner_person_id) except (ValueError, TypeError): raise HTTPException( status_code=400, detail="owner_person_id is not a UUID", ) from None owner_person = await db.get(IdentityPerson, owner_uuid) if owner_person is None: raise HTTPException(status_code=404, detail="owner not found") owner_id_str = str(owner_person.id) app = await upsert_app( db, display_name=body.display_name, platform=body.platform, auth_type=body.auth_type, user_access_state=body.user_access_state, certificate_expires_at=body.certificate_expires_at, acs_url=body.acs_url, entity_id=body.entity_id, x509_cert=body.x509_cert, owner_person_id=owner_id_str, external_id=body.external_id, source=SSO_APP_SOURCE_ADMIN_ADDED, ) # Round-49: audit catalog mutations from the HTTP path. The MCP # twins already audit via tool_runtime; the human-driven web UI # path was silent — forensic queries for "who created/changed # this SSO app" returned nothing for browser admins. await AuditService(db).log( trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()), actor_type="user", actor_user_id=user.id, action="sso.app.create", status="success", resource_type="sso_app", resource_id=str(app.id), detail={ "display_name": app.display_name, "platform": app.platform, "owner_person_id": owner_id_str, "source": SSO_APP_SOURCE_ADMIN_ADDED, }, ) await db.commit() await db.refresh(app) owner = await _hydrate_owner(db, app.owner_person_id) return success_response( data=_serialize(app, owner, grant_count=0), request=request, ) @router.patch("/{app_id}") async def update_sso_app( app_id: str, body: SsoAppUpdateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Update any subset of an app's fields. Pass ``owner_person_id`` explicitly as null in the JSON body to clear the owner; omit the field entirely to leave it unchanged.""" app = await db.get(SsoApp, app_id) if app is None: raise HTTPException(status_code=404, detail="SsoApp not found") payload = body.model_dump(exclude_unset=True) # Snapshot before-state for audit (Round-49). before_owner = app.owner_person_id before_keys = {k: getattr(app, k, None) for k in payload} # Owner needs a real-person lookup to validate the UUID. if "owner_person_id" in payload: new_owner = payload["owner_person_id"] if new_owner: try: owner_uuid = uuid.UUID(new_owner) except (ValueError, TypeError): raise HTTPException( status_code=400, detail="owner_person_id is not a UUID", ) from None owner_person = await db.get(IdentityPerson, owner_uuid) if owner_person is None: raise HTTPException(status_code=404, detail="owner not found") app.owner_person_id = str(owner_person.id) else: app.owner_person_id = None payload.pop("owner_person_id") for k, v in payload.items(): setattr(app, k, v) app.updated_at = datetime.now(UTC) # Round-49: per-field audit on PATCH — owner rebinds are the # catastrophic-if-stolen-PAT case the MCP twin already audits. await AuditService(db).log( trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()), actor_type="user", actor_user_id=user.id, action="sso.app.update", status="success", resource_type="sso_app", resource_id=str(app.id), detail={ "display_name": app.display_name, "before": { "owner_person_id": before_owner, **{k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in before_keys.items()}, }, "after": { "owner_person_id": app.owner_person_id, **{k: ( getattr(app, k).isoformat() if hasattr(getattr(app, k, None), "isoformat") else getattr(app, k, None) ) for k in before_keys}, }, }, ) await db.commit() await db.refresh(app) owner = await _hydrate_owner(db, app.owner_person_id) return success_response( data=_serialize(app, owner, grant_count=0), request=request, ) @router.delete("/{app_id}", status_code=204) async def delete_sso_app( app_id: str, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Hard-delete an app from the catalog. Linked sso_grants rows have ``ON DELETE SET NULL`` on ``created_app_id`` so grant history survives — only the catalog entry goes away.""" app = await db.get(SsoApp, app_id) if app is None: raise HTTPException(status_code=404, detail="SsoApp not found") # Resurrection guard: ``created_app_id`` is ``ON DELETE SET NULL``, # so deleting an app that a still-*approved* grant points at nulls # that grant's ``created_app_id`` → the cron # ``execute_pending_google_grants`` (status==approved AND # created_app_id IS NULL) immediately re-creates the catalog row and # re-pages IT on the next tick, undoing the delete in a loop. Refuse # the delete while any approved grant still references this app; the # admin must revoke/cancel those grants first. Grants in terminal # states (rejected/cancelled/revoked) are never re-selected, so they # don't block. Fixed 2026-06-19 (self-service authz review). live_grant_count = await db.scalar( select(_func.count()) .select_from(SsoGrant) .where( SsoGrant.created_app_id == app.id, SsoGrant.status == SSO_GRANT_STATUS_APPROVED, ) ) if live_grant_count: raise HTTPException( status_code=409, detail=( "Cannot delete: this app still has approved SSO grants " "linked to it (the catalog row would be auto-recreated). " "Revoke or cancel those grants first." ), ) # Round-49: capture display_name BEFORE delete so the audit row # carries a human-readable label after the row is gone. snapshot = { "display_name": app.display_name, "platform": app.platform, "owner_person_id": app.owner_person_id, "external_id": app.external_id, "source": app.source, } await AuditService(db).log( trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()), actor_type="user", actor_user_id=user.id, action="sso.app.delete", status="success", resource_type="sso_app", resource_id=str(app.id), detail=snapshot, ) await db.delete(app) await db.commit() @router.post("/import-csv") async def import_sso_apps_csv( body: SsoAppCsvImportBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Bulk-import a CSV of SAML apps. See :func:`app_catalog_service.import_apps_from_csv` for column spec + error semantics.""" summary = await import_apps_from_csv( db, csv_text=body.csv_text, actor_user_id=str(user.id), ) return success_response(data=summary, request=request) @router.get("/{app_id}/grants") async def list_app_grants( app_id: str, request: Request, _user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """List every ``sso_grants`` row that links to this catalog app.""" rows = list((await db.execute( select(SsoGrant) .where(SsoGrant.created_app_id == app_id) .order_by(SsoGrant.created_at.desc()) .limit(200) )).scalars()) person_ids = {r.person_id for r in rows} persons: dict[str, IdentityPerson] = {} if person_ids: person_rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in person_ids]), ) )).scalars().all() persons = {str(p.id): p for p in person_rows} payload = [] for r in rows: holder = persons.get(r.person_id) payload.append({ "grant_id": r.id, "person_id": r.person_id, "person_label": ( (holder.display_name or holder.work_email) if holder else None ), "status": r.status, "granted_at": r.granted_at.isoformat() if r.granted_at else None, "revoked_at": r.revoked_at.isoformat() if r.revoked_at else None, }) return success_response(data=payload, request=request) @router.post("/{app_id}/resend-notification") async def resend_app_notification( app_id: str, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Manually re-send the ``sso.app_create_pending`` notification for one ``helix_created`` catalog row. Used when the executor's automatic notify failed (Feishu webhook flap, all global_admins missing IdentityPerson rows). Refused for ``admin_added`` rows (no linked grant → no template context to render).""" outcome = await resend_it_notification_for_app( db, app_id, actor_user_id=str(user.id), ) return success_response(data=outcome, request=request) @router.post("/{app_id}/idp-config") async def set_app_idp_config_endpoint( app_id: str, body: SsoAppIdpConfigBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Record the Google-generated IdP config, optionally delivering it. ``deliver=false`` — save what IT has so far (they usually paste the URL and Entity ID before the certificate). ``deliver=true`` — additionally notify the requester and stamp ``config_delivered_at``. Requires all three values plus a linked grant; refuses (409) rather than marking an incomplete or requester-less config as delivered. """ actor_person_id: str | None = None person = await db.scalar( select(IdentityPerson).where( IdentityPerson.work_email == user.email, IdentityPerson.deleted_at.is_(None), ) ) if person is not None: actor_person_id = str(person.id) outcome = await set_app_idp_config( db, app_id, idp_sso_url=body.idp_sso_url, idp_entity_id=body.idp_entity_id, idp_certificate=body.idp_certificate, deliver=body.deliver, actor_user_id=str(user.id), actor_person_id=actor_person_id, trace_id=getattr(request.state, "trace_id", None), ) await db.commit() return success_response(data=outcome, request=request) @router.post("/grants/{grant_id}/run") async def run_grant_executor( grant_id: str, request: Request, _user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Re-trigger the catalog-row creation + IT notification for one grant. Idempotent — if ``created_app_id`` is set, skip.""" grant = await db.get(SsoGrant, grant_id) if grant is None: raise HTTPException(status_code=404, detail="SsoGrant not found") outcome = await execute_grant_for_google(db, grant) return success_response(data=outcome, request=request) __all__ = ["router"]