"""IdP-side SAML config + delivery tracking on ``sso_apps``. Closes the return leg of the Google SSO grant flow. Until now the flow carried metadata in ONE direction: the requester handed us the SP-side values (ACS URL / Entity ID / X.509), Helix parsed them, notified IT, and IT created the custom SAML app in Google admin by hand. But creating that app makes Google the **IdP**, and Google mints its own SSO URL, Entity ID and signing certificate — values the requester (or their vendor) MUST configure on the SP side or the integration never works. Helix had nowhere to store those three values and no flow to hand them back, so the final hop lived entirely in IT's memory: send an ad-hoc message, or the request silently stalls while Helix shows the grant as ``approved`` with a complete catalog row (i.e. reading as DONE). Five additive nullable columns: * ``idp_sso_url`` / ``idp_entity_id`` / ``idp_certificate`` — what Google generated. Filled by IT after the manual creation step. * ``config_delivered_at`` / ``config_delivered_by_person_id`` — proof the values actually reached the requester, so "waiting on IT to send the config" becomes a queryable state instead of a blind spot. All NULL for every existing row, which is honest: no historical app has a recorded delivery, and pretending otherwise would fabricate evidence. FK on ``config_delivered_by_person_id`` uses ``ON DELETE SET NULL``, matching ``owner_person_id`` on the same table — a deleted admin must not cascade away the app row. Revision ID: 515 Revises: 514 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import ( id_type_for, normalize_fk_column_collation, ) revision = "515" down_revision = "514" branch_labels = None depends_on = None _TABLE = "sso_apps" # ``config_delivered_by_person_id`` MUST take its type from # ``id_type_for("identity_persons")``, not a hand-written ``sa.String(36)``. # A hand-written type (or a bare CHAR(36), which inherits the TABLE # default collation) lands on TiDB error 3780 when the parent collates # differently — local MySQL 8 has one collation everywhere and cannot # reproduce it, so this only shows up as an aborted prod deploy with the # chain stuck mid-migration. Built lazily inside ``upgrade()`` because the # helper reads live parent metadata. _SIMPLE_COLUMNS: tuple[tuple[str, sa.types.TypeEngine], ...] = ( ("idp_sso_url", sa.String(1024)), ("idp_entity_id", sa.String(1024)), ("idp_certificate", sa.Text()), ("config_delivered_at", sa.DateTime(timezone=True)), ) _FK_COLUMN = "config_delivered_by_person_id" _FK_NAME = "fk_sso_apps_config_delivered_by" def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1" ), {"t": table}, ).scalar() is not None def _column_exists(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() AND table_name = :t " "AND column_name = :c LIMIT 1" ), {"t": table, "c": column}, ).scalar() is not None def _fk_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.table_constraints " "WHERE table_schema = DATABASE() AND table_name = :t " "AND constraint_name = :n AND constraint_type = 'FOREIGN KEY' " "LIMIT 1" ), {"t": table, "n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, _TABLE): return for name, coltype in _SIMPLE_COLUMNS: if not _column_exists(bind, _TABLE, name): op.add_column(_TABLE, sa.Column(name, coltype, nullable=True)) if not _column_exists(bind, _TABLE, _FK_COLUMN): op.add_column( _TABLE, sa.Column( _FK_COLUMN, id_type_for("identity_persons"), nullable=True, ), ) # FK added separately (and guarded) so a partially-applied run is # resumable — TiDB rejects a duplicate constraint name outright. if ( _table_exists(bind, "identity_persons") and _column_exists(bind, _TABLE, _FK_COLUMN) and not _fk_exists(bind, _TABLE, _FK_NAME) ): # Re-assert the collation on an existing column too: a half-applied # earlier run could have added it with the table default. normalize_fk_column_collation(_TABLE, _FK_COLUMN, "identity_persons") op.create_foreign_key( _FK_NAME, _TABLE, "identity_persons", [_FK_COLUMN], ["id"], ondelete="SET NULL", ) def downgrade() -> None: bind = op.get_bind() if not _table_exists(bind, _TABLE): return if _fk_exists(bind, _TABLE, _FK_NAME): op.drop_constraint(_FK_NAME, _TABLE, type_="foreignkey") for name in (_FK_COLUMN, *[c for c, _ in reversed(_SIMPLE_COLUMNS)]): if _column_exists(bind, _TABLE, name): op.drop_column(_TABLE, name)