API логирование и JSON: PII маскирование для...

Проблема: API логи содержат полные request/response

Типичный API лог

{
  "timestamp": "2025-03-07T10:15:30Z",
  "method": "POST",
  "endpoint": "/api/users",
  "request": {
    "name": "John Smith",
    "email": "john@example.com",
    "phone": "+1-555-0123",
    "ssn": "123-45-6789",
    "address": "123 Main St, City, State 12345"
  },
  "response_status": 201,
  "response": {
    "user_id": "user_12345",
    "email": "john@example.com"
  },
  "duration_ms": 145
}

Полная PII утечка: Все персональные данные залогированы.

Архитектура: Selective Logging

Подход 1: Whitelist (разрешить только нужное)

class SelectiveAPILogger:
    # Определяем какие поля безопасны для логирования
    LOGGABLE_FIELDS = {
        'user_id': True,        # Безопасно
        'email': False,         # PII
        'phone': False,         # PII
        'ssn': False,           # Критичная PII
        'address': False,       # PII
        'action': True,         # Безопасно
        'status': True          # Безопасно
    }
    
    def log_request(self, request_data):
        """
        Логирует только безопасные поля
        """
        safe_data = {}
        
        for key, value in request_data.items():
            if self.LOGGABLE_FIELDS.get(key, False):
                safe_data[key] = value
            else:
                safe_data[key] = '[REDACTED]'
        
        logger.info(safe_data)

Подход 2: Intelligent Masking (маскирование с уверенностью)

from presidio_analyzer import AnalyzerEngine

class IntelligentAPILogger:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
    
    def log_request(self, request_data):
        """
        Анализирует данные и маскирует PII с высокой уверенностью
        """
        masked_data = {}
        
        for key, value in request_data.items():
            if not isinstance(value, str):
                masked_data[key] = value
                continue
            
            # Анализируем значение
            entities = self.analyzer.analyze(
                text=value,
                language='en'
            )
            
            # Маскируем только высокоуверенную PII (>95%)
            high_confidence_entities = [
                e for e in entities if e.score > 0.95
            ]
            
            if high_confidence_entities:
                masked_data[key] = f'[MASKED_{high_confidence_entities[0].entity_type}]'
            else:
                masked_data[key] = value
        
        logger.info(masked_data)

Реальная реализация в FastAPI

from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
import json
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

class APILoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Читаем request body
        request_body = await request.body()
        
        try:
            request_json = json.loads(request_body)
        except:
            request_json = str(request_body)
        
        # Маскируем перед логированием
        masked_request = self.mask_pii(request_json)
        
        logger.info({
            'method': request.method,
            'path': request.url.path,
            'request': masked_request,
            'timestamp': datetime.utcnow().isoformat()
        })
        
        # Вызываем обработчик
        response = await call_next(request)
        
        # Логируем response
        logger.info({
            'status': response.status_code,
            'path': request.url.path
        })
        
        return response
    
    def mask_pii(self, data):
        """
        Рекурсивно маскирует PII в структуре данных
        """
        if isinstance(data, dict):
            return {
                k: self.mask_value(k, v)
                for k, v in data.items()
            }
        elif isinstance(data, list):
            return [self.mask_pii(item) for item in data]
        return data
    
    def mask_value(self, key, value):
        pii_keywords = [
            'email', 'phone', 'ssn', 'password',
            'token', 'credit_card', 'address'
        ]
        
        if any(kw in key.lower() for kw in pii_keywords):
            return '[REDACTED]'
        return value

app.add_middleware(APILoggingMiddleware)

GDPR требования для API логирования

Статья 32: Security

• ✅ Логирование только необходимых данных
• ✅ Маскирование PII в логах
• ✅ Шифрование логов at rest
• ✅ Ограничение доступа к логам

Статья 5: Minimization

• ✅ Логировать только для целей отладки и мониторинга
• ✅ Автоматическое удаление логов через 30-90 дней
• ❌ Не логировать полные request/response с PII

Лучшие практики

✅ Используйте whitelist approach (разрешить только безопасное) ✅ Маскируйте PII в логах автоматически ✅ Установите retention policy для автоудаления ✅ Обучите команду правилам логирования ✅ Регулярно аудируйте логи на утечки PII

Вывод

API логирование необходимо для отладки, но маскирование PII необходимо для GDPR соответствия.