"""SSO Apps catalog + sso_grants SAML metadata extension. Two coordinated changes in one migration: 1. **New ``sso_apps`` table** — local catalog of every Google Workspace SAML app (the "Web and mobile apps" list in Google admin). Two sources populate this table: * Daily ``sso.google_saml_app_sync`` cron pulls Google's ``inboundSamlSsoProfiles`` via Cloud Identity API and upserts a row per profile (source = ``google_synced``). * The ``sso_grant_request`` Feishu approval flow, when approved for ``sso_provider='Google'``, creates a new SAML profile in Google AND inserts the matching catalog row (source = ``helix_created``). Owner-attribution lives on this table — IT picks one ``IdentityPerson`` per app via the SSO admin page so the leaver handover flow knows who to ping when an owner offboards. 2. **``sso_grants`` extension** — four new nullable columns to carry the SAML metadata an employee supplies when filing a "new SSO app" request: * ``acs_url`` — Assertion Consumer Service URL * ``entity_id`` — SP Entity ID * ``x509_cert`` — PEM-encoded signing certificate * ``saml_metadata_xml`` — full metadata XML (parsed at execute time to derive the above three when not supplied separately) * ``created_app_id`` — FK back to ``sso_apps.id`` populated after a successful Google API create (``ON DELETE SET NULL`` so manual catalog cleanup doesn't cascade-delete grant history) All four are nullable because (a) the existing Lark grants don't carry SAML data, (b) the historical pre-2026-05-07 Google grants pre-date the new flow, and (c) ``created_app_id`` only fills after the post-approval Google API call lands. ``sso_provider='Authing'`` is being deprecated as part of this change — historical rows are preserved (the `SsoGrant.sso_provider` column is free-text VARCHAR so existing rows continue to load), but the application-level ``SSO_PROVIDERS`` tuple drops it so the request form no longer surfaces it as a choice. """ from __future__ import annotations import sqlalchemy as sa from sqlalchemy import inspect from alembic import op from app.core.migration_helpers import ( id_type_for, normalize_fk_column_collation, table_kw_for, ) # revision identifiers, used by Alembic. revision = "126" down_revision = "125" branch_labels = None depends_on = None def upgrade() -> None: # Match identity_persons.id charset/collation/data-type so the # FK on owner_person_id doesn't trip TiDB's strict CHAR↔VARCHAR # incompatibility check (error 3780). Local MySQL is lenient, # prod TiDB rejects. Same defensive pattern as the rest of the # alembic chain — see app/core/migration_helpers.py. PERSON_ID = id_type_for("identity_persons") PERSON_KW = table_kw_for("identity_persons") # Idempotency rationale: the original migration (PR #193) and the # PR #194 fix both crashed on prod TiDB at FK-create time with # error 3780 because ``sso_grants.created_app_id`` inherited # ``sso_grants``'s default collation (``utf8mb4_bin``) instead of # the parent table's (``utf8mb4_unicode_ci`` on # ``identity_persons.id`` / ``sso_apps.id``). TiDB DDL auto-commits # CREATE TABLE + ALTER ADD COLUMN so the failed run left # ``sso_apps`` (with the right collation) AND # ``sso_grants.created_app_id`` (with the WRONG collation) on # prod, but ``alembic_version`` rolled back to 125 so the next # ``alembic upgrade`` retries this migration from the top. The # idempotent guards below + the new # ``normalize_fk_column_collation`` call make that retry succeed. bind = op.get_bind() inspector = inspect(bind) has_sso_apps = inspector.has_table("sso_apps") sso_grants_cols = ( {c["name"] for c in inspector.get_columns("sso_grants")} if inspector.has_table("sso_grants") else set() ) sso_grants_fks = ( {fk.get("name") for fk in inspector.get_foreign_keys("sso_grants")} if inspector.has_table("sso_grants") else set() ) if has_sso_apps: # The earlier failed run already committed CREATE TABLE; skip # the create + just stamp the column metadata for the FK step. # (No idempotent IF NOT EXISTS shortcut in alembic — gate # explicitly.) return _resume_from_table_exists( sso_grants_cols=sso_grants_cols, sso_grants_fks=sso_grants_fks, person_id=PERSON_ID, ) op.create_table( "sso_apps", sa.Column("id", PERSON_ID, primary_key=True), # External identifier for the SAML SSO profile in Google # (the ``inboundSamlSsoProfiles/`` resource name's last # segment). NULL for catalog rows that were promised by # ``helix_created`` but the Google API call failed — the row # stays so the grant can retry without a duplicate insert. sa.Column("provider", sa.String(64), nullable=False, server_default="google_workspace"), sa.Column("external_id", sa.String(255), nullable=True), # Display name (shown in Google admin "Web and mobile apps" # column). Required — the SSO grant flow uses this as the # join key when an admin creates a row in Google admin # without going through Helix. sa.Column("display_name", sa.String(255), nullable=False), sa.Column("platform", sa.String(64), nullable=False, server_default="Web"), sa.Column("auth_type", sa.String(32), nullable=False, server_default="SAML"), # ``ON for everyone`` / ``OFF for everyone`` / ``Mixed`` — # mirror of Google admin's ``User access`` column. Cloud # Identity returns this as part of the ``inboundSsoAssignments`` # graph; we summarise to a string for the catalog list UI. sa.Column("user_access_state", sa.String(32), nullable=True), # PEM cert + expiry derived from the SAML signing cert so the # SSO admin page can highlight rows expiring < 60d. sa.Column("certificate_expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("acs_url", sa.String(1024), nullable=True), sa.Column("entity_id", sa.String(1024), nullable=True), sa.Column("x509_cert", sa.Text, nullable=True), # Helix-side ownership. SET NULL on delete so removing a # person doesn't orphan the catalog row — IT just gets a # "owner unassigned" badge in the UI. sa.Column("owner_person_id", PERSON_ID, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True), # Provenance — `google_synced` (catalog row pulled by the # daily cron), `helix_created` (catalog row inserted by the # SSO grant approval flow). Used in the catalog table column # so admins can tell at a glance which apps Helix manages # the lifecycle of vs. which were pre-existing in Google. sa.Column("source", sa.String(32), nullable=False, server_default="google_synced"), # Free-form blob: per-source raw payload (Cloud Identity # response for `google_synced`, or the request form details # for `helix_created`). Useful for debugging; not part of # the API surface. sa.Column("raw", sa.JSON, nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), # UNIQUE on (provider, external_id) — Google returns stable # resource ids, so two sync runs on the same profile dedup # cleanly. NULL external_id is allowed (helix_created rows # whose API call hasn't landed yet) and MySQL's UNIQUE allows # multiple NULLs, which is exactly what we want. sa.UniqueConstraint("provider", "external_id", name="uq_sso_apps_provider_external_id"), # Secondary lookup: case-insensitive display_name within a # provider so the grant approval flow can join on # `display_name == sso_grant.system_name` to detect # already-existing apps. sa.Index("ix_sso_apps_provider_display_name", "provider", "display_name"), sa.Index("ix_sso_apps_owner_person_id", "owner_person_id"), **PERSON_KW, ) # sso_grants extension — see module docstring. ``created_app_id`` # FK type tracks ``sso_apps.id`` which we just created using # ``id_type_for("identity_persons")`` (CHAR(36) on prod TiDB). SSO_APP_ID = id_type_for("identity_persons") with op.batch_alter_table("sso_grants") as batch: batch.add_column(sa.Column( "acs_url", sa.String(1024), nullable=True, )) batch.add_column(sa.Column( "entity_id", sa.String(1024), nullable=True, )) batch.add_column(sa.Column( "x509_cert", sa.Text, nullable=True, )) batch.add_column(sa.Column( "saml_metadata_xml", sa.Text, nullable=True, )) batch.add_column(sa.Column( "created_app_id", SSO_APP_ID, nullable=True, )) # CRITICAL on TiDB: even though both columns are CHAR(36), the # newly-added ``sso_grants.created_app_id`` inherits the # ``sso_grants`` table default collation (``utf8mb4_bin`` on # PingCAP prod), not the parent ``sso_apps.id``'s # ``utf8mb4_unicode_ci``. TiDB error 3780 fires on the FK create # below unless we MODIFY COLUMN to align collations first. Local # MySQL is lenient and accepts the FK without this — the helper # is a no-op in that case (matching parent collation = early # return). PR #194 attempted to fix this but only addressed # ``sso_apps.id``; it missed the same issue on # ``sso_grants.created_app_id`` so prod kept blocking on the FK. normalize_fk_column_collation( table="sso_grants", column="created_app_id", parent_table="sso_apps", ) # FK added separately so MySQL accepts the cross-table reference # to a table that was just created in this same upgrade. op.create_foreign_key( "fk_sso_grants_created_app", "sso_grants", "sso_apps", ["created_app_id"], ["id"], ondelete="SET NULL", ) def _resume_from_table_exists( *, sso_grants_cols: set[str], sso_grants_fks: set[str | None], person_id, ) -> None: """Resume the migration when ``sso_apps`` already exists. Path triggered when a prior attempt (PR #193's bad migration or PR #194's incomplete fix) committed the CREATE TABLE + ALTER COLUMN steps but rolled back at the FK-create boundary. We finish the work the prior run skipped: add any missing ``sso_grants`` columns, normalize the ``created_app_id`` collation to match ``sso_apps.id``, then create the FK if it isn't already present. Idempotent under repeat: a third run finds all columns + the FK in place and exits without writing anything. """ add_specs = [ ("acs_url", sa.String(1024)), ("entity_id", sa.String(1024)), ("x509_cert", sa.Text), ("saml_metadata_xml", sa.Text), ("created_app_id", person_id), ] missing = [(name, col_type) for name, col_type in add_specs if name not in sso_grants_cols] if missing: with op.batch_alter_table("sso_grants") as batch: for name, col_type in missing: batch.add_column(sa.Column(name, col_type, nullable=True)) normalize_fk_column_collation( table="sso_grants", column="created_app_id", parent_table="sso_apps", ) if "fk_sso_grants_created_app" not in sso_grants_fks: op.create_foreign_key( "fk_sso_grants_created_app", "sso_grants", "sso_apps", ["created_app_id"], ["id"], ondelete="SET NULL", ) def downgrade() -> None: op.drop_constraint( "fk_sso_grants_created_app", "sso_grants", type_="foreignkey", ) with op.batch_alter_table("sso_grants") as batch: batch.drop_column("created_app_id") batch.drop_column("saml_metadata_xml") batch.drop_column("x509_cert") batch.drop_column("entity_id") batch.drop_column("acs_url") op.drop_index("ix_sso_apps_owner_person_id", table_name="sso_apps") op.drop_index("ix_sso_apps_provider_display_name", table_name="sso_apps") op.drop_constraint( "uq_sso_apps_provider_external_id", "sso_apps", type_="unique", ) op.drop_table("sso_apps")