"""部门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", ]