"""Cloud account lifecycle — scopes / grants / role catalog / daily billing. Revision ID: 052 Revises: 051 Create Date: 2026-04-23 Introduces the multi-cloud account-lifecycle model on top of the existing ``integrations(kind='cloud')`` tenant layer: * ``cmdb_cloud_scopes`` — one row per sub-account / subscription / project / RAM account (aws_account / azure_subscription / gcp_project / aliyun_account / standalone_iam for single-tenant IAM-mode clouds like AWS SG + Tencent). Each scope has an owner (``owner_person_id``), a monthly budget, and an inactivity cutoff. * ``cmdb_cloud_access_grants`` — one row per member-on-scope with role + expiry + status. For IAM-mode scopes a grant may carry the IAM user name + access-key reference in ``detail``. * ``cmdb_cloud_roles`` — provider-side role catalog (predefined + custom) per scope, cached daily so the request form doesn't need a live API call on every render. * ``cmdb_cloud_billing_daily`` — daily cost rows keyed by (integration_id, scope_id, date). Supplements the monthly rollup; used for the scope-level drill-down page and the ``日环比异常`` alert. Also adds two ``scheduled_jobs`` handler rows per cloud integration — scopes/roles sync runs daily, daily billing runs daily. Extends the existing ``scheduled_jobs_provisioning`` module to seed them automatically on integration create/update. Registers a new employee request type ``cloud_access_request`` (used by the self-service /my/cloud/request flow) — the type list is maintained in Python (``ALLOWED_REQUEST_TYPES``), so the migration has no DB-side enum to alter. Idempotent — every DDL / seed is guarded with information_schema / NOT EXISTS. Downgrade drops the four new tables and their scheduled_jobs rows. """ from __future__ import annotations import hashlib import json as _json import uuid import sqlalchemy as sa from alembic import op revision = "052" down_revision = "051" branch_labels = None depends_on = None 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 _offset_minutes(key: str) -> int: return int(hashlib.sha256(key.encode()).hexdigest(), 16) % 60 def upgrade() -> None: bind = op.get_bind() # ── 1. cmdb_cloud_scopes ──────────────────────────────────── if not _table_exists(bind, "cmdb_cloud_scopes"): op.create_table( "cmdb_cloud_scopes", sa.Column("id", sa.String(36), primary_key=True), sa.Column("integration_id", sa.String(36), nullable=False), sa.Column("scope_type", sa.String(32), nullable=False), sa.Column("external_id", sa.String(128), nullable=False), sa.Column("display_name", sa.String(255)), sa.Column("owner_person_id", sa.String(36)), sa.Column("budget_usd_monthly", sa.Numeric(14, 2)), sa.Column( "auto_suspend_after_days", sa.Integer, nullable=False, server_default="60", ), sa.Column( "default_grant_ttl_days", sa.Integer, nullable=False, server_default="90", ), sa.Column( "status", sa.String(32), nullable=False, server_default="active", ), sa.Column("notes", sa.Text), sa.Column("raw_data", sa.JSON), sa.Column("last_seen_sync_job_id", sa.String(36)), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), 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(), ), sa.Column("deleted_at", sa.DateTime(timezone=True)), sa.UniqueConstraint( "integration_id", "external_id", name="uq_cmdb_cloud_scopes_integration_external", ), ) for fk_spec in [ ("fk_cmdb_cloud_scopes_integration", "integration_id", "integrations", "id", None), ("fk_cmdb_cloud_scopes_owner", "owner_person_id", "identity_persons", "id", "SET NULL"), ]: try: op.create_foreign_key( fk_spec[0], "cmdb_cloud_scopes", fk_spec[2], [fk_spec[1]], [fk_spec[3]], ondelete=fk_spec[4], ) except Exception: # noqa: BLE001 pass # ── 2. cmdb_cloud_access_grants ──────────────────────────── if not _table_exists(bind, "cmdb_cloud_access_grants"): op.create_table( "cmdb_cloud_access_grants", sa.Column("id", sa.String(36), primary_key=True), sa.Column("scope_id", sa.String(36), nullable=False), sa.Column("person_id", sa.String(36), nullable=False), sa.Column("role_key", sa.String(128), nullable=False), sa.Column("role_display_name", sa.String(255)), sa.Column( "status", sa.String(32), nullable=False, server_default="pending", ), sa.Column( "granted_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("expires_at", sa.DateTime(timezone=True)), sa.Column("last_renewed_at", sa.DateTime(timezone=True)), sa.Column("last_notified_stage", sa.String(16)), sa.Column("last_login_at", sa.DateTime(timezone=True)), sa.Column("granted_by_person_id", sa.String(36)), sa.Column("employee_request_id", sa.String(36)), sa.Column("external_ref", sa.String(512)), sa.Column("detail", sa.JSON), sa.Column("revoked_at", sa.DateTime(timezone=True)), sa.Column("revoked_by_person_id", sa.String(36)), sa.Column("revoke_reason", sa.String(64)), 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(), ), sa.UniqueConstraint( "scope_id", "person_id", "role_key", name="uq_cmdb_cloud_grants_scope_person_role", ), ) for fk_spec in [ ("fk_cmdb_cloud_grants_scope", "scope_id", "cmdb_cloud_scopes", "id", "CASCADE"), ("fk_cmdb_cloud_grants_person", "person_id", "identity_persons", "id", None), ("fk_cmdb_cloud_grants_granter", "granted_by_person_id", "identity_persons", "id", "SET NULL"), ("fk_cmdb_cloud_grants_revoker", "revoked_by_person_id", "identity_persons", "id", "SET NULL"), ]: try: op.create_foreign_key( fk_spec[0], "cmdb_cloud_access_grants", fk_spec[2], [fk_spec[1]], [fk_spec[3]], ondelete=fk_spec[4], ) except Exception: # noqa: BLE001 pass # Hot-path indexes for the expiry sweep + "my grants" list. try: op.create_index( "ix_cmdb_cloud_grants_expires", "cmdb_cloud_access_grants", ["status", "expires_at"], ) except Exception: # noqa: BLE001 pass try: op.create_index( "ix_cmdb_cloud_grants_person", "cmdb_cloud_access_grants", ["person_id", "status"], ) except Exception: # noqa: BLE001 pass # ── 3. cmdb_cloud_roles ───────────────────────────────────── if not _table_exists(bind, "cmdb_cloud_roles"): op.create_table( "cmdb_cloud_roles", sa.Column("id", sa.String(36), primary_key=True), sa.Column("integration_id", sa.String(36), nullable=False), # scope_id nullable for tenant-wide roles (AWS Identity # Center permission sets live at the tenant, not per # sub-account — the binding happens at assignment time). sa.Column("scope_id", sa.String(36)), sa.Column("role_key", sa.String(128), nullable=False), sa.Column("display_name", sa.String(255)), sa.Column("description", sa.Text), sa.Column( "is_custom", sa.Boolean, nullable=False, server_default=sa.false(), ), sa.Column("source", sa.String(32)), sa.Column("raw_data", sa.JSON), sa.Column("last_seen_sync_job_id", sa.String(36)), sa.Column( "first_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("deleted_at", sa.DateTime(timezone=True)), sa.UniqueConstraint( "integration_id", "scope_id", "role_key", name="uq_cmdb_cloud_roles_integration_scope_role", ), ) for fk_spec in [ ("fk_cmdb_cloud_roles_integration", "integration_id", "integrations", "id", None), ("fk_cmdb_cloud_roles_scope", "scope_id", "cmdb_cloud_scopes", "id", "CASCADE"), ]: try: op.create_foreign_key( fk_spec[0], "cmdb_cloud_roles", fk_spec[2], [fk_spec[1]], [fk_spec[3]], ondelete=fk_spec[4], ) except Exception: # noqa: BLE001 pass # ── 4. cmdb_cloud_billing_daily ──────────────────────────── if not _table_exists(bind, "cmdb_cloud_billing_daily"): op.create_table( "cmdb_cloud_billing_daily", sa.Column("id", sa.String(36), primary_key=True), sa.Column("integration_id", sa.String(36), nullable=False), # scope_id NULL = tenant-level aggregate rows (for clouds # where only the tenant total is available). sa.Column("scope_id", sa.String(36)), sa.Column("date", sa.Date, nullable=False), sa.Column( "total_cost_usd", sa.Numeric(14, 4), nullable=False, server_default="0", ), sa.Column("cost_by_service", sa.JSON), sa.Column( "currency", sa.String(8), nullable=False, server_default="USD", ), sa.Column("raw_response", sa.JSON), sa.Column( "fetched_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), 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(), ), # (integration, scope_or_null, date) is unique; MySQL # doesn't treat NULLs as equal in a unique index so we # model scope_id NULL vs non-NULL as two disjoint # uniqueness domains (aggregate rows collide per tenant×date, # scoped rows collide per scope×date). sa.UniqueConstraint( "integration_id", "scope_id", "date", name="uq_cmdb_cloud_billing_daily_key", ), ) for fk_spec in [ ("fk_cmdb_cloud_billing_daily_integration", "integration_id", "integrations", "id", None), ("fk_cmdb_cloud_billing_daily_scope", "scope_id", "cmdb_cloud_scopes", "id", "CASCADE"), ]: try: op.create_foreign_key( fk_spec[0], "cmdb_cloud_billing_daily", fk_spec[2], [fk_spec[1]], [fk_spec[3]], ondelete=fk_spec[4], ) except Exception: # noqa: BLE001 pass # ── 5. Seed scopes_sync + roles_sync + billing_daily_sync # scheduled_jobs rows for every existing cloud integration. # (access_expiry_sweep + scope_inactivity_sweep are org-wide, # no per-integration seed.) if _table_exists(bind, "scheduled_jobs") and _table_exists(bind, "integrations"): cloud_rows = bind.execute( sa.text( "SELECT id, name, slug, provider FROM integrations " "WHERE kind = 'cloud'" ) ).fetchall() for integ in cloud_rows: integ_id = str(integ[0]) integ_name = integ[1] slug = integ[2] or integ_id scopes_name = f"cmdb.scopes.{slug}" roles_name = f"cmdb.roles.{slug}" billing_name = f"cmdb.billing_daily.{slug}" _seed_job( bind, scopes_name, f"{integ_name} · 子账号/订阅同步", "每日 04:00 抓取 AWS Organizations / Azure Subscription / " "GCP Project / 阿里云资源目录下的全部子账号,写入 cmdb_cloud_scopes。" "Owner 为空的新 scope 会产生 compliance_incident。", "cmdb.scopes_sync", f"{_offset_minutes(scopes_name)} 4 * * *", _json.dumps({"integration_id": integ_id}), ) _seed_job( bind, roles_name, f"{integ_name} · 角色目录同步", "每日 04:30 抓取本 tenant 下每个 scope 的 role 目录(预定义 + 自定义)," "写入 cmdb_cloud_roles,申请页面的 role 下拉直接读这张表。", "cmdb.roles_sync", f"{_offset_minutes(roles_name)} 4 * * *", _json.dumps({"integration_id": integ_id}), ) _seed_job( bind, billing_name, f"{integ_name} · 每日账单抓取", "每日 06:00 抓取昨天的 scope 级日级账单(AWS Cost Explorer / " "Azure Cost Management / GCP Billing / 阿里云 BSS),写入 " "cmdb_cloud_billing_daily,支撑告警和每日对比。", "cmdb.billing_daily_sync", f"{_offset_minutes(billing_name)} 6 * * *", _json.dumps({"integration_id": integ_id}), ) # Org-wide sweeps (not per-integration) _seed_job( bind, "cmdb.access_expiry_sweep", "云访问过期扫描", "每日 08:00 扫 cmdb_cloud_access_grants:T-14/T-7/T-1 提醒," "T+0 自动 revoke;过期即调对应 driver 的 revoke_access 后置 status=expired。", "cmdb.access_expiry_sweep", "15 8 * * *", "{}", ) _seed_job( bind, "cmdb.scope_inactivity_sweep", "云账号不活跃扫描", "每日 08:15 扫每个 scope 下的 grants:超过 auto_suspend_after_days " "(默认 60 天)无登录通知 owner,T+14 通知 IT,T+30 自动 disable。" "口径为 IAM 控制台登录 (PasswordLastUsed) + AK 最近使用。", "cmdb.scope_inactivity_sweep", "30 8 * * *", "{}", ) # ── 6. Platform settings for the new thresholds ───────────── if _table_exists(bind, "platform_settings"): _seed_setting( bind, "cloud.grant_default_ttl_days", "90", "新建 grant 的默认过期时长(天)。可在每个 scope 上覆盖。", ) _seed_setting( bind, "cloud.grant_renewal_days", "90", "用户自助续期时,从续期时刻加多少天。", ) _seed_setting( bind, "cloud.grant_notify_days", "[14, 7, 1]", "过期前提醒的天数(JSON 列表)。", ) _seed_setting( bind, "cloud.inactivity_threshold_days", "60", "Grant 多少天无登录会触发不活跃通知。scope 可覆盖。", ) _seed_setting( bind, "cloud.inactivity_escalation_days", "14", "不活跃通知发出 N 天后若仍无登录,通知 IT。", ) _seed_setting( bind, "cloud.inactivity_disable_days", "30", "不活跃通知发出 N 天后若仍无登录,自动 disable grant。", ) _seed_setting( bind, "cloud.daily_spike_multiplier", "1.5", "当日花费 > 7 日均值 × multiplier 且绝对值 > spike_min_usd 即告警。", ) _seed_setting( bind, "cloud.daily_spike_min_usd", "10", "日环比异常告警的最小绝对值(美元),避免小额抖动误报。", ) _seed_setting( bind, "cloud.budget_alert_percents", "[80, 90, 100]", "月预算百分比告警阈值(JSON 列表),每个百分比每月最多 1 次。", ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "DELETE FROM scheduled_jobs WHERE handler_key IN " "('cmdb.scopes_sync', 'cmdb.roles_sync', " "'cmdb.billing_daily_sync', 'cmdb.access_expiry_sweep', " "'cmdb.scope_inactivity_sweep')" ) ) if _table_exists(bind, "cmdb_cloud_billing_daily"): op.drop_table("cmdb_cloud_billing_daily") if _table_exists(bind, "cmdb_cloud_roles"): op.drop_table("cmdb_cloud_roles") if _table_exists(bind, "cmdb_cloud_access_grants"): op.drop_table("cmdb_cloud_access_grants") if _table_exists(bind, "cmdb_cloud_scopes"): op.drop_table("cmdb_cloud_scopes") def _seed_job( bind, name: str, display_name: str, description: str, handler_key: str, cron_expr: str, config_json: str, ) -> None: 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) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), { "id": str(uuid.uuid4()), "name": name, "display": display_name, "desc": description, "handler": handler_key, "cron": cron_expr, "config": config_json, }, ) def _seed_setting(bind, key: str, value_str: str, description: str) -> None: bind.execute( sa.text( "INSERT INTO platform_settings (`key`, value, description, " "created_at, updated_at) " "SELECT :k, JSON_QUOTE(:v), :d, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM platform_settings WHERE `key` = :k)" ), {"k": key, "v": value_str, "d": description}, )