document-management/backend/dms/services/ip_entry_service.py

"""部门IP登录限制：IP条目池维护、部门分配与登录IP校验。"""

from __future__ import annotations

import ipaddress
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import AuditAction, AuditTarget
from dms.common.errors import (
    ConflictError,
    InvalidArgumentError,
    LoginIpForbiddenError,
    ResourceNotFoundError,
)
from dms.common.pagination import PageRequest, page_result
from dms.common.response import serialize_id
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models import IpEntry, OrgIpAssignment, Organization, User
from dms.security.auth_context import get_auth_context
from dms.services.admin_permission import is_super_admin
from dms.services.audit_service import business_audit, client_ip

IP_ENTRY_CREATE_FIELDS = {"ipValue", "remark"}
IP_ENTRY_UPDATE_FIELDS = {"ipValue", "remark", "rowVersion"}
ORG_ASSIGNMENT_FIELDS = {"ipEntryIds", "rowVersion"}
IP_VALUE_MAX_LENGTH = 64
REMARK_MAX_LENGTH = 128


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _row_version(value: Any) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return value


def _id_list(value: Any, *, field: str) -> list[int]:
    if not isinstance(value, list) or any(
        not isinstance(item, str) or not item.isdecimal() or int(item) <= 0
        for item in value
    ):
        raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID数组")
    ids = [int(item) for item in value]
    if len(set(ids)) != len(ids):
        raise InvalidArgumentError(f"{field}不能包含重复ID")
    return ids


def _exact_payload(payload: Any, expected: set[str]) -> dict[str, Any]:
    if not isinstance(payload, dict) or set(payload) != expected:
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(sorted(expected))
        )
    return payload


def normalize_ip_value(value: Any) -> str:
    """规范化精确IP、CIDR网段或IP范围文本；CIDR要求主机位为0（strict=True），
    范围形式为「起IP-止IP」（两端同为IPv4或IPv6且起≤止）。"""
    if not isinstance(value, str):
        raise InvalidArgumentError("ipValue必须是字符串")
    text = value.strip()
    if not text:
        raise InvalidArgumentError("ipValue不能为空")
    if len(text) > IP_VALUE_MAX_LENGTH:
        raise InvalidArgumentError(
            f"ipValue长度不能超过{IP_VALUE_MAX_LENGTH}"
        )
    normalized = _normalize_ip_text(text)
    # IPv6等形式的规范化展开可能长于输入文本，统一复查落库长度
    if len(normalized) > IP_VALUE_MAX_LENGTH:
        raise InvalidArgumentError(
            f"ipValue长度不能超过{IP_VALUE_MAX_LENGTH}"
        )
    return normalized


def _normalize_ip_text(text: str) -> str:
    """按「含/为网段、含-为范围、否则精确IP」三分支解析规范化。"""
    try:
        if "/" in text:
            return str(ipaddress.ip_network(text, strict=True))
        if "-" in text:
            parts = text.split("-")
            if len(parts) != 2:
                raise ValueError(text)
            start = ipaddress.ip_address(parts[0])
            end = ipaddress.ip_address(parts[1])
            if start.version != end.version:
                raise InvalidArgumentError(
                    f"IP范围两端必须同为IPv4或IPv6：{text}"
                )
            if start > end:
                raise InvalidArgumentError(
                    f"IP范围起始不能大于结束：{text}"
                )
            return f"{start}-{end}"
        return str(ipaddress.ip_address(text))
    except ValueError as exc:
        if "host bits" in str(exc):
            raise InvalidArgumentError(
                f"网段要求主机位为0，请输入如192.168.1.0/24的形式：{text}"
            ) from exc
        raise InvalidArgumentError(
            f"IP地址、网段或IP范围格式非法：{text}"
        ) from exc


def _remark(value: Any) -> str:
    if value is None:
        return ""
    if not isinstance(value, str):
        raise InvalidArgumentError("remark必须是字符串")
    normalized = value.strip()
    if len(normalized) > REMARK_MAX_LENGTH:
        raise InvalidArgumentError(f"remark长度不能超过{REMARK_MAX_LENGTH}")
    return normalized


def entry_summary(entry: IpEntry) -> dict[str, object]:
    return {
        "id": serialize_id(entry.id),
        "ipValue": entry.ip_value,
        "remark": entry.remark,
        "rowVersion": entry.row_version,
    }


def _entry_row(session, entry_id: int, *, for_update: bool = False) -> IpEntry:
    stmt = select(IpEntry).where(
        IpEntry.id == entry_id, IpEntry.is_deleted.is_(False)
    )
    if for_update:
        stmt = stmt.with_for_update()
    entry = session.scalar(stmt)
    if entry is None:
        raise ResourceNotFoundError("IP条目不存在")
    return entry


def _ensure_ip_free(session, ip_value: str, *, exclude_id: int) -> None:
    duplicated = session.scalar(
        select(IpEntry.id).where(
            IpEntry.ip_value == ip_value,
            IpEntry.is_deleted.is_(False),
            IpEntry.id != exclude_id,
        )
    )
    if duplicated is not None:
        raise ConflictError("该IP已存在")


# ─── IP条目池 CRUD ─────────────────────────────────────────────────────────


def list_ip_entries(
    *,
    keyword: str | None,
    page_request: PageRequest,
) -> dict[str, object]:
    filters = [IpEntry.is_deleted.is_(False)]
    text = keyword.strip() if isinstance(keyword, str) else ""
    if text:
        pattern = f"%{text}%"
        filters.append(
            or_(IpEntry.ip_value.like(pattern), IpEntry.remark.like(pattern))
        )

    total = db.session.scalar(
        select(func.count(IpEntry.id)).where(*filters)
    ) or 0
    entries = db.session.scalars(
        select(IpEntry)
        .where(*filters)
        .order_by(IpEntry.id.asc())
        .offset(page_request.offset)
        .limit(page_request.page_size)
    ).all()
    return page_result(
        [entry_summary(entry) for entry in entries],
        page=page_request.page,
        page_size=page_request.page_size,
        total=total,
    )


def create_ip_entry(payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, IP_ENTRY_CREATE_FIELDS)
    ip_value = normalize_ip_value(body["ipValue"])
    remark = _remark(body["remark"])
    actor_id = get_auth_context().user_id

    try:
        with transaction() as session:
            _ensure_ip_free(session, ip_value, exclude_id=0)
            entry = IpEntry(
                ip_value=ip_value,
                remark=remark,
                created_by=actor_id,
                updated_by=actor_id,
            )
            session.add(entry)
            session.flush()
            session.add(
                business_audit(
                    action=AuditAction.CREATE_IP_ENTRY,
                    target=AuditTarget.IP_ENTRY,
                    target_id=entry.id,
                    target_name=entry.ip_value,
                    detail={"ipValue": entry.ip_value, "remark": entry.remark},
                )
            )
        return entry_summary(entry)
    except IntegrityError as exc:
        raise ConflictError("该IP已存在") from exc


def update_ip_entry(entry_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, IP_ENTRY_UPDATE_FIELDS)
    ip_value = normalize_ip_value(body["ipValue"])
    remark = _remark(body["remark"])
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        entry = _entry_row(session, entry_id, for_update=True)
        if entry.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": entry.row_version},
            )
        _ensure_ip_free(session, ip_value, exclude_id=entry.id)
        before = {"ipValue": entry.ip_value, "remark": entry.remark}
        entry.ip_value = ip_value
        entry.remark = remark
        entry.updated_by = actor_id
        entry.updated_at = now
        entry.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.EDIT_IP_ENTRY,
                target=AuditTarget.IP_ENTRY,
                target_id=entry.id,
                target_name=entry.ip_value,
                detail={
                    "before": before,
                    "after": {"ipValue": entry.ip_value, "remark": entry.remark},
                },
            )
        )
    return entry_summary(entry)


def delete_ip_entry(entry_id: int, row_version: int) -> dict[str, object]:
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        entry = _entry_row(session, entry_id, for_update=True)
        if entry.row_version != row_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": entry.row_version},
            )
        referenced = session.scalar(
            select(func.count(OrgIpAssignment.id)).where(
                OrgIpAssignment.ip_entry_id == entry_id,
                OrgIpAssignment.is_deleted.is_(False),
            )
        )
        if referenced:
            raise ConflictError("该IP已分配给部门，请先取消分配")

        target_name = entry.ip_value
        entry.is_deleted = True
        entry.deleted_at = now
        entry.updated_by = actor_id
        entry.updated_at = now
        entry.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.DELETE_IP_ENTRY,
                target=AuditTarget.IP_ENTRY,
                target_id=entry_id,
                target_name=target_name,
                detail={"ipValue": target_name, "rowVersion": entry.row_version},
            )
        )
    return {"id": serialize_id(entry_id), "deleted": True}


# ─── 部门分配（整体替换语义） ───────────────────────────────────────────────


def list_org_ip_assignments(organization_id: int) -> dict[str, object]:
    organization = db.session.scalar(
        select(Organization.id).where(
            Organization.id == organization_id,
            Organization.is_deleted.is_(False),
        )
    )
    if organization is None:
        raise ResourceNotFoundError("机构不存在")
    entry_ids = db.session.scalars(
        select(OrgIpAssignment.ip_entry_id)
        .where(
            OrgIpAssignment.organization_id == organization_id,
            OrgIpAssignment.is_deleted.is_(False),
        )
        .order_by(OrgIpAssignment.ip_entry_id.asc())
    ).all()
    return {"ipEntryIds": [serialize_id(value) for value in entry_ids]}


def update_org_ip_assignments(
    organization_id: int,
    payload: Any,
) -> dict[str, object]:
    body = _exact_payload(payload, ORG_ASSIGNMENT_FIELDS)
    entry_ids = _id_list(body["ipEntryIds"], field="ipEntryIds")
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        organization = session.scalar(
            select(Organization)
            .where(
                Organization.id == organization_id,
                Organization.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if organization is None:
            raise ResourceNotFoundError("机构不存在")
        if organization.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": organization.row_version},
            )
        target_ids = set(entry_ids)
        rows: list = []
        if target_ids:
            rows = session.execute(
                select(IpEntry.id, IpEntry.ip_value).where(
                    IpEntry.id.in_(sorted(target_ids)),
                    IpEntry.is_deleted.is_(False),
                )
            ).all()
            valid_ids = {row.id for row in rows}
            if target_ids - valid_ids:
                raise InvalidArgumentError("ipEntryIds包含无效IP条目")

        existing_rows = session.scalars(
            select(OrgIpAssignment).where(
                OrgIpAssignment.organization_id == organization_id,
                OrgIpAssignment.is_deleted.is_(False),
            )
        ).all()
        existing_map = {row.ip_entry_id: row for row in existing_rows}
        for ip_entry_id, row in existing_map.items():
            if ip_entry_id not in target_ids:
                row.is_deleted = True
                row.deleted_at = now
                row.updated_by = actor_id
                row.updated_at = now
        for ip_entry_id in sorted(target_ids - existing_map.keys()):
            session.add(
                OrgIpAssignment(
                    organization_id=organization_id,
                    ip_entry_id=ip_entry_id,
                    created_by=actor_id,
                    updated_by=actor_id,
                )
            )

        organization.updated_by = actor_id
        organization.updated_at = now
        organization.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.ASSIGN_ORG_IP,
                target=AuditTarget.ORGANIZATION,
                target_id=organization_id,
                target_name=organization.org_name,
                detail={
                    "ipEntries": [
                        {
                            "id": serialize_id(row.id),
                            "ipValue": row.ip_value,
                        }
                        for row in sorted(
                            rows, key=lambda item: item.id
                        )
                    ]
                    if target_ids
                    else [],
                    "rowVersion": organization.row_version,
                },
            )
        )
    return {
        "id": serialize_id(organization_id),
        "ipEntryIds": [serialize_id(value) for value in sorted(target_ids)],
        "rowVersion": organization.row_version,
    }


# ─── 登录IP校验 ────────────────────────────────────────────────────────────


def _ip_matches(
    address: ipaddress.IPv4Address | ipaddress.IPv6Address, value: str
) -> bool:
    """单条ip_value（精确IP/CIDR网段/IP范围）是否命中来源地址；
    存量数据格式异常按不匹配处理（安全优先），不中断登录流程判断。"""
    try:
        if "/" in value:
            network = ipaddress.ip_network(value)
            return address.version == network.version and address in network
        if "-" in value:
            parts = value.split("-")
            if len(parts) != 2:
                return False
            start = ipaddress.ip_address(parts[0])
            end = ipaddress.ip_address(parts[1])
            return (
                address.version == start.version == end.version
                and start <= address <= end
            )
        return address == ipaddress.ip_address(value)
    except ValueError:
        return False


def verify_login_ip(user: User) -> None:
    """密码与状态校验通过后调用：部门分配了IP列表则校验来源IP命中。

    仅内置 admin 账号豁免（管理员类别不再整体豁免，避免普通机构管理员
    绕过本部门IP限制登录）；机构未分配任何IP段仍放行。
    """
    if is_super_admin(user):
        return
    if user.organization_id is None:
        return
    ip_values = db.session.execute(
        select(IpEntry.ip_value)
        .join(
            OrgIpAssignment, OrgIpAssignment.ip_entry_id == IpEntry.id
        )
        .where(
            OrgIpAssignment.organization_id == user.organization_id,
            OrgIpAssignment.is_deleted.is_(False),
            IpEntry.is_deleted.is_(False),
        )
    ).scalars().all()
    if not ip_values:
        return

    raw_ip = client_ip()
    if raw_ip is None:
        raise LoginIpForbiddenError()
    try:
        address = ipaddress.ip_address(raw_ip.strip())
    except ValueError as exc:
        raise LoginIpForbiddenError() from exc
    for value in ip_values:
        if _ip_matches(address, value):
            return
    raise LoginIpForbiddenError()


__all__ = [
    "create_ip_entry",
    "delete_ip_entry",
    "entry_summary",
    "list_ip_entries",
    "list_org_ip_assignments",
    "normalize_ip_value",
    "update_ip_entry",
    "update_org_ip_assignments",
    "verify_login_ip",
]