MeshSate/app/services/audit.py

"""审计日志服务，带哈希链防篡改。"""

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.config import get_settings
from app.models import AuditLog

_settings = get_settings()


def _canonical(detail: Dict[str, Any]) -> str:
    return json.dumps(detail, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)


def compute_row_hash(
    ts: datetime,
    actor: str,
    actor_role: str,
    action: str,
    target: str,
    channel: str,
    client_ip: str,
    outcome: str,
    detail: Dict[str, Any],
    prev_hash: str,
) -> str:
    """计算审计记录的链式哈希。

    哈希输入包含除 ``row_hash`` 外的全部字段以及前一条记录的哈希，
    任何字段被修改都会导致链断裂。
    """
    payload = "|".join(
        [
            ts.astimezone(timezone.utc).isoformat(),
            actor,
            actor_role,
            action,
            target,
            channel,
            client_ip,
            outcome,
            _canonical(detail),
            prev_hash,
        ]
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def record(
    db: Session,
    *,
    action: str,
    actor: str = "system",
    actor_role: str = "system",
    target: str = "",
    channel: str = "",
    client_ip: str = "",
    outcome: str = "success",
    detail: Optional[Dict[str, Any]] = None,
) -> AuditLog:
    """写入一条审计日志。"""
    detail = detail or {}

    prev_hash = ""
    if _settings.audit_hash_chain:
        last = db.execute(
            select(AuditLog).order_by(AuditLog.id.desc()).limit(1)
        ).scalar_one_or_none()
        if last is not None:
            prev_hash = last.row_hash or ""

    ts = datetime.now(timezone.utc)
    row_hash = ""
    if _settings.audit_hash_chain:
        row_hash = compute_row_hash(
            ts, actor, actor_role, action, target, channel, client_ip, outcome, detail, prev_hash
        )

    entry = AuditLog(
        ts=ts,
        actor=actor,
        actor_role=actor_role,
        action=action,
        target=target,
        channel=channel,
        client_ip=client_ip,
        outcome=outcome,
        detail=detail,
        prev_hash=prev_hash,
        row_hash=row_hash,
    )
    db.add(entry)
    db.commit()
    db.refresh(entry)
    return entry


def verify_chain(db: Session, limit: int = 1000) -> Dict[str, Any]:
    """校验最近若干条审计记录的哈希链完整性。"""
    rows = list(
        db.execute(select(AuditLog).order_by(AuditLog.id.desc()).limit(limit)).scalars().all()
    )
    rows.reverse()

    broken_at: Optional[int] = None
    prev_hash = ""
    for row in rows:
        if row.prev_hash != prev_hash:
            broken_at = row.id
            break
        expected = compute_row_hash(
            row.ts,
            row.actor,
            row.actor_role,
            row.action,
            row.target,
            row.channel,
            row.client_ip,
            row.outcome,
            row.detail or {},
            row.prev_hash,
        )
        if expected != row.row_hash:
            broken_at = row.id
            break
        prev_hash = row.row_hash

    return {
        "checked": len(rows),
        "intact": broken_at is None,
        "broken_at_id": broken_at,
        "verified_at": datetime.now(timezone.utc).isoformat(),
    }


def purge_expired(db: Session) -> int:
    """清理超出留存期的审计记录，返回删除条数。

    等保三级要求审计记录留存不少于 180 天，删除动作本身也会记入审计日志。
    """
    threshold = datetime.now(timezone.utc) - timedelta(days=_settings.audit_retention_days)
    rows = db.execute(select(AuditLog).where(AuditLog.ts < threshold)).scalars().all()
    for row in rows:
        db.delete(row)
    if rows:
        db.commit()
    return len(rows)