Опирается на правила: AUTH-15, AUTH-12, AUTH-16 из контракта Auth Patterns → раздел 6. Аудит admin-команд.

Важно знать

  • Каждая state-changing команда от роли admin обязана писать строку в *_audit_log; endpoint без audit при admin-вызове — нарушение AUTH-15.
  • Поля: actor_id (кто), occurred_at (когда), action (что), resource_type + resource_id (к чему), metadata JSONB (детали).
  • Реализация — декоратор @admin_audit или явный вызов в Handler; оба варианта допустимы.
  • Та же сессия SQLAlchemy / одна транзакция: бизнес-write + audit commit вместе, либо оба rollback.
  • При ABAC override (admin отменил чужой заказ) — AUTH-12 требует audit обязательно.
  • Append-only: только INSERT; REVOKE UPDATE, DELETE ON *_audit_log FROM app_role.
  • PII в metadata — только actor_id (sub), не email; AUTH-16 запрещает plain PII в персистентных артефактах.
  • Без audit компрометация admin-аккаунта = невидимый ущерб без возможности расследования.

Admin-роль обходит ABAC и открывает полный доступ к агрегатам — отмена чужого заказа, возврат, ручное изменение статуса. Без audit log это дыра: нельзя восстановить кто, когда, что сделал и к какому ресурсу. Каждое admin-действие должно быть воспроизводимо по записи.

Схема audit-таблицы

AUTH-15 — одна таблица на сервис или per-aggregate.

CREATE TABLE admin_audit_log (
    id            bigserial PRIMARY KEY,
    actor_id      text        NOT NULL,
    action        text        NOT NULL,
    resource_type text        NOT NULL,
    resource_id   text        NOT NULL,
    occurred_at   timestamptz NOT NULL DEFAULT now(),
    metadata      jsonb       NOT NULL DEFAULT '{}',
    request_id    text,
    trace_id      text
);

CREATE INDEX ix_admin_audit_actor    ON admin_audit_log (actor_id, occurred_at DESC);
CREATE INDEX ix_admin_audit_resource ON admin_audit_log (resource_type, resource_id);

REVOKE UPDATE, DELETE ON admin_audit_log FROM app_role;

metadata JSONB — расширяемое поле: previous_status, new_status, reason, owner_customer_id. Не структурируем колонки заранее — добавляем в metadata по необходимости.

Вариант per-aggregate (order_audit_log, product_audit_log) уместен, когда схема metadata принципиально различается и хочется типизированных индексов.

Модель и репозиторий

# adapters/out/db/audit_models.py
import uuid
from datetime import datetime, UTC
from sqlalchemy import Column, BigInteger, Text, DateTime, Index
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class AdminAuditLog(Base):
    __tablename__ = "admin_audit_log"

    id            = Column(BigInteger, primary_key=True, autoincrement=True)
    actor_id      = Column(Text, nullable=False)
    action        = Column(Text, nullable=False)
    resource_type = Column(Text, nullable=False)
    resource_id   = Column(Text, nullable=False)
    occurred_at   = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
    metadata      = Column(JSONB, nullable=False, default=dict)
    request_id    = Column(Text)
    trace_id      = Column(Text)

    __table_args__ = (
        Index("ix_admin_audit_actor",    "actor_id", "occurred_at"),
        Index("ix_admin_audit_resource", "resource_type", "resource_id"),
    )

# adapters/out/db/audit_repository.py
from dataclasses import dataclass
from datetime import datetime, UTC
from sqlalchemy.ext.asyncio import AsyncSession
from .audit_models import AdminAuditLog

@dataclass
class AuditRecord:
    actor_id:      str
    action:        str
    resource_type: str
    resource_id:   str
    metadata:      dict
    request_id:    str | None = None
    trace_id:      str | None = None

class AdminAuditRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def append(self, record: AuditRecord) -> None:
        self._session.add(AdminAuditLog(
            actor_id      = record.actor_id,
            action        = record.action,
            resource_type = record.resource_type,
            resource_id   = record.resource_id,
            occurred_at   = datetime.now(UTC),
            metadata      = record.metadata,
            request_id    = record.request_id,
            trace_id      = record.trace_id,
        ))

append не вызывает commit — это обязанность Handler-а (или AsyncSession в autocommit=False-режиме). Audit и бизнес-write оказываются в одном flush.

Реализация — декоратор

Для типовых команд admin удобен декоратор: нельзя забыть добавить audit при добавлении нового Handler-а.

# application/audit.py
import functools
from collections.abc import Callable
from typing import Any

def admin_audit(action: str, resource_type: str, resource_id_attr: str = "id"):
    """
    Оборачивает async-метод Handler-а: если principal.is_admin,
    записывает строку в audit_log после успешного выполнения.
    Требует, чтобы первый аргумент (command) имел атрибут resource_id_attr,
    а Handler — поле self._audit: AdminAuditRepository.
    """
    def decorator(fn: Callable) -> Callable:
        @functools.wraps(fn)
        async def wrapper(self, command, principal, *args, **kwargs) -> Any:
            result = await fn(self, command, principal, *args, **kwargs)
            if principal.is_admin:
                resource_id = str(getattr(command, resource_id_attr))
                await self._audit.append(AuditRecord(
                    actor_id      = principal.sub,
                    action        = action,
                    resource_type = resource_type,
                    resource_id   = resource_id,
                    metadata      = {},
                    request_id    = getattr(command, "request_id", None),
                    trace_id      = getattr(command, "trace_id", None),
                ))
            return result
        return wrapper
    return decorator

# application/cancel_order_handler.py
from .audit import admin_audit, AuditRecord
from domain.order import Order, OrderStatus
from ports.out.order_repository import OrderRepository
from adapters.out.db.audit_repository import AdminAuditRepository

class CancelOrderHandler:
    def __init__(
        self,
        orders: OrderRepository,
        audit:  AdminAuditRepository,
    ) -> None:
        self._orders = orders
        self._audit  = audit

    @admin_audit(action="cancel-order", resource_type="Order", resource_id_attr="order_id")
    async def handle(self, command: CancelOrderCommand, principal: Principal) -> Order:
        order = await self._orders.find_by_id(command.order_id)
        if order is None:
            raise OrderNotFoundError(command.order_id)
        if not principal.is_admin and order.customer_id != principal.sub:
            raise ForbiddenError()
        order.cancel()
        await self._orders.save(order)
        return order

@admin_audit срабатывает только при principal.is_admin, не трогает обычных пользователей.

Реализация — явный вызов

Когда metadata содержит специфический контекст (статус до/после, владелец), явный вызов читается проще:

# application/cancel_order_handler.py
class CancelOrderHandler:
    def __init__(
        self,
        orders: OrderRepository,
        audit:  AdminAuditRepository,
    ) -> None:
        self._orders = orders
        self._audit  = audit

    async def handle(self, command: CancelOrderCommand, principal: Principal) -> Order:
        order = await self._orders.find_by_id(command.order_id)
        if order is None:
            raise OrderNotFoundError(command.order_id)
        if not principal.is_admin and order.customer_id != principal.sub:
            raise ForbiddenError()

        previous_status: OrderStatus = order.status
        order.cancel()
        await self._orders.save(order)

        if principal.is_admin:
            await self._audit.append(AuditRecord(
                actor_id      = principal.sub,
                action        = "cancel-order",
                resource_type = "Order",
                resource_id   = str(command.order_id),
                metadata      = {
                    "previous_status":   previous_status.value,
                    "new_status":        order.status.value,
                    "owner_customer_id": str(order.customer_id),
                },
                request_id = command.request_id,
                trace_id   = command.trace_id,
            ))

        return order

Явный вызов — когда метаданные требуют previous_status или других данных, доступных только внутри метода. Декоратор — для типовых команд без сложного контекста.

Одна транзакция

Audit пишется в той же AsyncSession, что и бизнес-операция. Транзакция управляется на уровне FastAPI-зависимости или middleware.

# adapters/in/http/dependencies.py
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

@asynccontextmanager
async def db_session(factory: async_sessionmaker[AsyncSession]):
    async with factory() as session:
        async with session.begin():
            yield session

# adapters/in/http/routers/orders.py
from fastapi import APIRouter, Depends
from adapters.in.http.security import require_roles

router = APIRouter()

@router.delete("/orders/{order_id}")
async def cancel_order(
    order_id:  uuid.UUID,
    principal: Principal = Depends(require_roles("admin", "customer")),
    session:   AsyncSession = Depends(get_session),
):
    handler = CancelOrderHandler(
        orders = SqlOrderRepository(session),
        audit  = AdminAuditRepository(session),
    )
    return await handler.handle(CancelOrderCommand(order_id=order_id), principal)

session.begin() охватывает orders.save(order) + audit.append(...) — оба в одном COMMIT. Если Handler бросил исключение — оба откатились.

Не выносим audit в отдельный сервис через очередь: потеря события между commit и publish даёт missing audit для реальных изменений.

Refund и product-команды

Те же правила применяются к любой admin-команде на агрегате Product или Customer.

# application/force_publish_product_handler.py
class ForcePublishProductHandler:
    def __init__(self, products: ProductRepository, audit: AdminAuditRepository) -> None:
        self._products = products
        self._audit    = audit

    @admin_audit(action="force-publish-product", resource_type="Product", resource_id_attr="product_id")
    async def handle(self, command: ForcePublishProductCommand, principal: Principal) -> Product:
        product = await self._products.find_by_id(command.product_id)
        if product is None:
            raise ProductNotFoundError(command.product_id)
        product.publish()
        await self._products.save(product)
        return product

# application/block_customer_handler.py
class BlockCustomerHandler:
    def __init__(self, customers: CustomerRepository, audit: AdminAuditRepository) -> None:
        self._customers = customers
        self._audit     = audit

    async def handle(self, command: BlockCustomerCommand, principal: Principal) -> Customer:
        customer = await self._customers.find_by_id(command.customer_id)
        if customer is None:
            raise CustomerNotFoundError(command.customer_id)

        previous_status = customer.status
        customer.block(reason=command.reason)
        await self._customers.save(customer)

        await self._audit.append(AuditRecord(
            actor_id      = principal.sub,
            action        = "block-customer",
            resource_type = "Customer",
            resource_id   = str(command.customer_id),
            metadata      = {
                "previous_status": previous_status.value,
                "reason":          command.reason,
            },
        ))

        return customer

Что запрещено

АнтипаттернПравилоЧто взамен
Admin override без audit.appendAUTH-15audit обязателен при principal.is_admin
Audit через отдельный HTTP-вызов после commitAUTH-15та же AsyncSession, один session.begin()
Audit без actor_id или occurred_atAUTH-15оба обязательны
UPDATE admin_audit_log SET ...AUTH-15append-only; REVOKE UPDATE, DELETE
actor_email в plain в metadataAUTH-16только actor_id (sub из JWT)
audit.append вызывается вне транзакции (autocommit)AUTH-15session под session.begin()
Audit только для destructive-действий, не readAUTH-15критичные read (просмотр чужого PII admin-ом) тоже
Audit без trace_id / request_idAUTH-15для связки с distributed trace обязательно

Куда дальше

  • ABAC: владение ресурсом — admin override как триггер audit; AUTH-12.
  • PII и секреты — что нельзя класть в metadata; AUTH-16.
  • JWT validation — как из токена получить principal.sub для actor_id.
  • RBAC: роли — require_roles("admin") на endpoint перед Handler-ом.
  • Идемпотентность — audit идемпотентен; дубликат команды = один audit.
  • Service-to-service — system-роль от другого сервиса тоже аудируется.
  • Где делается проверка — ABAC и audit в Handler, не в роутере.
  • Хранение токенов — AUTH-20/21 как часть общей auth-цепочки.