document-management/backend/migrations/versions/0015_login_ip_restriction.py

"""部门IP登录限制：sys_ip_entry + sys_org_ip_assignment + 审计CHECK扩展。

Revision ID: 0015_login_ip_restriction
Revises: 0014_document_relations
Create Date: 2026-08-22

变更要点：
1. 新建 sys_ip_entry：全局单一IP池，每条为规范化后的精确IP或CIDR网段文本，
   软删唯一约束保证 ip_value 在有效行内唯一；
2. 新建 sys_org_ip_assignment：部门可登录IP分配（整体替换语义），
   至少一行未删记录即视为该部门受限；
3. sys_audit_log 的 action_type CHECK 扩展（CREATE_IP_ENTRY / EDIT_IP_ENTRY /
   DELETE_IP_ENTRY / ASSIGN_ORG_IP），target_type CHECK 新增 IP_ENTRY；
4. 降级保护：存在IP条目、分配行或IP审计记录时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

revision: str = "0015_login_ip_restriction"
down_revision: str | None = "0014_document_relations"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


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

AUDIT_TARGETS_V4 = (
    "AUTH",
    "CATEGORY",
    "DOCUMENT",
    "ATTACHMENT",
    "PERMISSION",
    "ATTACHMENT_BINDING",
    "ORGANIZATION",
    "USER",
    "ROLE",
    "IP_ENTRY",
)
AUDIT_TARGETS_V3 = AUDIT_TARGETS_V4[:-1]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"


def _business_columns() -> list[sa.Column]:
    return [
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("created_by", sa.BigInteger(), nullable=True),
        sa.Column("updated_by", sa.BigInteger(), nullable=True),
        sa.Column(
            "created_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column("deleted_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.Column(
            "is_deleted",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "row_version",
            sa.Integer(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    ]


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_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 _recreate_audit_checks(actions: tuple[str, ...], targets: tuple[str, ...]) -> None:
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(actions),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(targets),
    )


def upgrade() -> None:
    op.create_table(
        "sys_ip_entry",
        *_business_columns(),
        sa.Column("ip_value", sa.String(length=64), nullable=False),
        sa.Column(
            "remark",
            sa.String(length=128),
            server_default=sa.text("''"),
            nullable=False,
        ),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.UniqueConstraint(
            "ip_value",
            "active_marker",
            name="uq_sys_ip_entry_ip_value_active",
        ),
    )
    op.create_index(
        "ix_sys_ip_entry_active",
        "sys_ip_entry",
        ["is_deleted", "id"],
    )

    op.create_table(
        "sys_org_ip_assignment",
        *_business_columns(),
        sa.Column("organization_id", sa.BigInteger(), nullable=False),
        sa.Column("ip_entry_id", sa.BigInteger(), nullable=False),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.ForeignKeyConstraint(
            ["organization_id"],
            ["sys_organization.id"],
            name="fk_sys_org_ip_assignment_organization_id_sys_organization",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["ip_entry_id"],
            ["sys_ip_entry.id"],
            name="fk_sys_org_ip_assignment_ip_entry_id_sys_ip_entry",
            ondelete="RESTRICT",
        ),
        sa.UniqueConstraint(
            "organization_id",
            "ip_entry_id",
            "active_marker",
            name="uq_sys_org_ip_assignment_active",
        ),
    )
    op.create_index(
        "ix_sys_org_ip_assignment_org_active",
        "sys_org_ip_assignment",
        ["organization_id", "is_deleted"],
    )
    op.create_index(
        "ix_sys_org_ip_assignment_entry_active",
        "sys_org_ip_assignment",
        ["ip_entry_id", "is_deleted"],
    )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    _recreate_audit_checks(AUDIT_ACTIONS_V7, AUDIT_TARGETS_V4)


def downgrade() -> None:
    bind = op.get_bind()
    ip_rows = bind.scalar(sa.text("SELECT COUNT(*) FROM sys_ip_entry"))
    assignment_rows = bind.scalar(
        sa.text("SELECT COUNT(*) FROM sys_org_ip_assignment")
    )
    audit_rows = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type IN ('CREATE_IP_ENTRY', 'EDIT_IP_ENTRY', "
            "'DELETE_IP_ENTRY', 'ASSIGN_ORG_IP')"
        )
    )
    if ip_rows or assignment_rows or audit_rows:
        raise RuntimeError(
            "0015已存在IP条目、部门分配或IP审计记录，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    _drop_audit_checks(_existing_constraint_names("sys_audit_log"))
    _recreate_audit_checks(AUDIT_ACTIONS_V6, AUDIT_TARGETS_V3)

    op.drop_table("sys_org_ip_assignment")
    op.drop_table("sys_ip_entry")