/
azathd
/
mutiagent
Обзор
Документация
Войти
/
azathd
/
mutiagent
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/agents/executor/audit_store.py
65 строк
2 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47
359c81e
Код
Авторство
О чём код?
"""Append-only audit log with hash chain.""" from __future__ import annotations import hashlib import json import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.common.orm import AuditLog async def last_hash(session: AsyncSession) -> str: row = await session.scalar(select(AuditLog).order_by(AuditLog.created_at.desc()).limit(1)) return row.entry_hash if row else "" def compute_hash(prev: str, payload: dict[str, Any]) -> str: blob = prev + json.dumps(payload, sort_keys=True, default=str) return hashlib.sha256(blob.encode()).hexdigest() async def append_audit( session: AsyncSession, *, action_id: str, action_type: str, actor: str, target: str, payload: dict[str, Any], result: dict[str, Any], ) -> AuditLog: prev = await last_hash(session) entry_payload = { "action_id": action_id, "action_type": action_type, "actor": actor, "target": target, "payload": payload, "result": result, "ts": datetime.now(tz=UTC).isoformat(), } entry_hash = compute_hash(prev, entry_payload) row = AuditLog( id=uuid.uuid4(), action_id=action_id, action_type=action_type, actor=actor, target=target, payload=payload, result=result, prev_hash=prev, entry_hash=entry_hash, ) session.add(row) await session.commit() return row async def find_by_action_id(session: AsyncSession, action_id: str) -> AuditLog | None: return await session.scalar(select(AuditLog).where(AuditLog.action_id == action_id))