← назад к разделу

Представьте: разработчик добавляет logger.info("User registered: email=%s", user.email) — всего одна строчка. Через полгода лог-агрегатор попадает в утечку, и email-адреса пользователей оказываются в открытом доступе. Так выглядит типичная PII-утечка — одна невинная строка, которая потом стоит компании штрафа и потери доверия.

Разберём, что такое PII, почему эти данные нельзя появляться в логах, ответах и очередях сообщений, и как правильно хранить секреты приложения.

Что такое PII и почему это важно

PII (Personally Identifiable Information) — данные, по которым можно идентифицировать конкретного человека: email, телефон, ФИО, адрес, паспортные данные, IP-адрес, биометрия.

Регуляторы (GDPR, 152-ФЗ и другие) относят PII к защищаемым данным. Утечка через лог, HTTP-ответ или очередь сообщений — это инцидент безопасности со всеми последствиями: аудит, штраф, уведомление пользователей.

Главное правило: PII не должен выходить за пределы того слоя, которому он нужен для работы. Логи, HTTP-ответы с ошибками и события в очереди — не то место.

PII в логах

Вот распространённая ошибка, которую делают в первые дни работы с новым сервисом:

import logging

logger = logging.getLogger(__name__)

# Так делать нельзя
logger.info("User registered: email=%s phone=%s", customer.email, customer.phone)

# Правильно — только внутренний идентификатор
logger.info("User registered: customer_id=%s", customer.id)

Логи идут в агрегаторы (ELK, Datadog, Loki), часто доступны широкому кругу людей в команде, иногда экспортируются в сторонние системы аналитики. Персональные данные там накапливаются незаметно.

Если для диагностики действительно нужно что-то от пользователя, используют маскирование — показывают только часть значения:

# 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:]

# Приемлемо для диагностики
logger.info("Email verification sent: customer_id=%s email_mask=%s",
            customer.id, mask_email(customer.email))   # u***@example.com

__str__ и __repr__ объектов с PII

Ещё одна скрытая ловушка: если класс с персональными данными не переопределяет __str__ и __repr__, то logger.info("%s", customer) автоматически выведет все поля.

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)

Если используете structlog, следите за тем, что попадает в контекст запроса:

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 в тексте исключений

Ещё одна частая ошибка — включать значение PII прямо в текст исключения:

# Так делать нельзя
raise ValueError(f"Email {email} is invalid format")

Что происходит потом с этим текстом:

  • logger.exception(...) пишет str(exc) в лог — и email оказывается там.
  • Exception-handler возвращает detail=str(exc) клиенту — пользователь (или злоумышленник) видит подтверждение адреса.

Правильный подход — доменные исключения с кодом ошибки, без 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",
        )

Сообщение "Provided email is in invalid format" говорит о проблеме, но не называет сам адрес. Если нужна диагностика — отдельный лог с mask_email(email), не через исключение.

Exception-handler в FastAPI

Exception-handler — место, где особенно легко случайно «проскочить» PII в ответ клиенту.

Типичная ошибка:

# Так делать нельзя
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()},  # структура кода наружу
)

Правильно — явный маппинг кода ошибки на безопасное сообщение:

# 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}",
        },
    )

Регистрация в приложении:

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)

В случае неожиданной ошибки клиент получает request_id — по нему можно найти детали в логах внутри системы, не раскрывая их наружу.

PII в событиях Kafka

Kafka — широковещательный канал. Все подписчики, в том числе те, которые появятся через год, увидят каждое сообщение. Если туда попало PII — оно хранится в топике до истечения retention и доступно любому consumer в системе.

# Плохо — все consumer'ы видят персональные данные
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_email: str   # утечка
    customer_phone: str   # утечка
    total_amount: Decimal


# Правильно — только идентификатор
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_id: str
    total_amount: Decimal

Если notification-сервису нужен email для отправки письма, он запрашивает его напрямую у customer-сервиса:

# 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"]

Это даёт точечный доступ с аудит-логом на стороне customer-сервиса — видно, кто и когда запросил email.

Секреты не в коде и не в git

Пароли к базе, API-ключи, клиентские секреты — их нельзя хранить в коде или в .env-файлах, зафиксированных в git.

# Так делать нельзя — секрет прямо в коде
DB_PASSWORD = "super-secret-password-prod"

Правильный подход — pydantic-settings читает значения из переменных окружения:

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

Сам .env-файл с реальными значениями — только локально, в .gitignore:

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

Откуда брать значения в production:

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

Защита от случайного коммита

Даже при аккуратной работе случайный коммит секрета случается. detect-secrets находит похожие на секреты строки до того, как они попадут в историю:

# .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-истории — менять его нужно немедленно. Удаление коммита не помогает: он может быть в форках, CI-кэше, у кого-то уже скачан.

Коротко

  • PII — email, телефон, ФИО, адрес, IP — нельзя писать в логи ни на каком уровне (включая DEBUG).
  • Для диагностики используй маскирование (mask_email, mask_phone), не сам адрес.
  • Переопредели __str__ и __repr__ у классов с персональными данными — иначе случайное логирование объекта раскроет всё.
  • Исключения не должны содержать PII-значений — только код ошибки и общее сообщение.
  • Exception-handler в FastAPI возвращает заранее заданный текст по коду ошибки, не str(exc) или трассировку стека.
  • В Kafka-события кладут только идентификатор; сервис, которому нужны детали, запрашивает их напрямую.
  • Секреты в коде и в git — недопустимо; pydantic-settings + переменные окружения + Vault/SealedSecrets/Cloud SM.
  • Если секрет попал в git-историю — смени его немедленно, даже если коммит удалён.

Что почитать дальше