Segurança e Hardening

# .env
MASTER_KEY=sua_chave_super_secreta_aqui_min_32_chars

# Segurança JWT
JWT_SECRET_KEY=chave_secreta_diferente_da_secret_key_min_32_chars
JWT_ALGORITHM=HS256
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=60
JWT_REFRESH_TOKEN_EXPIRE_DAYS=7

# Para produção, considere RS256
JWT_ALGORITHM=RS256
JWT_PRIVATE_KEY_PATH=/path/to/private.pem
JWT_PUBLIC_KEY_PATH=/path/to/public.pem

{
  "sub": "user_id",
  "type": "access|refresh",
  "exp": 1640995200,
  "iat": 1640991600,
  "jti": "unique-token-id",
  "permissions": ["agravos.listar", "agravos.criar"]
}

# core/security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
import secrets

class TokenService:
    def create_access_token(self, user_id: int) -> str:
        expire = datetime.now(timezone.utc) + timedelta(
            minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
        )
        
        to_encode = {
            "sub": str(user_id),
            "type": "access",
            "exp": expire,
            "iat": datetime.now(timezone.utc),
            "jti": secrets.token_urlsafe(16)
        }
        
        return jwt.encode(
            to_encode, 
            settings.JWT_SECRET_KEY, 
            algorithm=settings.JWT_ALGORITHM
        )

from fastapi import Depends
from core.permissions import requer_permissao, PermissoesBase

# Endpoint protegido por permissão
@router.post("/agravos/casos")
async def criar_caso(
    caso: CasoSchema,
    current_user = Depends(requer_permissao(PermissoesBase.AGRAVOS_CRIAR))
):
    # Usuário tem permissão garantida
    return await criar_novo_caso(caso, current_user)

# Verificação programática
from core.permissions import verificar_permissao

async def processar_dados(usuario: UsuarioModel, dados: dict):
    if not verificar_permissao(usuario, PermissoesBase.ADMIN):
        raise HTTPException(403, "Operação requer permissão de admin")
    
    # Processar dados sensíveis

from passlib.context import CryptContext
import re

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def validar_senha(senha: str) -> List[str]:
    erros = []
    
    if len(senha) < 8:
        erros.append("Senha deve ter no mínimo 8 caracteres")
    if not re.search(r"[A-Z]", senha):
        erros.append("Senha deve conter pelo menos 1 letra maiúscula")
    if not re.search(r"[a-z]", senha):
        erros.append("Senha deve conter pelo menos 1 letra minúscula")
    if not re.search(r"\d", senha):
        erros.append("Senha deve conter pelo menos 1 número")
    if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", senha):
        erros.append("Senha deve conter pelo menos 1 caractere especial")
    
    return erros

def hash_senha(senha: str) -> str:
    return pwd_context.hash(senha)

def verificar_senha(senha_plana: str, senha_hash: str) -> bool:
    return pwd_context.verify(senha_plana, senha_hash)

# Rate Limiting
SECURITY_MAX_LOGIN_ATTEMPTS=5
SECURITY_LOCKOUT_DURATION=900  # 15 minutos
SECURITY_PROGRESSIVE_DELAY=true
SECURITY_MAX_IP_ATTEMPTS_HOUR=20
SECURITY_IP_BLOCK_DURATION=3600  # 1 hora

# API Rate Limits
RATE_LIMIT_PER_MINUTE=60
RATE_LIMIT_PER_HOUR=1000
RATE_LIMIT_BURST=10

from core.cache import redis_client
import time

class RateLimiter:
    async def check_rate_limit(
        self, 
        key: str, 
        limit: int, 
        window: int
    ) -> bool:
        current = int(time.time())
        window_start = current - window
        
        # Remove entradas antigas
        await redis_client.zremrangebyscore(key, 0, window_start)
        
        # Conta requisições na janela
        count = await redis_client.zcard(key)
        
        if count >= limit:
            return False
        
        # Adiciona nova requisição
        await redis_client.zadd(key, {str(current): current})
        await redis_client.expire(key, window)
        
        return True

# Middleware de rate limiting
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}"
    
    limiter = RateLimiter()
    if not await limiter.check_rate_limit(key, 60, 60):
        return JSONResponse(
            status_code=429,
            content={"error": "Too Many Requests"},
            headers={"Retry-After": "60"}
        )
    
    return await call_next(request)

import structlog
from datetime import datetime

logger = structlog.get_logger()

async def log_security_event(
    event_type: str,
    user_id: Optional[int],
    details: dict,
    request: Request
):
    await logger.ainfo(
        "security_event",
        event_type=event_type,
        timestamp=datetime.utcnow().isoformat(),
        user_id=user_id,
        ip_address=request.client.host,
        user_agent=request.headers.get("user-agent"),
        method=request.method,
        path=request.url.path,
        details=details
    )

# Exemplo de uso
await log_security_event(
    event_type="permission_denied",
    user_id=current_user.id,
    details={
        "required_permission": "agravos.deletar",
        "user_permissions": current_user.permissoes
    },
    request=request
)

# Política de retenção
log_retention:
  security_events: 365d  # 1 ano
  access_logs: 90d       # 3 meses
  error_logs: 180d       # 6 meses
  debug_logs: 7d         # 1 semana

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

# CORS configurado restritivamente
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,  # Lista específica
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    max_age=3600,
)

# Trusted hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=settings.ALLOWED_HOSTS
)

# Security headers
@app.middleware("http")
async def security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

from pydantic import BaseModel, validator, EmailStr
from typing import Optional
import re

class UsuarioCreate(BaseModel):
    email: EmailStr
    senha: str
    nome: str
    cpf: Optional[str] = None
    
    @validator('senha')
    def senha_forte(cls, v):
        erros = validar_senha(v)
        if erros:
            raise ValueError('; '.join(erros))
        return v
    
    @validator('cpf')
    def cpf_valido(cls, v):
        if v and not validar_cpf(v):
            raise ValueError('CPF inválido')
        return v
    
    @validator('nome')
    def nome_seguro(cls, v):
        # Previne injection
        if re.search(r'[<>\"\'%;()&+]', v):
            raise ValueError('Nome contém caracteres inválidos')
        return v

from cryptography.fernet import Fernet
import base64

class DataEncryption:
    def __init__(self, key: str):
        self.cipher = Fernet(base64.urlsafe_b64encode(key.encode()[:32]))
    
    def encrypt_pii(self, data: str) -> str:
        """Criptografa dados pessoais identificáveis"""
        return self.cipher.encrypt(data.encode()).decode()
    
    def decrypt_pii(self, encrypted: str) -> str:
        """Descriptografa dados pessoais"""
        return self.cipher.decrypt(encrypted.encode()).decode()

# Uso
encryption = DataEncryption(settings.ENCRYPTION_KEY)
encrypted_cpf = encryption.encrypt_pii("123.456.789-00")