Неизменяемый журнал событий (chutils.audit)
Модуль chutils.audit предоставляет средства для ведения криптографически защищенного (Immutable) журнала событий
безопасности и бизнес-логики (Audit Log). Журнал обеспечивает соответствие стандартам безопасности (compliance) путем
связывания записей в неизменяемую цепочку хэшей (hash chain), что позволяет мгновенно обнаружить несанкционированное
изменение или удаление записей.
Ключевые возможности
- Криптографическая цепочка целостности: каждая новая запись содержит SHA-256 хэш предыдущей записи. При проверке
целостности (
verify_integrity()) цепочка последовательно пересчитывается и валидируется. - Автоматический сбор окружения: для каждой записи автоматически собирается контекст исполнения:
hostname,pid,thread_name. - Разнообразные бэкенды хранения:
FileBackend— хранение в виде append-only JSON Lines (JSONL) файла.SqliteBackend— хранение в локальной базе данных SQLite (без внешних ORM-зависимостей, в режиме WAL).PostgresBackend— хранение в PostgreSQL (поддерживает любое стандартное DBAPI2-совместимое соединение).
- Ленивый импорт: отсутствие библиотеки Pydantic в окружении не мешает работе остальной части
chutils, но классAuditEventтребует наличияpydantic. - Потокобезопасность: все бэкенды потокобезопасны и используют блокировки при конкурентной записи и расчете хэшей.
Схема записи (AuditEvent)
Каждая запись представляется моделью AuditEvent со следующими полями:
id(str): Уникальный UUID записи.timestamp(datetime): Время создания записи (UTC).actor(str): Субъект действия (например,user_123илиsystem).action(str): Тип совершаемого действия (например,user.login).target(str | None): Объект действия (например,document_456).status(str): Статус операции (success/failed).details(dict): Произвольные дополнительные данные о событии.env(dict): Системное окружение (заполняется автоматически).prev_hash(str): Хэш предыдущей записи (пустая строка для первой записи).hash(str): Итоговый SHA-256 хэш текущей записи.
Примеры использования
1. Низкоуровневый API записи в JSONL-файл
from chutils import FileBackend
# Инициализируем бэкенд
backend = FileBackend("logs/audit.jsonl")
# Записываем событие
backend.log(
action="user.login",
actor="user_123",
target="web_app",
status="success",
details={"ip": "192.168.1.50"}
)
# Проверяем целостность файла
try:
backend.verify_integrity()
print("Целостность журнала подтверждена.")
except Exception as e:
print(f"Журнал поврежден: {e}")
2. Контекстный менеджер audit_context
Используется для оборачивания блоков кода. Позволяет изменять status и details в процессе выполнения блока. В случае
необработанного исключения автоматически ставит статус "failed".
from chutils import SqliteBackend, audit_context
backend = SqliteBackend("logs/audit.db")
with audit_context(action="payment.process", actor="user_99", target="invoice_555", backend=backend) as ctx:
# Выполнение бизнес-логики...
ctx.details["amount"] = 1500
ctx.details["currency"] = "USD"
# Если в блоке возникнет ошибка, запись получит статус "failed"
3. Декоратор @audit_event
Позволяет автоматически регистрировать вызовы синхронных и асинхронных функций. Поддерживает динамическое разрешение
параметров actor и target с помощью callable-объектов.
from chutils import FileBackend, audit_event
backend = FileBackend("logs/audit.jsonl")
# Пример динамического определения target по аргументам функции
def get_doc_target(doc_id: str, *args, **kwargs) -> str:
return f"document_{doc_id}"
@audit_event(action="document.delete", actor="admin", target=get_doc_target, backend=backend)
def delete_document(doc_id: str) -> None:
print(f"Документ {doc_id} удален.")
# Вызов функции автоматически создаст запись в журнале аудита
delete_document("doc_abc_123")
Ограничения
- Криптографическая защита основана на SHA-256 цепочке (без асимметричных ключей подписи).
- Модуль
chutils.auditпредназначен для хранения последовательностей событий. В распределенных системах для гарантии порядка записей рекомендуется использовать централизованную БД.