Autenticação e JWT — Um Sistema de Identidade de API Seguro
Um JWT é como uma chave de quarto de hotel—o Access Token é um passe diário (válido por um curto período), e o Refresh Token é um passe mensal (válido por um longo tempo mas pode ser revogado a qualquer momento); a recepção (serviço de autenticação) é responsável por emitir e verificar as chaves.
1. O Que Você Vai Aprender
- Princípios e Estrutura do JWT: Explicação de Header, Payload e Signature
python-joseEmissão e Verificação de Access Tokens e Refresh Tokens (Token Duplo)- Integração
OAuth2PasswordBearercom Ferramentas de Segurança do FastAPI - Hashing de Senha: Melhores Práticas
passlib+bcrypt - Cenário da Alice: Autenticação Multi-Tenant SaaS do PriceTracker—Matriz de Permissões para Diferentes Planos de Assinatura
2. A História Real da Alice
(1) Problema: APIs não autenticadas sendo crawleadas maliciosamente
A API do PriceTracker da Alice está completamente aberta, e qualquer pessoa pode chamar todos os seus endpoints. Um concorrente escreveu um script que faz scraping de um milhão de entradas de dados de preço por minuto. Para piorar, alguém usou a API para submeter dados de preço falsos, corrompendo todo o banco de dados. Alice precisa implementar registro e login de usuário, autenticação de API, e controle de acesso baseado em diferentes níveis de assinatura, mas não sabe como fazer isso de forma segura.
(2) Soluções com Autenticação JWT
JWT (JSON Web Token) é uma solução de autenticação stateless: Usuários obtêm um token ao fazer login, incluem o token em requisições subsequentes, e o servidor verifica a assinatura—não há necessidade de armazenar sessões, e suporta inerentemente sistemas distribuídos.
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
@app.get("/me")
async def get_me(token: str = Depends(oauth2_scheme)):
user = verify_token(token)
return user
(3) Resultado
Uma vez que a autenticação de API é habilitada, usuários não registrados não poderão acessar endpoints protegidos, e scraping malicioso será bloqueado tanto pelo middleware de limitação de taxa quanto pela autenticação. Usuários Pro podem importar dados em lote, enquanto usuários Free são limitados a 1.000 registros; concorrentes só podem ver dados públicos.
3. Princípios e Estrutura do JWT
(1) A Estrutura de Três Partes de um JWT
Um JWT consiste em três partes: Header (algoritmo), Payload (dados), e Signature, separadas por ..
sequenceDiagram
participant Client as Bob Frontend
participant Auth as /auth/login
participant API as API Protegida
participant Verify as Verificador de Token
Client->>Auth: POST email + password
Auth->>Auth: Verificar credenciais
Auth-->>Client: Access Token + Refresh Token
Client->>API: GET /products (Bearer Token)
API->>Verify: Decodificar e verificar assinatura
Verify-->>API: Payload do usuário
API-->>Client: 200 OK + dados
Note over Client,Auth: Access Token expira após 30 min
Client->>Auth: POST /auth/refresh (Refresh Token)
Auth-->>Client: Novo Access Token
| Seção | Conteúdo | Exemplo |
|---|---|---|
| Header | Tipo de Algoritmo | {"alg": "HS256", "typ": "JWT"} |
| Payload | Dados do Usuário (Claims) | {"sub": "alice", "role": "admin", "exp": 1700000000} |
| Signature | Assinatura | HMACSHA256(header.payload, secret) |
(1) ▶ Exemplo: Codificação e Decodificação JWT
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=30))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> dict:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
raise ValueError("Invalid token")
# Teste
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")
payload = verify_token(token)
print(f"Payload: {payload}")
Saída:
# Função definida com sucesso
(2) Access Token vs. Refresh Token
| Dimensão | Access Token | Refresh Token |
|---|---|---|
| Período de Validade | 30 minutos | 7 dias |
| Propósito | Acessar Recursos da API | Renovar Access Token |
| Armazenamento | Memória (Frontend) | HttpOnly Cookie |
| Risco de Exposição | Alto (incluído em cada requisição) | Baixo (usado apenas durante renovações) |
| Método de Revogação | Aguardar Expiração | Blacklist no Servidor |
4. Integração OAuth2PasswordBearer
(1) ▶ Exemplo: Configuração Completa de Autenticação
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
app = FastAPI()
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
# Em produção: consultar usuário do banco de dados
user = {"username": username, "role": payload.get("role", "user")}
return user
@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# Em produção: verificar senha do banco de dados
if form_data.username != "alice" or form_data.password != "secret":
raise HTTPException(status_code=401, detail="Incorrect credentials")
access_token = create_access_token(
data={"sub": form_data.username, "role": "admin"},
expires_delta=ACCESS_TOKEN_EXPIRE,
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
return current_user
Saída:
# Função definida com sucesso
5. Hashing de Senha
(1) passlib + bcrypt
Senhas nunca são armazenadas em texto puro; elas são salgadas e hasheadas usando o algoritmo bcrypt.
(1) ▶ Exemplo: Hashing e Autenticação de Senha
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
# Teste
hashed = hash_password("my-secret-password")
print(f"Hashed: {hashed[:30]}...")
print(f"Verify correct: {verify_password('my-secret-password', hashed)}")
print(f"Verify wrong: {verify_password('wrong-password', hashed)}")
Saída:
Hashed: $2b$12$K3YQ8z9wB5eF7gH1j...
Verify correct: True
Verify wrong: False
(2) Melhores Práticas para Segurança de Senha
| Prática | Descrição |
|---|---|
| Usar bcrypt | Hashing salgado adaptativo para prevenir rainbow tables |
| Não usa MD5/SHA256 | Sem salt; cálculos muito rápidos, vulnerável a ataques de força bruta |
| Comprimento da senha ≥ 8 | Impor comprimento mínimo |
| Hash antes da verificação | Prevenir ataques de timing (integrado no passlib) |
| SECRET_KEY é longo o suficiente | String aleatória de pelo menos 32 caracteres; armazene em variável de ambiente |
6. Matriz de Permissões Multi-Tenant SaaS
(1) Design de Permissões por Nível de Assinatura
(1) ▶ Exemplo: Dependências de Permissão Baseadas em Assinatura
from fastapi import Depends, HTTPException
SUBSCRIPTION_LIMITS = {
"free": {"max_import": 1000, "websocket": False, "export": False},
"pro": {"max_import": 100000, "websocket": True, "export": True},
"enterprise": {"max_import": 1000000, "websocket": True, "export": True},
}
def require_subscription(min_level: str):
LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
async def check_subscription(user: dict = Depends(get_current_user)):
user_level = user.get("subscription", "free")
if LEVELS.get(user_level, 0) < LEVELS.get(min_level, 0):
raise HTTPException(
status_code=403,
detail=f"Requires {min_level} subscription. Current: {user_level}",
)
return user
return check_subscription
@app.get("/api/v1/analytics")
async def get_analytics(user=Depends(require_subscription("pro"))):
return {"total_products": 1000000, "active_users": 5000}
@app.post("/api/v1/prices/bulk")
async def bulk_import(
prices: list[PriceCreate],
user=Depends(require_subscription("free")),
):
limits = SUBSCRIPTION_LIMITS[user.get("subscription", "free")]
if len(prices) > limits["max_import"]:
raise HTTPException(
status_code=403,
detail=f"Import limit: {limits['max_import']} for {user['subscription']} plan",
)
return {"imported": len(prices)}
Saída:
# Função definida com sucesso
| Endpoint | Free | Pro | Enterprise |
|---|---|---|---|
| GET /products | 1.000/dia | Ilimitado | Ilimitado |
| POST /prices | 1.000/lote | 100.000/lote | 1.000.000/lote |
| WebSocket /ws/prices | - | ✓ | ✓ |
| GET /analytics | - | ✓ | ✓ |
| Exportação CSV | - | ✓ | ✓ |
| Limitação de Taxa de API | 100/minuto | 1.000/minuto | Ilimitado |
❓ Perguntas Frequentes
📖 Resumo
- Um JWT consiste em três partes: Header, Payload e Signature. Autenticação stateless suporta inerentemente sistemas distribuídos.
- O Access Token (curto prazo) é usado para acesso à API, enquanto o Refresh Token (longo prazo) é usado para renovação; este sistema de dois tokens melhora a segurança.
OAuth2PasswordBearerIntegra ferramentas de segurança do FastAPI; Swagger UI exibe automaticamente o formulário de login- Senhas são armazenadas como hashes bcrypt; passlib gerencia a lógica de salting e verificação.
- Matriz de Permissões de Assinatura de Três Níveis do PriceTracker: Free/Pro/Enterprise, implementada via DI
require_subscription
📝 Exercícios
- Exercício Básico (Dificuldade ⭐): Implemente o endpoint
/auth/loginpara receber nome de usuário e senha; após autenticação, retorne um JWT Access Token (válido por 30 minutos). UseDepends(oauth2_scheme)para parsear o usuário no endpoint/me. Dica:OAuth2PasswordRequestForm+jwt.encode - Exercício Avançado (Dificuldade ⭐⭐): Adicione um mecanismo de refresh token. O access token é válido por 15 minutos, e o refresh token é válido por 7 dias. Implemente o endpoint
/auth/refreshpara trocar o refresh token por um novo access token. Dica: Duas funçõescreate_token+ diferentes tempos de expiração - Desafio (Dificuldade ⭐⭐⭐): Implemente um sistema completo de registro/login/permissões—armazene senhas para o endpoint de registro usando hashing bcrypt; use a cadeia DI
require_subscription(min_level)para verificar níveis de assinatura; limite usuários Free a importar 1.000 registros, enquanto usuários Pro não têm limite. Dica:hash_password+verify_password+ DIrequire_subscription
---|



