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

"""认证审计日志构造与独立失败事务。"""

from __future__ import annotations

import ipaddress
import socket
import time
from typing import Any

from flask import request

from dms.common.enums import AuditAction, AuditResult, AuditTarget
from dms.common.request_context import get_request_id
from dms.extensions import db
from dms.models import AuditLog, User
from dms.security.auth_context import get_auth_context


def client_ip() -> str | None:
    forwarded = request.headers.get("X-Forwarded-For", "")
    value = forwarded.split(",", 1)[0].strip() if forwarded else request.remote_addr
    if not value:
        return None
    # 同机浏览器 localhost 访问时 TCP 对端为回环地址（::1/127.0.0.1），
    # 审计展示替换为本机网卡 IPv4（部署形态无入向代理；有代理时以 X-Forwarded-For 为准）
    if _is_loopback(value):
        value = _primary_local_address()
    return value[:64]


def _is_loopback(ip: str) -> bool:
    try:
        # 去掉 IPv6 zone id；覆盖 ::1、0:0:0:0:0:0:0:1、127.* 及 v4 映射形态
        # （::ffff:a.b.c.d 的 is_loopback 随 Python 版本存在差异，显式解映射后判定）
        parsed = ipaddress.ip_address(ip.split("%", 1)[0])
        if isinstance(parsed, ipaddress.IPv6Address) and parsed.ipv4_mapped is not None:
            parsed = parsed.ipv4_mapped
        return parsed.is_loopback
    except ValueError:
        return False


# 本机网卡首选 IPv4 缓存（TTL 1 分钟，避免每次审计重复选路）
_primary_address_cache: tuple[str, float] | None = None


def _primary_local_address() -> str:
    """本机首选非回环 IPv4；无可用网卡时回退 127.0.0.1（兼作 ::1 规范化显示）。"""
    global _primary_address_cache
    now = time.monotonic()
    if _primary_address_cache is not None and now < _primary_address_cache[1]:
        return _primary_address_cache[0]
    try:
        # UDP connect 不实际发包，仅让内核按路由表选出站网卡（192.0.2.1 为 RFC 5737 文档地址）
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
            probe.connect(("192.0.2.1", 80))
            resolved = probe.getsockname()[0]
        if _is_loopback(resolved):
            resolved = "127.0.0.1"
    except OSError:
        resolved = "127.0.0.1"
    _primary_address_cache = (resolved, now + 60)
    return resolved


def auth_audit(
    *,
    action: AuditAction,
    result: AuditResult,
    user: User | None = None,
    attempted_username: str | None = None,
    failure_reason: str | None = None,
    detail: dict[str, Any] | None = None,
) -> AuditLog:
    username = user.username if user is not None else attempted_username
    return AuditLog(
        user_id=user.id if user is not None else None,
        username=username,
        real_name=user.real_name if user is not None else None,
        organization_id=user.organization_id if user is not None else None,
        organization_name=user.organization_name if user is not None else None,
        action_type=action.value,
        target_type=AuditTarget.AUTH.value,
        target_id=user.id if user is not None else None,
        target_name=username,
        operation_result=result.value,
        failure_reason=failure_reason,
        client_ip=client_ip(),
        user_agent=(request.user_agent.string or "")[:500] or None,
        request_id=get_request_id(),
        operation_detail=detail,
    )


def business_audit(
    *,
    action: AuditAction,
    target: AuditTarget,
    target_id: int,
    target_name: str,
    detail: dict[str, Any] | None = None,
) -> AuditLog:
    """使用请求级认证快照构造修改类业务成功审计。"""
    context = get_auth_context()
    return AuditLog(
        user_id=context.user_id,
        username=context.username,
        real_name=context.real_name,
        organization_id=context.organization_id,
        organization_name=context.organization_name,
        action_type=action.value,
        target_type=target.value,
        target_id=target_id,
        target_name=target_name,
        operation_result=AuditResult.SUCCESS.value,
        failure_reason=None,
        client_ip=client_ip(),
        user_agent=(request.user_agent.string or "")[:500] or None,
        request_id=get_request_id(),
        operation_detail=detail,
    )


def record_login_failure(
    *,
    user: User | None,
    attempted_username: str,
    failure_reason: str,
) -> None:
    """在登录主流程之外用短事务持久化失败审计。"""
    db.session.rollback()
    try:
        db.session.add(
            auth_audit(
                action=AuditAction.LOGIN,
                result=AuditResult.FAILURE,
                user=user,
                attempted_username=attempted_username,
                failure_reason=failure_reason,
                detail={"username": attempted_username},
            )
        )
        db.session.commit()
    except Exception:
        db.session.rollback()
        raise