"""Security Auditor — 10 tables for the eight auditor sub-features. Revision ID: 093 Revises: 092 Create Date: 2026-04-30 Eight auditor sub-features (described in the user-facing requirement) and the data model that backs them: 1. MFA compliance audit → auditor_mfa_status 2. Privileged role quarterly review → auditor_privileged_review_cycles + auditor_privileged_role_holders 3. Public exposure scan → auditor_public_exposure_findings 4. Secret rotation ledger → auditor_secret_rotation_records 5. Signin anomaly monitoring → auditor_signin_events 6. PAT / MCP / OAuth audit stream → auditor_pat_audits 7. Cert + ICP expiry → auditor_cert_expiries 8. Risk score + evidence package → auditor_risk_scores + auditor_evidence_packs Six scheduled jobs are seeded: * auditor.mfa_sync_daily (08:00) * auditor.public_exposure_scan_daily (02:00) * auditor.secret_rotation_check_daily (03:00) * auditor.signin_anomaly_score_hourly (every hour) * auditor.pat_audit_daily (04:00) * auditor.cert_expiry_check_daily (06:00) Risk-score recompute and evidence pack generation are on-demand only — not on a fixed schedule (risk score recomputes after each upstream write; evidence pack runs from a worker queue when an auditor clicks "build pack"). ═══════════════════════════════════════════════════════════════════════ TiDB FK strictness ═══════════════════════════════════════════════════════════════════════ Every FK column matches the parent's actual DATA_TYPE + COLLATION_NAME via ``_fk_source_col_type`` (same shape as 088 / 091) so TiDB accepts the FKs (errno 3780). ═══════════════════════════════════════════════════════════════════════ Idempotency ═══════════════════════════════════════════════════════════════════════ Every CREATE TABLE / ADD COLUMN / INSERT INTO scheduled_jobs is guarded with an existence check. """ from __future__ import annotations import json import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "093" down_revision = "092" branch_labels = None depends_on = None # ── Idempotency helpers (lifted from 091) ───────────────────────────── def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).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" ), {"t": table, "c": column}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", length: int = 36, ): """Match parent column's DATA_TYPE + COLLATION_NAME so TiDB accepts FK.""" 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] else: data_type = fallback_type.lower() collation = None type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) _IDEMPOTENT_RETRY_HINTS = ( "duplicate key name", "duplicate entry", "duplicate foreign key", "errno: 121", "check that column/key exists", ) def _try_swallow(label: str, fn) -> None: try: fn() except Exception as exc: # noqa: BLE001 msg = str(exc).lower() if any(hint in msg for hint in _IDEMPOTENT_RETRY_HINTS): logger.warning( "migration 093: %s skipped (idempotent retry): %s", label, exc, ) return logger.error( "migration 093: %s failed with non-idempotent error: %s", label, exc, ) raise # ── upgrade ───────────────────────────────────────────────────────── def upgrade() -> None: bind = op.get_bind() persons_id = _fk_source_col_type(bind, "identity_persons", "id") # ════════════════════════════════════════════════════════════ # 1. MFA compliance audit # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_mfa_status"): op.create_table( "auditor_mfa_status", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("person_id", persons_id, nullable=False), sa.Column("provider", sa.String(32), nullable=False), sa.Column("provider_user_id", sa.String(255), nullable=True), sa.Column("provider_email", sa.String(255), nullable=True), sa.Column("is_enrolled", sa.Boolean, nullable=False, server_default=sa.text("0")), sa.Column("enrollment_methods", sa.JSON, nullable=True), sa.Column("is_enforced", sa.Boolean, nullable=False, server_default=sa.text("0")), sa.Column("last_verified_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=False), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_auditor_mfa_status_person", ), sa.UniqueConstraint( "person_id", "provider", name="uq_auditor_mfa_status_person_provider", ), sa.Index("idx_auditor_mfa_provider_enrolled", "provider", "is_enrolled"), sa.Index("idx_auditor_mfa_synced", "last_synced_at"), ) # ════════════════════════════════════════════════════════════ # 2. Privileged role quarterly review # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_privileged_review_cycles"): op.create_table( "auditor_privileged_review_cycles", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("cycle_label", sa.String(32), nullable=False), sa.Column( "status", sa.String(32), nullable=False, server_default=sa.text("'in_progress'"), ), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("due_at", sa.DateTime(timezone=True), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("created_by_person_id", persons_id, nullable=True), sa.Column("snapshot_holder_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("decided_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("revoked_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("notes", sa.Text, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["created_by_person_id"], ["identity_persons.id"], name="fk_auditor_priv_cycle_creator", ), sa.UniqueConstraint("cycle_label", name="uq_auditor_priv_cycle_label"), sa.Index("idx_auditor_priv_cycle_status", "status"), sa.Index("idx_auditor_priv_cycle_due", "due_at"), ) if not _table_exists(bind, "auditor_privileged_role_holders"): op.create_table( "auditor_privileged_role_holders", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("cycle_id", sa.CHAR(36), nullable=False), sa.Column("person_id", persons_id, nullable=True), sa.Column("scope", sa.String(32), nullable=False), sa.Column("role_name", sa.String(128), nullable=False), sa.Column("holder_external_id", sa.String(255), nullable=True), sa.Column("holder_email", sa.String(255), nullable=True), sa.Column("holder_display_name", sa.String(255), nullable=True), sa.Column("snapshot_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "decision", sa.String(32), nullable=False, server_default=sa.text("'pending'"), ), 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("revocation_completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("revocation_trace_id", sa.CHAR(36), nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["cycle_id"], ["auditor_privileged_review_cycles.id"], ondelete="CASCADE", name="fk_auditor_priv_holder_cycle", ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_auditor_priv_holder_person", ), sa.ForeignKeyConstraint( ["decided_by_person_id"], ["identity_persons.id"], name="fk_auditor_priv_holder_decider", ), sa.Index("idx_auditor_priv_holder_cycle_decision", "cycle_id", "decision"), sa.Index("idx_auditor_priv_holder_scope_role", "scope", "role_name"), sa.Index("idx_auditor_priv_holder_person", "person_id"), ) # ════════════════════════════════════════════════════════════ # 3. Public exposure scan # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_public_exposure_findings"): op.create_table( "auditor_public_exposure_findings", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("scan_run_id", sa.CHAR(36), nullable=False), sa.Column("source", sa.String(64), nullable=False), sa.Column("provider_account_id", sa.CHAR(36), nullable=True), sa.Column("region", sa.String(64), nullable=True), sa.Column("resource_id", sa.String(255), nullable=False), sa.Column("resource_arn", sa.String(512), nullable=True), sa.Column("resource_name", sa.String(255), nullable=True), sa.Column("exposure_type", sa.String(64), nullable=False), sa.Column("exposure_detail", sa.JSON, nullable=True), sa.Column( "severity", sa.String(16), nullable=False, server_default=sa.text("'medium'"), ), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'open'"), ), sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True), sa.Column("resolved_by_person_id", persons_id, nullable=True), sa.Column("resolution_note", sa.Text, nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["resolved_by_person_id"], ["identity_persons.id"], name="fk_auditor_pubex_resolver", ), sa.UniqueConstraint( "source", "resource_id", "exposure_type", name="uq_auditor_pubex_resource", ), sa.Index("idx_auditor_pubex_status_severity", "status", "severity"), sa.Index("idx_auditor_pubex_source", "source"), sa.Index("idx_auditor_pubex_last_seen", "last_seen_at"), ) # ════════════════════════════════════════════════════════════ # 4. Secret rotation ledger # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_secret_rotation_records"): op.create_table( "auditor_secret_rotation_records", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("secret_kind", sa.String(64), nullable=False), sa.Column("source", sa.String(64), nullable=False), sa.Column("resource_id", sa.String(255), nullable=False), sa.Column("resource_name", sa.String(255), nullable=True), sa.Column("owner_person_id", persons_id, nullable=True), sa.Column("owner_email", sa.String(255), nullable=True), sa.Column("created_at_source", sa.DateTime(timezone=True), nullable=True), sa.Column("last_rotated_at", sa.DateTime(timezone=True), nullable=True), sa.Column("rotation_period_days", sa.Integer, nullable=False, server_default=sa.text("90")), sa.Column("next_rotation_due_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'current'"), ), sa.Column("days_until_due", sa.Integer, nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], name="fk_auditor_secret_owner", ), sa.UniqueConstraint( "source", "secret_kind", "resource_id", name="uq_auditor_secret_resource", ), sa.Index("idx_auditor_secret_status_due", "status", "next_rotation_due_at"), sa.Index("idx_auditor_secret_kind", "secret_kind"), sa.Index("idx_auditor_secret_owner", "owner_person_id"), ) # ════════════════════════════════════════════════════════════ # 5. Signin anomaly monitoring # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_signin_events"): op.create_table( "auditor_signin_events", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("person_id", persons_id, nullable=True), sa.Column("provider", sa.String(32), nullable=False), sa.Column("provider_event_id", sa.String(255), nullable=True), sa.Column("provider_user_id", sa.String(255), nullable=True), sa.Column("provider_email", sa.String(255), nullable=True), sa.Column("event_at", sa.DateTime(timezone=True), nullable=False), sa.Column("event_kind", sa.String(64), nullable=False), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'success'")), sa.Column("source_ip", sa.String(64), nullable=True), sa.Column("source_country", sa.String(8), nullable=True), sa.Column("source_city", sa.String(128), nullable=True), sa.Column("user_agent", sa.String(512), nullable=True), sa.Column("device_id", sa.String(255), nullable=True), sa.Column("anomaly_score", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("anomaly_level", sa.String(16), nullable=False, server_default=sa.text("'normal'")), sa.Column("anomaly_reasons", sa.JSON, nullable=True), sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True), sa.Column("acknowledged_by_person_id", persons_id, nullable=True), sa.Column("acknowledged_note", sa.Text, nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_auditor_signin_person", ), sa.ForeignKeyConstraint( ["acknowledged_by_person_id"], ["identity_persons.id"], name="fk_auditor_signin_acker", ), sa.UniqueConstraint( "provider", "provider_event_id", name="uq_auditor_signin_provider_event", ), sa.Index("idx_auditor_signin_person_at", "person_id", "event_at"), sa.Index("idx_auditor_signin_anomaly", "anomaly_level", "event_at"), sa.Index("idx_auditor_signin_provider_at", "provider", "event_at"), sa.Index("idx_auditor_signin_event_at", "event_at"), ) # ════════════════════════════════════════════════════════════ # 6. PAT / MCP / OAuth audit # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_pat_audits"): op.create_table( "auditor_pat_audits", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("kind", sa.String(32), nullable=False), sa.Column("source", sa.String(32), nullable=False), sa.Column("token_external_id", sa.String(255), nullable=False), sa.Column("person_id", persons_id, nullable=True), sa.Column("owner_email", sa.String(255), nullable=True), sa.Column("token_name", sa.String(255), nullable=True), sa.Column("scopes", sa.JSON, nullable=True), sa.Column("created_at_source", sa.DateTime(timezone=True), nullable=True), sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'active'"), ), sa.Column("risk_flags", sa.JSON, nullable=True), sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), sa.Column("revoked_by_person_id", persons_id, nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_auditor_pat_person", ), sa.ForeignKeyConstraint( ["revoked_by_person_id"], ["identity_persons.id"], name="fk_auditor_pat_revoker", ), sa.UniqueConstraint( "source", "kind", "token_external_id", name="uq_auditor_pat_token", ), sa.Index("idx_auditor_pat_status_kind", "status", "kind"), sa.Index("idx_auditor_pat_person", "person_id"), sa.Index("idx_auditor_pat_expires", "expires_at"), ) # ════════════════════════════════════════════════════════════ # 7. Cert + ICP expiry # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_cert_expiries"): op.create_table( "auditor_cert_expiries", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("asset_kind", sa.String(32), nullable=False), sa.Column("source", sa.String(32), nullable=False), sa.Column("external_id", sa.String(512), nullable=False), sa.Column("name", sa.String(512), nullable=False), sa.Column("subject", sa.String(512), nullable=True), sa.Column("issuer", sa.String(255), nullable=True), sa.Column("issued_at", sa.DateTime(timezone=True), nullable=True), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), sa.Column("days_until_expiry", sa.Integer, nullable=True), sa.Column("alert_level", sa.String(16), nullable=False, server_default=sa.text("'none'")), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'active'")), sa.Column("owner_person_id", persons_id, nullable=True), sa.Column("owner_email", sa.String(255), nullable=True), sa.Column("environment", sa.String(32), nullable=True), sa.Column("notes", sa.Text, nullable=True), sa.Column("raw_payload", sa.JSON, nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], name="fk_auditor_cert_owner", ), sa.UniqueConstraint( "source", "asset_kind", "external_id", name="uq_auditor_cert_external", ), sa.Index("idx_auditor_cert_alert_level", "alert_level"), sa.Index("idx_auditor_cert_expires", "expires_at"), sa.Index("idx_auditor_cert_kind", "asset_kind"), ) # ════════════════════════════════════════════════════════════ # 8a. Risk scores # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_risk_scores"): op.create_table( "auditor_risk_scores", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("person_id", persons_id, nullable=False), sa.Column("score", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("level", sa.String(16), nullable=False, server_default=sa.text("'low'")), sa.Column("factors", sa.JSON, nullable=False), sa.Column("breakdown", sa.JSON, nullable=False), sa.Column("computed_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_auditor_risk_person", ), sa.UniqueConstraint("person_id", name="uq_auditor_risk_person"), sa.Index("idx_auditor_risk_level", "level", "score"), sa.Index("idx_auditor_risk_computed", "computed_at"), ) # ════════════════════════════════════════════════════════════ # 8b. Evidence packs # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "auditor_evidence_packs"): op.create_table( "auditor_evidence_packs", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("requested_by_person_id", persons_id, nullable=False), sa.Column("title", sa.String(255), nullable=False), sa.Column("period_start", sa.DateTime(timezone=True), nullable=False), sa.Column("period_end", sa.DateTime(timezone=True), nullable=False), sa.Column("modules", sa.JSON, nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default=sa.text("'pending'"), ), sa.Column("file_path", sa.String(512), nullable=True), sa.Column("file_size", sa.BigInteger, nullable=True), sa.Column("checksum_sha256", sa.String(64), nullable=True), sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("error_message", sa.Text, nullable=True), sa.Column("download_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("last_downloaded_at", sa.DateTime(timezone=True), nullable=True), sa.Column("retention_until", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["requested_by_person_id"], ["identity_persons.id"], name="fk_auditor_evidence_requester", ), sa.Index("idx_auditor_evidence_status", "status"), sa.Index("idx_auditor_evidence_requester", "requested_by_person_id"), sa.Index("idx_auditor_evidence_period", "period_start", "period_end"), ) # ════════════════════════════════════════════════════════════ # Scheduled jobs seed # ════════════════════════════════════════════════════════════ if _table_exists(bind, "scheduled_jobs"): seeds = [ { "name": "auditor.mfa_sync_daily", "display_name": "Auditor — MFA daily sync", "cron_expr": "0 8 * * *", "description": ( "Sync MFA enrollment status for every active person from " "Google / Feishu / Atlassian / GitHub providers." ), "config": {"providers": ["google", "feishu", "atlassian", "github"]}, }, { "name": "auditor.public_exposure_scan_daily", "display_name": "Auditor — public exposure daily scan", "cron_expr": "0 2 * * *", "description": ( "Scan AWS Security Groups, S3 ACLs and Atlassian site " "permissions for unintended public exposure." ), "config": {"sources": ["aws_security_group", "aws_s3", "atlassian"]}, }, { "name": "auditor.secret_rotation_check_daily", "display_name": "Auditor — secret rotation daily check", "cron_expr": "0 3 * * *", "description": ( "Refresh secret rotation ledger from API keys, cloud " "access keys and SSL certificates; flag overdue secrets." ), "config": {"default_period_days": 90}, }, { "name": "auditor.signin_anomaly_score_hourly", "display_name": "Auditor — signin anomaly hourly pull", "cron_expr": "5 * * * *", "description": ( "Pull recent signin events from Google / Feishu / Helix; " "compute anomaly score and flag suspicious events." ), "config": {"lookback_minutes": 70}, }, { "name": "auditor.pat_audit_daily", "display_name": "Auditor — PAT / MCP daily audit", "cron_expr": "0 4 * * *", "description": ( "Refresh PAT / MCP token / OAuth grant audit stream; " "flag dormant or over-scoped tokens." ), "config": {"dormant_days": 90}, }, { "name": "auditor.cert_expiry_check_daily", "display_name": "Auditor — cert / ICP daily expiry check", "cron_expr": "0 6 * * *", "description": ( "Refresh SSL/TLS, domain and ICP filing expiry; raise " "alert level at 90/60/30 days." ), "config": {"alert_thresholds_days": [90, 60, 30]}, }, { "name": "auditor.risk_score_recompute_hourly", "display_name": "Auditor — risk score hourly recompute", "cron_expr": "10 * * * *", "description": ( "Recompute per-person risk score from all upstream " "auditor signals." ), "config": {}, }, { "name": "auditor.evidence_pack_worker", "display_name": "Auditor — evidence pack background worker", "cron_expr": "*/5 * * * *", "description": ( "Drain pending auditor_evidence_packs rows and build " "PDF + ZIP packages." ), "config": {"max_per_tick": 3}, }, ] for seed in seeds: existing = bind.execute( sa.text( "SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1" ), {"n": seed["name"]}, ).scalar() if existing: continue _try_swallow( f"seed scheduled_jobs[{seed['name']}]", lambda s=seed: bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, " " handler_key, trigger_type, cron_expr, " " timezone, enabled, config, " " run_count, failure_count, " " created_at, updated_at) " "VALUES (:id, :n, :dn, :d, :hk, 'cron', :ce, " " 'UTC', 1, :cfg, 0, 0, NOW(), NOW())" ), { "id": str(uuid.uuid4()), "n": s["name"], "dn": s["display_name"], "d": s["description"], "hk": s["name"], # handler_key matches @register_scheduled_job key "ce": s["cron_expr"], "cfg": json.dumps(s["config"]), }, ), ) # ── downgrade ─────────────────────────────────────────────────────── def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "DELETE FROM scheduled_jobs WHERE name LIKE 'auditor.%'" ) ) for table in ( "auditor_evidence_packs", "auditor_risk_scores", "auditor_cert_expiries", "auditor_pat_audits", "auditor_signin_events", "auditor_secret_rotation_records", "auditor_public_exposure_findings", "auditor_privileged_role_holders", "auditor_privileged_review_cycles", "auditor_mfa_status", ): if _table_exists(bind, table): op.drop_table(table)