"""Unified auditor findings — cross-control actionable worklist. Revision ID: 255 Revises: 254 Create Date: 2026-06-05 One table, ``auditor_findings``: a uniform lifecycle (open → acknowledged → accepted_risk / resolved) for every detective control's violating items so a single inbox (and a single posture rollup) spans all controls. The two controls that already kept a per-item decision (public exposure, sign-in anomaly) are backfilled here, carrying their prior lifecycle state so the worklist is populated on deploy without waiting for a cron; the native producers (MFA, secret rotation, cert expiry, PAT) start emitting findings in the follow-up PR. TiDB-FK strictness: id/FK columns CHAR(36); person FKs use ``id_type_for`` and the table inherits identity_persons' charset/collation via ``table_kw_for``. DROP-if-VARCHAR pre-flight for re-run convergence. """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "255" down_revision = "254" branch_labels = None depends_on = None _TABLE = "auditor_findings" def _table_exists(name: str) -> bool: return op.get_bind().execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1" ), {"n": name}, ).scalar() is not None def _id_is_varchar(name: str) -> bool: row = op.get_bind().execute( sa.text( "SELECT DATA_TYPE FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n " "AND COLUMN_NAME = 'id' LIMIT 1" ), {"n": name}, ).scalar() return bool(row) and str(row).lower() == "varchar" def upgrade() -> None: persons_id = id_type_for("identity_persons") table_kw = table_kw_for("identity_persons") if _id_is_varchar(_TABLE): op.execute(f"DROP TABLE IF EXISTS `{_TABLE}`") if _table_exists(_TABLE): return op.create_table( _TABLE, sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("control_key", sa.String(64), nullable=False), sa.Column("finding_type", sa.String(64), nullable=False), sa.Column("dedup_key", sa.String(255), nullable=False), sa.Column("subject_type", sa.String(32), nullable=False), sa.Column("subject_ref", sa.String(512), nullable=True), sa.Column("subject_display", sa.String(255), nullable=True), sa.Column("person_id", persons_id, nullable=True), sa.Column("severity", sa.String(16), nullable=False, server_default=sa.text("'medium'")), sa.Column("title", sa.String(512), nullable=False), sa.Column("detail", sa.JSON(), nullable=True), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'open'")), sa.Column("source_table", sa.String(64), nullable=True), sa.Column("source_id", sa.CHAR(36), nullable=True), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("decided_by_person_id", persons_id, nullable=True), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), sa.Column("decision_note", sa.Text(), nullable=True), sa.Column("accepted_until", sa.DateTime(timezone=True), nullable=True), sa.Column("auto_resolved", sa.Boolean(), nullable=False, server_default=sa.text("0")), sa.Column("resolved_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), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_auditor_finding_person", ), sa.ForeignKeyConstraint( ["decided_by_person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_auditor_finding_decider", ), sa.UniqueConstraint("control_key", "dedup_key", name="uq_auditor_finding_dedup"), **table_kw, ) op.create_index("idx_auditor_finding_status_sev", _TABLE, ["status", "severity"]) op.create_index("idx_auditor_finding_control_status", _TABLE, ["control_key", "status"]) op.create_index("idx_auditor_finding_person", _TABLE, ["person_id"]) op.create_index("idx_auditor_finding_last_seen", _TABLE, ["last_seen_at"]) # ── Backfill: public exposure (open + accepted_risk) ──────────────── op.execute( """ INSERT INTO auditor_findings ( id, control_key, finding_type, dedup_key, subject_type, subject_ref, subject_display, person_id, severity, title, detail, status, source_table, source_id, first_seen_at, last_seen_at, auto_resolved, created_at, updated_at ) SELECT UUID(), 'public_exposure_scan', 'public_exposure', e.id, 'resource', COALESCE(e.resource_arn, e.resource_id), COALESCE(e.resource_name, e.resource_id), NULL, e.severity, CONCAT(e.exposure_type, ' — ', COALESCE(e.resource_name, e.resource_id)), JSON_OBJECT('source', e.source, 'region', e.region, 'exposure_type', e.exposure_type), CASE e.status WHEN 'accepted_risk' THEN 'accepted_risk' ELSE 'open' END, 'auditor_public_exposure_findings', e.id, e.first_seen_at, e.last_seen_at, 0, NOW(), NOW() FROM auditor_public_exposure_findings e WHERE e.status IN ('open', 'accepted_risk') """ ) # ── Backfill: high/critical unacknowledged sign-in anomalies (30d) ── op.execute( """ INSERT INTO auditor_findings ( id, control_key, finding_type, dedup_key, subject_type, subject_ref, subject_display, person_id, severity, title, detail, status, source_table, source_id, first_seen_at, last_seen_at, auto_resolved, created_at, updated_at ) SELECT UUID(), 'signin_anomaly_monitoring', 'signin_anomaly', s.id, 'person', s.provider_email, s.provider_email, s.person_id, CASE s.anomaly_level WHEN 'critical' THEN 'critical' ELSE 'high' END, CONCAT(s.provider, ': ', COALESCE(s.provider_email, 'unknown principal')), JSON_OBJECT('provider', s.provider, 'source_ip', s.source_ip, 'anomaly_score', s.anomaly_score), 'open', 'auditor_signin_events', s.id, s.event_at, s.event_at, 0, NOW(), NOW() FROM auditor_signin_events s WHERE s.anomaly_level IN ('high', 'critical') AND s.acknowledged_at IS NULL AND s.event_at >= (NOW() - INTERVAL 30 DAY) """ ) def downgrade() -> None: op.execute(f"DROP TABLE IF EXISTS `{_TABLE}`")