"""SSO grant + SSO app catalog ORM models. * :class:`SsoGrant` — see migration 099_sso_grants for the original schema. Migration 126_sso_apps_catalog (2026-05-07) added the SAML metadata columns + ``created_app_id`` FK for the Google SAML-app auto-create flow. * :class:`SsoApp` — local catalog of every Google Workspace SAML app (one row per Cloud Identity ``inboundSamlSsoProfile``). Populated by the daily ``sso.google_saml_app_sync`` cron AND by the SSO grant approval flow when an admin onboards a brand-new app via Helix. """ from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, Text, func from sqlalchemy.dialects.mysql import MEDIUMTEXT from sqlalchemy.orm import Mapped, mapped_column from app.core.database import Base from app.core.sql_types import GUID # State machine constants — keep aligned with the values written by # SsoGrantService and the Feishu callback handler. SSO_GRANT_STATUS_PENDING = "pending" SSO_GRANT_STATUS_APPROVED = "approved" SSO_GRANT_STATUS_REJECTED = "rejected" SSO_GRANT_STATUS_CANCELLED = "cancelled" SSO_GRANT_STATUS_REVOKED = "revoked" # ``SsoGrant.removal_status`` — the SECOND half of a revoke (migration 603). # # ``status='revoked'`` records the DECISION ("Helix says this access is # gone"). Nothing about that decision turns anything off in Google admin / # Atlassian / Okta — SSO removal is a human action in a vendor console. The # removal lifecycle tracks whether that action happened and whether the # vendor agrees: # # not_applicable grant is not revoked (pending / approved / rejected / # cancelled) — the column is inert # pending revoked in Helix; a manual_notify provisioning task is # open for the system owner (or IT) to disable it # confirmed a human completed that task WITH evidence # (``removal_external_ref``: ticket / console URL / note) # verified vendor evidence agrees — Google Admin Reports shows no # SAML login for this (user, app) since the revoke, for # at least ``sso.removal.verified_after_days`` after the # human confirmation # contradicted vendor evidence DISAGREES — a SAML login was observed # after ``revoked_at``; the task is re-opened and the # system owner + IT are paged with the login timestamp # unverified_legacy revoked before migration 603 with no task ever opened. # A distinct, visible state (not a backfilled ``pending`` # — that would claim a task exists) that admins clear by # opening a review task from the ledger. REMOVAL_NOT_APPLICABLE = "not_applicable" REMOVAL_PENDING = "pending" REMOVAL_CONFIRMED = "confirmed" REMOVAL_VERIFIED = "verified" REMOVAL_CONTRADICTED = "contradicted" REMOVAL_UNVERIFIED_LEGACY = "unverified_legacy" REMOVAL_STATUSES: tuple[str, ...] = ( REMOVAL_NOT_APPLICABLE, REMOVAL_PENDING, REMOVAL_CONFIRMED, REMOVAL_VERIFIED, REMOVAL_CONTRADICTED, REMOVAL_UNVERIFIED_LEGACY, ) # Statuses where the vendor-side removal is NOT yet established. The # reconcile cron's worklist and the admin ledger's "outstanding" badge both # read this tuple — one definition of "still open". REMOVAL_OPEN_STATUSES: tuple[str, ...] = ( REMOVAL_PENDING, REMOVAL_CONTRADICTED, REMOVAL_UNVERIFIED_LEGACY, ) # SsoApp.source values — provenance badge for the catalog list UI. # # History note (2026-05-07 pivot): the original design assumed a # daily cron would sync from Google's outbound SAML apps API, with # `google_synced` marking those rows. That API doesn't exist (see # memory `feedback_verify_third_party_api_direction.md` and migration # 127 docstring). The catalog is now admin-managed, so: # # * ``admin_added`` — admin entered manually via the "+ 添加应用" # modal, OR via CSV bulk import. Most rows. # * ``helix_created`` — created automatically by the SSO grant # approval flow when an employee submitted SAML metadata for a # new app. Helix inserts the catalog row + sends IT a notification # to manually create the matching app in Google admin (since no # Google API exists for outbound SAML app create either). SSO_APP_SOURCE_ADMIN_ADDED = "admin_added" SSO_APP_SOURCE_HELIX_CREATED = "helix_created" class SsoGrant(Base): """An employee's access grant to an SSO-protected internal system. A row lands here when a Feishu-approved ``sso_grant_request`` settles. The leaver flow scans for rows where ``person_id`` is leaving and demands a handover before allowing the offboarding state machine to advance — keeping the institutional knowledge of "who can sign into Authing for our admin console" attached to a Helix identity at all times. For ``sso_provider='Google'`` requests, post-approval the service layer ALSO calls Google Cloud Identity to create an ``inboundSamlSsoProfile`` from the SAML metadata fields below and writes a matching ``sso_apps`` row, then back-fills ``created_app_id`` here. (Lark grants don't carry SAML metadata and don't trigger any Google-side write — historical behavior.) """ __tablename__ = "sso_grants" id: Mapped[str] = mapped_column( String(36), primary_key=True, default=lambda: str(uuid.uuid4()), ) person_id: Mapped[str] = mapped_column( String(36), ForeignKey("identity_persons.id"), nullable=False, ) sso_provider: Mapped[str] = mapped_column(String(64), nullable=False) system_name: Mapped[str] = mapped_column(String(255), nullable=False) system_owner_label: Mapped[str] = mapped_column(String(255), nullable=False) system_owner_person_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) reason: Mapped[str | None] = mapped_column(Text, nullable=True) status: Mapped[str] = mapped_column( String(32), nullable=False, default=SSO_GRANT_STATUS_PENDING, ) feishu_instance_code: Mapped[str | None] = mapped_column( String(64), nullable=True, ) approval_request_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("approval_requests.id", ondelete="SET NULL"), nullable=True, ) granted_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) revoked_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) revoke_reason: Mapped[str | None] = mapped_column(String(255), nullable=True) handover_from_person_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) handover_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) handover_actor_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) # SAML metadata (Google flow only) — see migration 126. # All four are nullable: Lark grants leave them empty, and even # for Google grants the requester may supply EITHER the full # ``saml_metadata_xml`` (parsed at execute time) OR the three # discrete fields. The submit validator enforces "at least one # complete shape is present" for Google requests. acs_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) entity_id: Mapped[str | None] = mapped_column(String(1024), nullable=True) x509_cert: Mapped[str | None] = mapped_column(Text, nullable=True) saml_metadata_xml: Mapped[str | None] = mapped_column( MEDIUMTEXT, nullable=True, ) # Back-filled after a successful Cloud Identity create. NULL # while the post-approval Google call is in flight or after a # transient failure; the executor retries until this lands. # ON DELETE SET NULL so manual catalog cleanup doesn't cascade- # delete grant history. created_app_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("sso_apps.id", ondelete="SET NULL"), nullable=True, ) # ── Removal lifecycle (migration 603) — see REMOVAL_* above ────── removal_status: Mapped[str] = mapped_column( String(24), nullable=False, default=REMOVAL_NOT_APPLICABLE, server_default=REMOVAL_NOT_APPLICABLE, ) removal_task_id: Mapped[uuid.UUID | None] = mapped_column( GUID, ForeignKey("provisioning_tasks.id", ondelete="SET NULL"), nullable=True, ) removal_confirmed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) removal_confirmed_by_person_id: Mapped[uuid.UUID | None] = mapped_column( GUID, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) removal_external_ref: Mapped[str | None] = mapped_column( String(512), nullable=True, ) removal_verified_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) removal_last_checked_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # Most recent SAML login observed for (holder, app) at the IdP. Set by # the reconcile cron; ``None`` means "no login observed since revoke". removal_last_login_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # Why the last vendor check could not run (typically a missing Reports # API scope on the DwD service account). Cleared on a successful check. removal_check_error: Mapped[str | None] = mapped_column( String(255), nullable=True, ) removal_reminder_count: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) removal_last_reminded_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False, ) class SsoApp(Base): """One row per Google Workspace SAML SSO app. A Helix-side ledger of the Google admin console "Apps → Web and mobile apps" listing. **Helix never calls Google for any of this** — see the 2026-05-07 pivot note in :mod:`app.domains.sso.app_catalog_service`: Cloud Identity's ``inboundSamlSsoProfiles`` API is the INBOUND direction (third-party IdP → Google), and Google publishes no write API for this outbound catalog. Two ingestion paths populate the table, both local: * **Admin-managed** — the "+ Add app" modal on the SSO Apps page or the CSV bulk import. Source = ``admin_added``. * **SSO grant approval flow** — when an employee's ``sso_grant_request`` is approved with ``sso_provider='Google'``, ``execute_pending_google_grants`` parses the requester's SAML metadata, inserts a row here with the SP-side ACS / Entity / cert, links it back via ``sso_grants.created_app_id`` and notifies IT to create the app in Google admin **by hand**. Source = ``helix_created``; ``external_id`` stays NULL until an admin back-fills the Google-side id for reconciliation. ``owner_person_id`` is Helix-side metadata IT picks via the SSO admin page — used by the leaver flow to ping the right person when an app's owner offboards. **Two directions of SAML metadata live on this row.** ``acs_url`` / ``entity_id`` / ``x509_cert`` are the **SP side** — what the requester (or their vendor) gave us, and what IT types into Google. ``idp_sso_url`` / ``idp_entity_id`` / ``idp_certificate`` are the **IdP side** — what Google generated when IT created the app, and what has to travel back to the requester before the integration can possibly work. ``config_delivered_at`` records that return trip actually happening. """ __tablename__ = "sso_apps" id: Mapped[str] = mapped_column( String(36), primary_key=True, default=lambda: str(uuid.uuid4()), ) # ``google_workspace`` for now; the column exists so a future # Lark / Authing / OneLogin integration can co-tenant the catalog # without a schema change. provider: Mapped[str] = mapped_column( String(64), nullable=False, default="google_workspace", ) # The Google-side app identifier, back-filled BY HAND after IT # creates the app in Google admin. Purely for reconciliation — no # code path depends on it, and nothing populates it automatically # (there is no Google API to read it from; see the class docstring). external_id: Mapped[str | None] = mapped_column(String(255), nullable=True) display_name: Mapped[str] = mapped_column(String(255), nullable=False) platform: Mapped[str] = mapped_column( String(64), nullable=False, default="Web", ) auth_type: Mapped[str] = mapped_column( String(32), nullable=False, default="SAML", ) # ``ON for everyone`` / ``OFF for everyone`` / ``Mixed``. **Admin- # entered only** (app CRUD or CSV import) — there is no sync that # refreshes it, so a row created from a grant starts NULL and stays # NULL until someone records what they set in Google admin. Treat a # NULL as "unknown", never as "off". user_access_state: Mapped[str | None] = mapped_column( String(32), nullable=True, ) # Also admin-entered only — the cert-expiry governance surface can # only see apps whose expiry an admin actually recorded. certificate_expires_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # ── SP side: what the requester gave us, what IT types into Google ── acs_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) entity_id: Mapped[str | None] = mapped_column(String(1024), nullable=True) x509_cert: Mapped[str | None] = mapped_column(Text, nullable=True) # ── IdP side: what Google generated, owed back to the requester ── # Filled by IT after the manual create step in Google admin. Google # shows these as "SSO URL", "Entity ID" and the downloadable # certificate on the app's setup screen. idp_sso_url: Mapped[str | None] = mapped_column( String(1024), nullable=True, ) idp_entity_id: Mapped[str | None] = mapped_column( String(1024), nullable=True, ) idp_certificate: Mapped[str | None] = mapped_column(Text, nullable=True) # Set when the IdP config was actually sent to the requester. NULL on # a row with IdP values present = the config exists but nobody has # handed it over yet, which is the exact state that used to be # invisible. config_delivered_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) config_delivered_by_person_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) owner_person_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) source: Mapped[str] = mapped_column( String(32), nullable=False, default=SSO_APP_SOURCE_ADMIN_ADDED, ) raw: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) last_synced_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False, )