Autenticação e autorização em APIs FastAPI costumam começar simples e rápidas, mas podem virar o ponto mais lento do sistema quando entram validação de tokens, busca de chaves públicas, consultas a políticas e enriquecimento de contexto de usuário. Esse tipo de custo aparece especialmente na cauda de latência (p95 e p99), onde poucos requests “lentos” prejudicam a percepção geral de desempenho.
Um conjunto de padrões de arquitetura permite manter segurança alta sem adicionar chamadas de rede desnecessárias no caminho crítico. A seguir estão dez desenhos práticos com JWT (JSON Web Token, um token assinado com declarações), OAuth2 (padrão para delegação de acesso) e OPA (Open Policy Agent, motor de políticas) focados em reduzir latência, controlar revogação e evitar verificações redundantes.
1) JWT sem estado com TTL curto e revogação por JTI
JWT “sem estado” significa que o servidor valida o token apenas pela assinatura e pelo tempo, sem buscar sessão em banco. Para manter controle de segurança, um TTL (tempo de vida) curto reduz a janela de abuso em caso de vazamento. O campo jti (JWT ID, identificador único do token) permite revogar tokens específicos em eventos críticos como logout ou suspeita de comprometimento. Um conjunto pequeno de revogação em Redis evita custo por request, pois só é consultado quando necessário.
Esse modelo costuma ter latência baixa porque a validação principal é local (CPU) e não envolve rede. A revogação vira uma exceção e não a regra, mantendo o “caminho quente” rápido. O TTL curto também reduz o volume de revogações que precisam ser mantidas. Em produção, é comum preferir algoritmos assimétricos, mas o exemplo usa HS256 por simplicidade.
O código abaixo mostra emissão e verificação de JWT com jti e expiração curta, além do ponto onde uma checagem de revogação poderia ser acoplada.
import time
import jwt
from fastapi import FastAPI, Depends, HTTPException, Security, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
# Em produção, preferir RS256/ES256 com chaves rotativas e JWKS.
SEGREDO = "trocar-por-segredo-forte"
ALGORITMO = "HS256"
bearer = HTTPBearer(auto_error=False)
app = FastAPI()
def criar_token_acesso(sub: str, ttl_segundos: int = 600) -> str:
agora = int(time.time())
payload = {
"sub": sub, # sujeito (usuário/serviço)
"iat": agora, # emitido em
"exp": agora + ttl_segundos,# expira em
"jti": f"{sub}:{agora}" # id único do token
}
return jwt.encode(payload, SEGREDO, algorithm=ALGORITMO)
def verificar_jwt(token_bruto: str) -> dict:
try:
return jwt.decode(token_bruto, SEGREDO, algorithms=[ALGORITMO])
except jwt.PyJWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token inválido"
)
def token_revogado(jti: str) -> bool:
# Ponto de integração: consultar Redis apenas para JTIs revogados.
# Em muitos sistemas, o conjunto é pequeno e com expiração automática.
return False
async def usuario_atual(
cred: HTTPAuthorizationCredentials = Security(bearer),
) -> dict:
if not cred:
raise HTTPException(status_code=401, detail="Token ausente")
payload = verificar_jwt(cred.credentials)
if token_revogado(payload.get("jti", "")):
raise HTTPException(status_code=401, detail="Token revogado")
return payload
@app.get("/token-demo")
def token_demo():
return {"access_token": criar_token_acesso("usuario-123")}
@app.get("/privado")
def privado(payload: dict = Depends(usuario_atual)):
return {"sub": payload["sub"], "jti": payload["jti"]}2) Cache de JWKS para RS256 com rotação de chaves sem custo por request
Em cenários multi-serviços, algoritmos assimétricos como RS256 permitem que serviços validem tokens usando apenas uma chave pública. Essa chave pública costuma ser publicada via JWKS (JSON Web Key Set), um endpoint que contém as chaves ativas. Buscar JWKS em toda requisição adiciona latência e fragilidade, então a abordagem eficiente é fazer cache por alguns minutos e atualizar de forma preguiçosa (lazy). O uso de kid (Key ID, identificador da chave) permite selecionar rapidamente a chave correta.
Uma estratégia estável mantém o conjunto antigo enquanto busca o novo, evitando falhas durante rotação. Também é comum limitar timeout e aplicar backoff simples em caso de indisponibilidade do emissor. Assim, o caminho quente vira: extrair header, localizar a chave em memória e validar assinatura localmente. A chamada de rede só ocorre de tempos em tempos, e não por request.
O exemplo abaixo ilustra cache em memória com atualização por janela de tempo e resolução da chave pelo kid.
import time
import jwt
import httpx
from fastapi import HTTPException, status
URL_JWKS = "https://emissor-exemplo/.well-known/jwks.json"
ALGORITMOS_ACEITOS = ["RS256"]
_jwks_cache = None
_jwks_cache_ts = 0.0
TTL_JWKS_SEGUNDOS = 600
async def obter_jwks() -> dict:
global _jwks_cache, _jwks_cache_ts
agora = time.time()
if _jwks_cache and (agora - _jwks_cache_ts) < TTL_JWKS_SEGUNDOS:
return _jwks_cache
try:
async with httpx.AsyncClient(timeout=2.0) as cliente:
resp = await cliente.get(URL_JWKS)
resp.raise_for_status()
jwks_novo = resp.json()
except Exception:
# Se houver cache antigo, mantém para reduzir impacto de falhas temporárias.
if _jwks_cache:
return _jwks_cache
raise HTTPException(status_code=503, detail="Falha ao obter JWKS")
_jwks_cache = jwks_novo
_jwks_cache_ts = agora
return _jwks_cache
def chave_publica_por_kid(token_jws: str, jwks: dict):
header = jwt.get_unverified_header(token_jws)
kid = header.get("kid")
if not kid:
raise HTTPException(status_code=401, detail="Token sem kid")
for k in jwks.get("keys", []):
if k.get("kid") == kid:
return jwt.algorithms.RSAAlgorithm.from_jwk(k)
raise HTTPException(status_code=401, detail="kid não encontrado no JWKS")
async def verificar_rs256(token_jws: str, audience: str | None = None) -> dict:
jwks = await obter_jwks()
chave = chave_publica_por_kid(token_jws, jwks)
try:
return jwt.decode(
token_jws,
chave,
algorithms=ALGORITMOS_ACEITOS,
audience=audience,
options={"require": ["exp", "iat", "sub"]},
)
except jwt.PyJWTError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token inválido")3) Claims enxutas e enriquecimento sob demanda com cache local
“Claims” são os campos dentro do JWT, e tokens “gordos” aumentam tamanho de header, custo de parse e exposição de dados. Um token enxuto costuma conter apenas identidade (sub), audiência (aud) e escopos (scope, permissões em forma de texto). Quando informações adicionais são necessárias, o enriquecimento ocorre sob demanda e é armazenado em cache por poucos segundos. Essa técnica reduz custo recorrente e evita carregar dados sensíveis no token.
Um cache em memória pode ser suficiente para um único processo, enquanto Redis ajuda em múltiplas réplicas. O TTL pequeno reduz risco de dados desatualizados, preservando desempenho. O desenho também separa autenticação (quem é) de autorização contextual (quais atributos influenciam decisões). No p99, a diferença aparece quando endpoints deixam de fazer chamadas repetidas ao “serviço de usuário”.
O exemplo mostra um cache simples em memória e a montagem de um contexto de autorização a partir do sub.
from functools import lru_cache
@lru_cache(maxsize=10_000)
def obter_perfil_cacheado(sub: str) -> dict:
# Em produção: em caso de cache miss, buscar em um serviço interno com timeout curto.
# Exemplo de retorno mínimo para decisões: plano e região.
return {"plano": "pro", "regiao": "br"}
def montar_contexto_autorizacao(payload_token: dict) -> dict:
contexto = {
"sub": payload_token["sub"],
"scope": payload_token.get("scope", "")
}
contexto.update(obter_perfil_cacheado(payload_token["sub"]))
return contexto4) Introspecção de token opaco na borda com cache
Tokens opacos são credenciais que não carregam claims interpretáveis localmente, exigindo introspecção (consulta ao provedor de identidade para validar e obter metadados). Se a aplicação chamar o endpoint de introspecção em toda requisição, a latência e a disponibilidade ficam acopladas ao IdP. Um padrão eficiente empurra a introspecção para a borda, como um gateway ou sidecar, e aplica cache pelo tempo restante do token. A aplicação passa a receber apenas headers com claims já verificados.
Esse desenho transforma o custo variável de rede em um custo amortizado, com mais previsibilidade. Também reduz complexidade no código FastAPI, pois a validação passa a ser “confiar em headers internos” de uma camada já autenticada. Para segurança, é importante garantir que apenas o gateway possa injetar esses headers. Assim, o caminho quente no app vira simples parse de headers e checagens locais.
O trecho abaixo ilustra como ler identidade e escopos enviados pelo gateway, mantendo validações defensivas mínimas.
from fastapi import Request, HTTPException, status
def contexto_a_partir_de_headers(request: Request) -> dict:
sub = request.headers.get("x-auth-sub")
scope = request.headers.get("x-auth-scope", "")
if not sub:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Identidade ausente")
return {"sub": sub, "scope": scope}5) Dependências por rota em vez de middleware global
Middleware roda em toda requisição, inclusive em endpoints públicos, o que adiciona custo fixo desnecessário. Em FastAPI, dependências são funções executadas apenas onde são declaradas, permitindo que rotas públicas continuem sem custo de autenticação. Esse padrão melhora o desempenho médio e também reduz ruído no p99 de endpoints que não exigem segurança. A separação por router também facilita aplicar níveis de proteção diferentes.
Além de performance, dependências tornam explícito quais endpoints exigem token, evitando decisões implícitas. Também é possível combinar dependências: autenticação básica, verificação de escopo e checagem de política. O custo de validação passa a ser pago apenas quando o endpoint realmente precisa. Esse desenho é simples e costuma trazer ganhos imediatos em sistemas com muitas rotas públicas.
O código abaixo mostra uma rota pública e outra protegida com dependency de bearer token.
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
app = FastAPI()
bearer = HTTPBearer(auto_error=False)
def validar_token_exemplo(token: str) -> dict:
# Exemplo: substituir por JWT RS256/HS256 real.
if token != "token-ok":
raise HTTPException(status_code=401, detail="Token inválido")
return {"sub": "usuario-123", "scope": "read:me"}
async def usuario_atual(cred: HTTPAuthorizationCredentials = Security(bearer)) -> dict:
if not cred:
raise HTTPException(status_code=401, detail="Token ausente")
return validar_token_exemplo(cred.credentials)
@app.get("/publico")
def publico():
return {"ok": True}
@app.get("/me")
def me(usuario: dict = Depends(usuario_atual)):
return {"sub": usuario["sub"], "scope": usuario["scope"]}6) Checagens de autorização em lote para endpoints de lista
Em endpoints que retornam listas, é comum checar permissão item a item, o que cria um padrão O(N) de chamadas para política ou banco. Esse desenho piora rapidamente o p99 quando N cresce, e fica instável sob carga. A alternativa é realizar uma checagem em lote: enviar todos os IDs para o motor de políticas e receber um mapa de decisões. O filtro final acontece localmente, com custo previsível.
Esse padrão combina bem com OPA, serviços de autorização próprios e até regras internas em memória. Também reduz overhead de rede e de serialização de payloads repetidos. Em termos de segurança, a consistência tende a melhorar, porque a decisão é centralizada e aplicada uniformemente. O resultado costuma ser uma queda visível na latência da “cauda” em endpoints de listagem.
O exemplo abaixo ilustra a ideia de uma única chamada que retorna IDs permitidos.
from typing import Iterable
class Item:
def __init__(self, item_id: str):
self.id = item_id
def pode_ver_itens_em_lote(contexto: dict, ids: list[str]) -> set[str]:
# Ponto de integração: uma única consulta ao motor de políticas.
# Exemplo: permitir apenas ids que começam com "pub-".
return {i for i in ids if i.startswith("pub-")}
def filtrar_itens_permitidos(contexto: dict, itens: Iterable[Item]) -> list[Item]:
lista = list(itens)
ids = [i.id for i in lista]
permitidos = pode_ver_itens_em_lote(contexto, ids)
return [i for i in lista if i.id in permitidos]7) OPA em WASM no processo para remover salto de rede
OPA via HTTP adiciona um salto de rede por requisição, frequentemente entre 3 e 10 ms, além de variância em momentos de pico. Uma alternativa é compilar políticas Rego (linguagem de políticas do OPA) para WASM (WebAssembly, formato executável portátil) e avaliar no mesmo processo da aplicação. Com isso, a decisão de política vira CPU local e elimina dependência de rede no caminho crítico. O input pode ser mantido pequeno, contendo apenas identidade, escopos e dados do recurso.
Esse desenho é mais adequado quando as políticas são relativamente estáveis e quando há disciplina para manter dados de políticas enxutos. Também reduz risco de timeouts entre app e OPA, melhorando previsibilidade do p99. A compilação e carregamento acontecem no startup, e cada request apenas chama a função de avaliação. Mesmo sem mostrar integração completa de runtime WASM, a estrutura do input e do resultado já orienta uma implementação consistente.
O trecho abaixo apresenta um exemplo de política Rego simples e o formato de entrada que costuma ser enviado para avaliação.
Exemplo conceitual de Rego (política de permitir leitura quando existe o escopo adequado):
allow { input.scope[_] == "read:items" }
O exemplo a seguir mostra como padronizar o dicionário de entrada para a política, mantendo o payload mínimo.
def montar_input_politica(contexto: dict, recurso: dict) -> dict:
# Input mínimo: identidade, escopos e atributos essenciais do recurso.
scope_str = contexto.get("scope", "")
escopos = [s.strip() for s in scope_str.split() if s.strip()]
return {
"sub": contexto["sub"],
"scope": escopos,
"resource": {
"tipo": recurso.get("tipo"),
"id": recurso.get("id"),
"dono": recurso.get("dono"),
},
}8) Token Exchange para comunicação serviço-a-serviço com audiência correta
Em arquiteturas com múltiplos serviços, reutilizar token de usuário final em chamadas internas aumenta superfície de risco e custo de validação. Token Exchange (troca de token) é um padrão OAuth2 em que um token é trocado por outro, com audiência (aud) específica para o serviço alvo. Isso melhora o princípio do menor privilégio e reduz tamanho de tokens propagados. Para performance, a troca deve ocorrer uma vez por janela e o token resultante deve ser reutilizado até perto de expirar.
Esse desenho reduz a frequência de chamadas ao servidor OAuth2 e melhora previsibilidade. Também evita validações mais complexas em serviços internos, porque o token já nasce com claims mais adequadas ao hop. A cache pode ser em memória por processo, com margem de segurança para renovar alguns segundos antes do exp. Com isso, o caminho quente de cada chamada interna não inclui troca, apenas envio do token de serviço válido.
O exemplo abaixo ilustra cache simples de token de serviço com renovação por tempo.
import time
import httpx
_cache_token_servico = None
_cache_exp = 0
async def trocar_token_por_servico(token_entrada: str) -> tuple[str, int]:
# Exemplo: chamada ao servidor OAuth2 para token exchange.
# Em produção, incluir client authentication e parâmetros RFC 8693.
async with httpx.AsyncClient(timeout=2.0) as cliente:
resp = await cliente.post(
"https://auth-exemplo/oauth/token",
data={"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange"},
headers={"authorization": f"Bearer {token_entrada}"},
)
resp.raise_for_status()
dados = resp.json()
return dados["access_token"], int(time.time()) + int(dados.get("expires_in", 300))
async def obter_token_servico(token_entrada: str) -> str:
global _cache_token_servico, _cache_exp
agora = int(time.time())
margem = 90 # renova antes de expirar para reduzir falhas por corrida
if _cache_token_servico and (agora + margem) < _cache_exp:
return _cache_token_servico
token, exp = await trocar_token_por_servico(token_entrada)
_cache_token_servico = token
_cache_exp = exp
return _cache_token_servico9) Proteção de rotas de alto custo com mTLS e escopos
mTLS (mutual TLS) é TLS com autenticação bilateral, onde o cliente também apresenta certificado. Esse modelo fortalece a identidade de serviços internos, reduzindo o risco de um token válido ser usado por um chamador não autorizado. Para desempenho, a verificação de certificado pode ser feita no gateway, e o custo do handshake TLS é amortizado por conexões keep-alive. Mesmo com mTLS, escopos continuam importantes para granularidade e menor privilégio.
Esse desenho é útil para rotas sensíveis, como administrativas e internas, que não deveriam depender apenas de bearer tokens. O gateway pode inserir headers internos após validar o certificado do cliente, permitindo que a aplicação tome decisões sem criptografia pesada em Python. A combinação reduz risco e mantém latência consistente. Assim, a aplicação continua focada em lógica e políticas, e não em detalhes de TLS por request.
O exemplo abaixo mostra como validar presença de um “sinal” confiável vindo do gateway, representando um cliente mTLS já verificado.
from fastapi import Request, HTTPException, status
def exigir_chamada_interna_mtls(request: Request) -> None:
# Exemplo: header inserido pelo gateway após validar mTLS e identidade do cliente.
cliente = request.headers.get("x-mtls-client")
if not cliente:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Rota restrita a mTLS")
def exigir_escopo(contexto: dict, escopo_necessario: str) -> None:
scope_str = contexto.get("scope", "")
escopos = set(s for s in scope_str.split() if s)
if escopo_necessario not in escopos:
raise HTTPException(status_code=403, detail="Escopo insuficiente")10) Medição do custo de autenticação e autorização como parte do produto
Sem medição, custos de autenticação viram suposições, e suposições raramente capturam o que afeta p99. Instrumentar tempos de decodificação de JWT, busca de JWKS, consultas de cache e avaliação de políticas permite localizar gargalos reais. Métricas como p50, p95 e p99 mostram tanto o desempenho típico quanto a cauda. Alertas de regressão de poucos milissegundos ajudam a impedir que uma mudança pequena vire um problema grande em produção.
Uma abordagem simples mede trechos críticos com um temporizador e registra eventos estruturados. Em sistemas maiores, esses dados normalmente alimentam tracing distribuído, mas a base é sempre a mesma: medir o que importa no caminho quente. Com isso, decisões como “mover OPA para WASM” ou “cachear JWKS por 10 minutos” deixam de ser palpite. O resultado é previsibilidade e evolução segura do sistema.
O exemplo abaixo cria um medidor de tempo com context manager e envolve as fases principais.
import time
import contextlib
@contextlib.contextmanager
def medir_tempo(nome_metrica: str):
inicio = time.perf_counter_ns()
try:
yield
finally:
duracao_ms = (time.perf_counter_ns() - inicio) / 1e6
print({"metrica": nome_metrica, "ms": round(duracao_ms, 3)})
def processar_autenticacao(token_bruto: str) -> dict:
with medir_tempo("jwt.decode"):
payload = {"sub": "exemplo"} # Substituir por verificação real
with medir_tempo("authz.contexto"):
contexto = {"sub": payload["sub"], "scope": "read:items"}
return contextoPanorama de escolha entre padrões e encerramento
Os desenhos mais rápidos mantêm o caminho quente local, evitando chamadas de rede por request e reduzindo trabalho repetido. JWT com TTL curto e revogação pontual minimiza dependências, enquanto cache de JWKS torna RS256 viável sem custo constante. Tokens enxutos, enriquecimento com cache e checagens em lote reduzem peso de CPU e evitam explosão de chamadas em endpoints de lista. Quando políticas são pesadas, OPA em WASM elimina latência de rede e torna a decisão mais previsível.
Em ambientes corporativos, introspecção na borda permite usar tokens opacos sem acoplar a aplicação ao IdP, e token exchange melhora a segurança e o desempenho em comunicação serviço-a-serviço. Para rotas internas críticas, mTLS no gateway adiciona identidade forte sem impor criptografia por requisição no app. A medição contínua fecha o ciclo ao transformar “sensação de lentidão” em números acionáveis. Com esses elementos combinados, autenticação e autorização deixam de ser o componente mais lento e passam a ser uma parte estável do desempenho do sistema.