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

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

from __future__ import annotations

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
    return value[:64] if value else None


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