"""部门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")