PII и секреты — что нельзя в логи, detail и Kafka (Python)

PII (email, phone, ФИО) не в логах, не в HTTPException.detail, не в Kafka-событиях — только id. Секреты через pydantic-settings из env или Vault, не в git. AUTH-16..18, FastAPI.

Опубликовано 20 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: AUTH-16 … AUTH-18 из Auth Patterns → раздел 7. PII и секреты.

Важно знать

PII не в логах (даже DEBUG), не в str(exc), не в HTTPException.detail, не в Kafka-событиях широкого scope.

Передавать только id — payload подгружается потребителем через специальный сервис по запросу.

Секреты не в git — только через pydantic-settings из env или Vault / SealedSecrets.

Exception-handler не выводит str(cause) в detail — только заранее заданное сообщение по коду ошибки.

__str__ / __repr__ агрегата с PII-полями — потенциальная утечка при любом logger.info("%s", customer).

PII leak в одном слое — пробивает безопасность всей системы.

Логи структурированы через structlog или logging — в dict-поля попадает только customer_id, не email.

PII (Personally Identifiable Information) — email, телефон, ФИО, адрес, паспорт, IP, биометрия. Compliance и регуляторы считают их защищаемыми: утечка через лог, response или Kafka — инцидент, штраф, потеря доверия. Правило UCP: «PII никогда не покидает domain layer».

PII не в логах

AUTH-16: запрет на все уровни — DEBUG, INFO, WARNING, ERROR.

import logging

logger = logging.getLogger(__name__)

# КАТАСТРОФА
logger.info("User registered: email=%s phone=%s", customer.email, customer.phone)

# ХОРОШО — только internal id
logger.info("User registered: customer_id=%s", customer.id)

# ЕСЛИ нужен PII для диагностики — masked
logger.info("Email verification sent: customer_id=%s email_mask=%s",
            customer.id, mask_email(customer.email))   # u***@example.com

Функции маскирования:

# domain/pii_masking.py

def mask_email(email: str) -> str:
    if not email or "@" not in email:
        return "***"
    local, domain = email.split("@", 1)
    return local[0] + "***@" + domain


def mask_phone(phone: str) -> str:
    if not phone or len(phone) < 4:
        return "***"
    return "***" + phone[-4:]

Дополнительно — __repr__ и __str__ агрегата не возвращают PII:

# domain/customer/customer.py
from dataclasses import dataclass

@dataclass
class Customer:
    id: str
    email: str
    phone: str
    full_name: str

    def __repr__(self) -> str:
        return f"Customer(id={self.id!r})"

    def __str__(self) -> str:
        return f"Customer[id={self.id}]"

При использовании structlog — bound-поля в контексте запроса содержат только customer_id:

import structlog

log = structlog.get_logger()

# ХОРОШО
log.info("order_created", order_id=order.id, customer_id=order.customer_id)

# ПЛОХО
log.info("order_created", order_id=order.id, customer_email=customer.email)

PII не в `str(exc)`

AUTH-16 + AUTH-18: текст исключения может попасть и в логи, и в response.

# КАТАСТРОФА
raise ValueError(f"Email {email} is invalid format")

Что происходит:

logger.exception(...) пишет str(exc) в лог → PII leak.
Exception-handler делает detail=str(exc) → клиент (включая атакующего, перебирающего email) видит подтверждение значения.

Корректно — доменные исключения с кодом, без PII-значения:

# domain/customer/errors.py

class CustomerDomainError(Exception):
    def __init__(self, code: str, message: str) -> None:
        self.code = code
        self.message = message
        super().__init__(message)


class InvalidEmailFormatError(CustomerDomainError):
    def __init__(self) -> None:
        super().__init__(
            code="INVALID_EMAIL_FORMAT",
            message="Provided email is in invalid format",
        )

В message и detail — общее сообщение без значения. Если нужно для диагностики — отдельный лог с mask_email(email), не через исключение.

Exception-handler без `str(cause)`

AUTH-18: маппинг явный, не «прокинуть исключение дальше».

# adapters/in/http/exception_handlers.py
from fastapi import Request
from fastapi.responses import JSONResponse
from domain.order.errors import OrderDomainError

async def order_domain_exception_handler(
    request: Request,
    exc: OrderDomainError,
) -> JSONResponse:
    detail = {
        "ORDER_NOT_FOUND": "Order with given id not found",
        "ORDER_NOT_CANCELLABLE": "Order in current status cannot be cancelled",
    }.get(exc.code, "Order operation failed")

    return JSONResponse(
        status_code=400,
        content={
            "type": "urn:order:domain",
            "title": "Order operation failed",
            "detail": detail,
            "errorCode": exc.code,
        },
    )


async def generic_exception_handler(
    request: Request,
    exc: Exception,
) -> JSONResponse:
    request_id = request.headers.get("X-Request-Id", "unknown")
    return JSONResponse(
        status_code=500,
        content={
            "title": "Internal server error",
            "detail": f"An unexpected error occurred. Reference: {request_id}",
        },
    )

Регистрация в app:

# adapters/in/http/app.py
from fastapi import FastAPI
from domain.order.errors import OrderDomainError
from adapters.in.http.exception_handlers import (
    order_domain_exception_handler,
    generic_exception_handler,
)

app = FastAPI()
app.add_exception_handler(OrderDomainError, order_domain_exception_handler)
app.add_exception_handler(Exception, generic_exception_handler)

Что не делаем:

# ПЛОХО
return JSONResponse(
    status_code=400,
    content={"detail": str(exc)},             # может содержать PII
)

return JSONResponse(
    status_code=500,
    content={"detail": str(exc.__cause__)},   # тем более внутренние детали
)

return JSONResponse(
    status_code=500,
    content={"traceback": traceback.format_exc()},  # утечка структуры
)

PII не в Kafka широкого scope

AUTH-16: Kafka — широковещательный канал, все consumer'ы видят payload.

# ПЛОХО — все consumers видят PII
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_email: str   # ← leak
    customer_phone: str   # ← leak
    total_amount: Decimal


# ХОРОШО — id, PII подгружается через customer-service по необходимости
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_id: str
    total_amount: Decimal

Notification-сервис, которому нужен email, делает GET /customers/{id}/email к customer-service. Это даёт точечный access + audit log на стороне customer-service.

# adapters/out/http/customer_client.py
import httpx
from settings import Settings

class CustomerClient:
    def __init__(self, settings: Settings, http: httpx.AsyncClient) -> None:
        self._base = settings.customer_service_url
        self._http = http

    async def get_email(self, customer_id: str) -> str:
        response = await self._http.get(
            f"{self._base}/customers/{customer_id}/email",
        )
        response.raise_for_status()
        return response.json()["email"]

Секреты не в git

AUTH-17: правило для всех окружений, включая staging.

# КАТАСТРОФА — в коде или .env, закоммиченном в git
DB_PASSWORD = "super-secret-password-prod"

# ХОРОШО — pydantic-settings читает из env, Vault или SealedSecrets
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    db_password: str
    client_secret: str
    jwks_uri: str
    audience: str
    issuer: str

Откуда брать значения переменных окружения:

Vault (HashiCorp) — vault-secrets-operator или hvac Python-клиент в startup.
SealedSecrets (Kubernetes) — шифрованный secret в git, расшифровывается оператором в кластере.
Cloud Secret Manager (AWS SM, GCP SM) — IAM-роль pod'а получает secret напрямую.
Kubernetes Secret — минимальный baseline.

.gitignore со списком чувствительных файлов:

.env
.env.local
.env.prod
*.pem
*.key
secrets.yaml

Pre-commit hook через detect-secrets или trufflehog — отдельная защита от случайного коммита:

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/Yelp/detect-secrets
    rev: v1.4.0
    hooks:
      - id: detect-secrets
        args: ["--baseline", ".secrets.baseline"]

При обнаружении секрета в git history — ротировать секрет немедленно (даже если коммит удалён — он может быть в forks, CI cache, attacker уже клонировал).

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

Антипаттерн	Правило	Что взамен
PII в логах (`email`, `phone`)	`AUTH-16`	`mask_email` или только `id`
PII в тексте исключения	`AUTH-16` + `AUTH-18`	код ошибки, общее сообщение
`detail=str(exc)` в exception-handler	`AUTH-18`	маппинг по коду ошибки
`detail=str(exc.__cause__)`	`AUTH-18`	заранее заданное сообщение
PII в Kafka-событиях широкого scope	`AUTH-16`	только `id` + lazy fetch
Секрет как константа в `.py` или в закоммиченном `.env`	`AUTH-17`	`pydantic-settings` из env
Секреты в git history	`AUTH-17`	rotate immediately
`traceback` в response	`AUTH-18`	только `request_id` для cross-ref
`logger.info("%s", customer)` с PII в `__str__`	`AUTH-16`	`__str__` без PII-полей

Куда дальше

Auth → раздел 7. PII и секреты — нормативные формулировки.
ABAC: владение ресурсом — ownership-проверка с principal.sub.
Аудит admin-команд — audit log без PII в plain.
JWT validation — Principal, require_roles через JWKS.
RBAC: маппинг ролей — маппинг ролей из claim без PII в authorities.
Идемпотентность — Idempotency-Key для money-команд.
Хранение токенов на клиенте — HttpOnly cookie, RT rotation.
Service-to-service — mTLS и Client Credentials для inter-service.
Где какая проверка — PII-логика остаётся в Domain Service.