"""Track which IAM/RAM user an API key assignment is bound to. Revision ID: 073 Revises: 072 Create Date: 2026-04-27 Bedrock auto-grant currently mints a fresh IAM user per request and stores its name in ``api_key_assignments.detail.bedrock.iam_user_name``. Admins now need to be able to: 1. View the IAM user a key is bound to (without parsing JSON detail), 2. Re-bind an assignment to a different IAM user (e.g. consolidate two access keys onto one IAM user, or move a request from the wrong AWS account to the right one), 3. Re-assign the Helix-side ``owner_person_id`` of any ``cmdb_iam_users`` row independently of API-key flow. The columns added here: * ``iam_user_id`` — FK into ``cmdb_iam_users`` so the assignment row is resolvable via JOIN. Nullable because legacy assignments and non-cloud providers (OpenAI, Azure OpenAI, …) have no IAM user. * ``iam_user_arn`` — denormalised display label, populated alongside ``iam_user_id`` so the admin UI can render the binding without an extra query. Updated by the auto-grant + rebind paths. * ``iam_integration_id`` — points at the cloud integration the IAM user lives in. The same RAM/IAM user name can recur across different cloud accounts; the integration is the disambiguator. The owner_person_id column on cmdb_iam_users already exists (added in the original cmdb migration) — admins just couldn't reach it from the UI. The new ``PATCH /cmdb/iam-users/{id}/owner`` endpoint surfaces it. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "073" down_revision = "072" branch_labels = None depends_on = 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", ), {"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'", ), {"t": table, "n": name}, ).scalar() is not None def _index_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND INDEX_NAME = :n", ), {"t": table, "n": name}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "varchar", fallback_collation: str | None = None, length: int = 36, ): """Return an SA type instance that mirrors an FK target's type+collation. TiDB rejects FKs whose source column's data type differs from the target's — even CHAR(36) vs VARCHAR(36) is rejected, and a collation mismatch yields ER_FK_INCOMPATIBLE_COLUMNS (3780). MySQL 8.0 is more permissive on the CHAR/VARCHAR axis but still strict on collation. We look up the target's actual ``DATA_TYPE`` and ``COLLATION_NAME`` at runtime so the source column we are about to add will match the cluster regardless of which historical migration created the target. Falls back to the explicit defaults the caller passes so test bootstraps where the target table does not yet exist still pick something plausible. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND COLUMN_NAME = :c", ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def upgrade() -> None: bind = op.get_bind() # ── Recovery: previous runs of this migration added the columns # without explicit type+COLLATE pinning. Two FK target mismatches # leaked through: # # 1. ``cmdb_iam_users.id`` is ``utf8mb4_bin`` while the host # table is ``utf8mb4_unicode_ci`` (collation mismatch — first # failure mode, fixed by the prior attempt for iam_user_id). # 2. ``integrations.id`` is ``CHAR(36)`` while the source column # defaulted to ``VARCHAR(36)`` — TiDB rejects CHAR↔VARCHAR # FKs even when collation matches. # # Both surfaced as ER_FK_INCOMPATIBLE_COLUMNS (3780). Since MySQL # DDL is non-transactional, prod was left with the columns added # but FKs/index unbuilt. Drop the partial state so we can rebuild # both columns with the right type and collation. if _index_exists( bind, "api_key_assignments", "ix_api_key_assignments_iam_user_id", ): op.drop_index( "ix_api_key_assignments_iam_user_id", table_name="api_key_assignments", ) if _fk_exists( bind, "api_key_assignments", "fk_api_key_assignments_iam_user_id", ): op.drop_constraint( "fk_api_key_assignments_iam_user_id", "api_key_assignments", type_="foreignkey", ) if _fk_exists( bind, "api_key_assignments", "fk_api_key_assignments_iam_integration_id", ): op.drop_constraint( "fk_api_key_assignments_iam_integration_id", "api_key_assignments", type_="foreignkey", ) if _column_exists(bind, "api_key_assignments", "iam_integration_id"): op.drop_column("api_key_assignments", "iam_integration_id") if _column_exists(bind, "api_key_assignments", "iam_user_arn"): op.drop_column("api_key_assignments", "iam_user_arn") if _column_exists(bind, "api_key_assignments", "iam_user_id"): op.drop_column("api_key_assignments", "iam_user_id") # Build the FK source columns to mirror the target's exact # type+collation. ``cmdb_iam_users.id`` is ``VARCHAR(36) # COLLATE utf8mb4_bin`` on prod while ``integrations.id`` is # ``CHAR(36) COLLATE utf8mb4_unicode_ci`` — the helper resolves # both at runtime so we don't hard-code one over the other. op.add_column( "api_key_assignments", sa.Column( "iam_user_id", _fk_source_col_type( bind, "cmdb_iam_users", "id", fallback_type="varchar", fallback_collation="utf8mb4_bin", ), nullable=True, ), ) op.add_column( "api_key_assignments", sa.Column("iam_user_arn", sa.String(512), nullable=True), ) op.add_column( "api_key_assignments", sa.Column( "iam_integration_id", _fk_source_col_type( bind, "integrations", "id", fallback_type="char", fallback_collation="utf8mb4_unicode_ci", ), nullable=True, ), ) op.create_foreign_key( "fk_api_key_assignments_iam_user_id", "api_key_assignments", "cmdb_iam_users", ["iam_user_id"], ["id"], ondelete="SET NULL", ) op.create_foreign_key( "fk_api_key_assignments_iam_integration_id", "api_key_assignments", "integrations", ["iam_integration_id"], ["id"], ondelete="SET NULL", ) op.create_index( "ix_api_key_assignments_iam_user_id", "api_key_assignments", ["iam_user_id"], ) # Best-effort backfill for existing Bedrock auto-grant rows: parse # detail.bedrock.iam_user_name and try to match a cmdb_iam_users row # by user_name. Failures here are non-fatal — a future iam-sync run # will populate the link, and the admin can rebind manually. rows = bind.execute(sa.text( "SELECT id, detail FROM api_key_assignments " "WHERE detail IS NOT NULL" )).fetchall() for row in rows: ass_id = row[0] det = row[1] if det is None: continue if isinstance(det, (bytes, bytearray)): try: det = det.decode("utf-8") except UnicodeDecodeError: continue if isinstance(det, str): import json as _json try: det = _json.loads(det) except (TypeError, ValueError): continue if not isinstance(det, dict): continue bedrock = det.get("bedrock") or {} user_name = bedrock.get("iam_user_name") integration_slug = bedrock.get("integration_slug") if not user_name: continue # Find the integration id by slug, then the iam_users row. intg = bind.execute(sa.text( "SELECT id FROM integrations WHERE slug = :s LIMIT 1" ), {"s": integration_slug or ""}).fetchone() if not intg: continue iam = bind.execute(sa.text( "SELECT id, arn FROM cmdb_iam_users " "WHERE integration_id = :i AND user_name = :u " "AND deleted_at IS NULL LIMIT 1" ), {"i": str(intg[0]), "u": user_name}).fetchone() if not iam: continue bind.execute(sa.text( "UPDATE api_key_assignments " "SET iam_user_id = :u, iam_user_arn = :a, " " iam_integration_id = :i " "WHERE id = :id" ), { "u": str(iam[0]), "a": iam[1] or bedrock.get("user_arn"), "i": str(intg[0]), "id": ass_id, }) def downgrade() -> None: op.drop_index( "ix_api_key_assignments_iam_user_id", table_name="api_key_assignments", ) op.drop_constraint( "fk_api_key_assignments_iam_integration_id", "api_key_assignments", type_="foreignkey", ) op.drop_constraint( "fk_api_key_assignments_iam_user_id", "api_key_assignments", type_="foreignkey", ) op.drop_column("api_key_assignments", "iam_integration_id") op.drop_column("api_key_assignments", "iam_user_arn") op.drop_column("api_key_assignments", "iam_user_id")