document-management/backend/migrations/versions/0013_document_password_lock.py

"""方案密码锁：doc_document 密码哈希 + 7 操作保护位 + 审计CHECK扩展。

Revision ID: 0013_document_password_lock
Revises: 0012_role_management
Create Date: 2026-08-20

变更要点：
1. doc_document 新增 password_hash（Argon2id，NULL=未设密）与 7 个操作保护布尔列，
   仅主案（document_type='MAIN'）可设密（CHECK 约束）；
2. sys_audit_log 的 action_type CHECK 扩展，覆盖 DOCUMENT_PASSWORD_SET /
   DOCUMENT_PASSWORD_REMOVE；
3. 降级保护：存在设密文档时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa

revision: str = "0013_document_password_lock"
down_revision: str | None = "0012_role_management"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# 0012 的全集 + 本期新增动作
AUDIT_ACTIONS_V6 = (
    "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",
)
AUDIT_ACTIONS_V5 = AUDIT_ACTIONS_V6[:-2]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"

PROTECT_COLUMNS = (
    "protect_view",
    "protect_download",
    "protect_edit",
    "protect_delete",
    "protect_config_permission",
    "protect_bind_attachment",
    "protect_unbind_attachment",
)


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 _target_values() -> tuple[str, ...]:
    return (
        "AUTH",
        "CATEGORY",
        "DOCUMENT",
        "ATTACHMENT",
        "PERMISSION",
        "ATTACHMENT_BINDING",
        "ORGANIZATION",
        "USER",
        "ROLE",
    )


def _existing_constraint_names(table: str) -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {item["name"] for item in inspector.get_check_constraints(table)}


def _drop_audit_checks(existing: set[str]) -> None:
    # op.drop_constraint 会按 metadata.naming_convention 再次前缀化约束名
    # （"ck_%(table_name)s_%(constraint_name)s"），因此必须传入原始约束名，
    # 让 SQLAlchemy 渲染出最终名；数据库实际存的也是渲染后的最终名。
    for raw_name, final_name in (
        (ACTION_CONSTRAINT, f"ck_sys_audit_log_{ACTION_CONSTRAINT}"),
        (TARGET_CONSTRAINT, f"ck_sys_audit_log_{TARGET_CONSTRAINT}"),
    ):
        if final_name in existing:
            op.drop_constraint(raw_name, "sys_audit_log", type_="check")


def upgrade() -> None:
    op.add_column(
        "doc_document",
        sa.Column("password_hash", sa.String(length=255), nullable=True),
    )
    for column in PROTECT_COLUMNS:
        op.add_column(
            "doc_document",
            sa.Column(
                column,
                sa.Boolean(),
                server_default=sa.text("0"),
                nullable=False,
            ),
        )
    op.create_check_constraint(
        "document_password_main_only",
        "doc_document",
        "(password_hash IS NULL) OR (document_type = 'MAIN')",
    )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V6),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(_target_values()),
    )


def downgrade() -> None:
    bind = op.get_bind()
    locked_rows = bind.scalar(
        sa.text("SELECT COUNT(*) FROM doc_document WHERE password_hash IS NOT NULL")
    )
    audit_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ('DOCUMENT_PASSWORD_SET', "
            "'DOCUMENT_PASSWORD_REMOVE')"
        )
    )
    if locked_rows or audit_rows:
        raise RuntimeError(
            "0013已存在设密方案或密码审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V5),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(_target_values()),
    )

    op.drop_constraint(
        "document_password_main_only", "doc_document", type_="check"
    )
    for column in reversed(PROTECT_COLUMNS):
        op.drop_column("doc_document", column)
    op.drop_column("doc_document", "password_hash")