document-management/backend/migrations/versions/0028_backup_audit_actions.py

"""备份功能审计动作与目标类型：sys_audit_log 两组 CHECK 白名单扩展。

Revision ID: 0028_backup_audit_actions
Revises: 0027_ui_theme
Create Date: 2026-09-06

变更要点：
1. 备份功能（契约1.21）新增审计动作 BACKUP_CREATE（手动/定时触发备份均留痕；
   恢复为离线 CLI，不经过 HTTP 审计链路，由 CLI 自行打印与插入）；
2. 重建 sys_audit_log 的 audit_action_type CHECK 约束（0025 的 V8 全集 + 1 = 39 项）；
3. 新增审计目标类型 BACKUP，重建 audit_target_type CHECK 约束（原 10 项 + 1 = 11 项）；
4. 达梦兼容：约束名在 DM 下渲染为大写（CK_SYS_AUDIT_LOG_AUDIT_ACTION_TYPE），
   探测按小写比较后 drop 原始逻辑名（naming convention 渲染最终名）；
5. 升级幂等：drop-if-exists + 重建（与 Flyway V13 并行执行结果一致，可重复跑）；
6. 降级拒绝已存在 BACKUP_CREATE 审计记录的库。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0028_backup_audit_actions"
down_revision: str | None = "0027_ui_theme"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

# 0025 的 V8 全集 + 本期新增动作
AUDIT_ACTIONS_V9 = (
    "LOGIN",
    "LOGOUT",
    "VIEW_DOCUMENT",
    "DOWNLOAD_DOCUMENT",
    "UPLOAD_DOCUMENT",
    "BATCH_IMPORT",
    "EDIT_DOCUMENT",
    "DELETE_DOCUMENT",
    "CREATE_CATEGORY",
    "EDIT_CATEGORY",
    "DELETE_CATEGORY",
    "CHANGE_PERMISSION",
    "BIND_ATTACHMENT",
    "UNBIND_ATTACHMENT",
    "RESTORE_DOCUMENT",
    "CREATE_ORGANIZATION",
    "EDIT_ORGANIZATION",
    "DELETE_ORGANIZATION",
    "CREATE_USER",
    "EDIT_USER",
    "DELETE_USER",
    "RESET_USER_PASSWORD",
    "REORDER_DOCUMENT",
    "REORDER_ATTACHMENT",
    "REORDER_CATEGORY",
    "CREATE_ROLE",
    "EDIT_ROLE",
    "DELETE_ROLE",
    "CHANGE_ROLE_PERMISSION",
    "CHANGE_ROLE_USER",
    "DOCUMENT_PASSWORD_SET",
    "DOCUMENT_PASSWORD_REMOVE",
    "CREATE_IP_ENTRY",
    "EDIT_IP_ENTRY",
    "DELETE_IP_ENTRY",
    "ASSIGN_ORG_IP",
    "MOVE_DOCUMENT",
    "MOVE_ATTACHMENT",
    "BACKUP_CREATE",
)
AUDIT_ACTIONS_V8 = AUDIT_ACTIONS_V9[:-1]

# 现行 target_type 白名单（0015 起 10 项）+ 本期新增 BACKUP
AUDIT_TARGETS_V2 = (
    "AUTH",
    "CATEGORY",
    "DOCUMENT",
    "ATTACHMENT",
    "PERMISSION",
    "ATTACHMENT_BINDING",
    "ORGANIZATION",
    "USER",
    "ROLE",
    "IP_ENTRY",
    "BACKUP",
)
AUDIT_TARGETS_V1 = AUDIT_TARGETS_V2[:-1]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"


def _check_sql(values: tuple[str, ...]) -> str:
    # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险
    inline = ", ".join(f"'{v}'" for v in values)
    return f"action_type IN ({inline})"


def _target_check_sql(values: tuple[str, ...]) -> str:
    inline = ", ".join(f"'{v}'" for v in values)
    return f"(target_type IS NULL OR target_type IN ({inline}))"


def _existing_check_names() -> set[str]:
    # op.drop_constraint 按 metadata.naming_convention 前缀化约束名，
    # 须传原始约束名渲染最终名；达梦返回的约束名为大写，统一小写比较。
    return {
        item["name"].lower()
        for item in sa.inspect(op.get_bind()).get_check_constraints("sys_audit_log")
    }


def _drop_checks(names: set[str]) -> None:
    for raw_name in (ACTION_CONSTRAINT, TARGET_CONSTRAINT):
        if f"ck_sys_audit_log_{raw_name}" in names:
            op.drop_constraint(raw_name, "sys_audit_log", type_="check")


def _has_backup_records() -> bool:
    result = op.get_bind().execute(
        sa.text("SELECT COUNT(*) FROM sys_audit_log WHERE action_type = 'BACKUP_CREATE'")
    )
    return bool(result.scalar())


def upgrade() -> None:
    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V9),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V2),
    )


def downgrade() -> None:
    if _has_backup_records():
        raise RuntimeError("已存在 BACKUP_CREATE 审计记录，拒绝有损降级")
    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V8),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V1),
    )