orfeu

Cliente Python do autorizador Argos (OIDC + RBAC): autentica o backend no Argos, sincroniza o catálogo de operações, valida as credenciais de quem chama (JWT, token de aplicação, token de API), decide a permissão de cada operação, registra a auditoria e emite um log estruturado por chamada.

Sem OpenTelemetry e sem dependência de observabilidade: a correlação é o header W3C traceparent, a auditoria vai direto para o Argos e o log usa o logging padrão do Python.

pip install "orfeu==0.9.9"
pip install "orfeu[fastapi]==1.2.0"      # ou [flask], [django], [strawberry]

A 0.9.9 é o pré-lançamento (2026-10-05), publicado enquanto a validação nos apps acontece. A 1.0.0 sai depois, sem mudança de API prevista.

Python ≥ 3.11. Dependências: httpx, PyJWT[crypto].

FastAPI

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from orfeu import Level, Method
from orfeu.adapters.fastapi import Orfeu, install_envelope, operation


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield                     # o seu lifespan; o orfeu NÃO precisa ser chamado aqui


app = FastAPI(lifespan=lifespan)
install_envelope(app)
orfeu = Orfeu(
    app,
    authorizer_url="https://argos.exemplo/api",
    realm_id="meu-realm",
    app_id="<uuid do backend em auth.apps>",
    secret="<segredo>",
)


@app.get("/public")
@operation(id="public.info", level=Level.PUBLIC)
async def public_info():
    return {"status": 200, "message": "", "data": "olá"}


@app.get("/items")
@operation(id="items.list", level=Level.PRIVATE)
async def list_items(request: Request):
    user = request.state.user  # TokenClaims (None no trilho do x-api-token)
    return {"status": 200, "message": "", "data": {"user": user.sub}}

O Orfeu(app, ...) liga o bootstrap (token do app → JWKS → sync → operações) ao lifespan do app: roda no startup com lifespan= ou com on_startup (e agenda a recarga das permissões); no shutdown, para a recarga e faz o flush da auditoria. Para controlar o momento (testes, scripts) há await orfeu.startup_async() e orfeu.startup(), públicos e idempotentes.

Parâmetros de Orfeu(...) (e de OrfeuClient(...)): authorizer_url, realm_id, app_id, secret e, só por nome, tls_verify=True (desliga a verificação TLS só nas chamadas ao Argos), audit=True (False desliga a auditoria; AuditOptions(...) ajusta lote/buffer) e permissions_refresh=True (False desliga a recarga periódica das permissões; PermissionsRefreshOptions(...) ajusta) e accept_any_realm=False (True aceita tokens de qualquer realm, buscando o JWKS do realm do token, como o acceptAnyRealm do Node; limites: realm_id só com letras, números, - e _ (nome ou UUID), codificado na URL; até 64 realms em cache (o menos usado sai); realm sem JWKS (404, inexistente, authorizer fora) recusado por 30 s sem nova busca; até 20 realms ainda não vistos por minuto). O mesmo vale para Flask e Strawberry; no Django, os settings ORFEU_TLS_VERIFY, ORFEU_AUDIT, ORFEU_PERMISSIONS_REFRESH e ORFEU_ACCEPT_ANY_REALM.

A guarda do FastAPI resolve a rota exatamente como o roteador vai despachar — include_router com prefixo, routers aninhados, Mount, path params — em qualquer FastAPI ≥ 0.111 (testado até 0.142). Ela falha fechada: se não conseguir resolver a rota, responde 500 route_unresolved sem rodar o handler; e se uma rota com @operation for alcançada sem guarda, a resposta é trocada por 500.

Rotas WebSocket com @operation também são guardadas: mesmos 3 headers no handshake e a mesma regra do HTTP; recusa = fechar com 1008 (policy violation) antes do accept, inclusive quando a rota não puder ser resolvida. Na conexão aceita, websocket.state.user (ou websocket.state.api_token).

Flask

from flask import Flask, g
from orfeu import Level, orfeu_headers
from orfeu.adapters.flask import Orfeu, operation

app = Flask(__name__)

@app.get("/items")
@operation(id="items.list", level=Level.PRIVATE)
def list_items():
    return {"user": g.orfeu_user.sub}      # g.orfeu_api_token no trilho do x-api-token

orfeu = Orfeu(app, authorizer_url="...", realm_id="...", app_id="...", secret="...")  # faz o startup na hora
# encerramento: orfeu.shutdown() (para a recarga das permissões e faz o flush); também roda no atexit

Django

MIDDLEWARE = ["orfeu.adapters.django.OrfeuMiddleware", ...]

ORFEU_AUTHORIZER_URL = "https://argos.exemplo/api"
ORFEU_REALM_ID = "meu-realm"
ORFEU_APP_ID = "<uuid>"
ORFEU_SECRET = "<segredo>"
ORFEU_TLS_VERIFY = True          # opcional
ORFEU_AUDIT = True               # opcional (False ou orfeu.AuditOptions)
ORFEU_PERMISSIONS_REFRESH = True # opcional (False ou orfeu.PermissionsRefreshOptions)
ORFEU_ALLOWED_CLIENTS = ["meu-front"]   # opcional: client_id aceitos (ver "Validação do x-access-token")
ORFEU_ISSUER_URL = "https://sso.exemplo"  # opcional: emissor exato
ORFEU_TRUSTED_PROXIES = ["10.0.0.0/8"]  # opcional: proxies do x-forwarded-for (padrão: loopback + redes privadas)
ORFEU_CHECK_REVOCATION = False   # opcional: True ou lista de operações sensíveis

Na view: request.orfeu_user (TokenClaims; None numa chamada anônima a optional) ou request.orfeu_api_token (ApiTokenClaims). WSGI ou ASGI: as chamadas de rede rodam no loop próprio do client (OrfeuClient.run_sync), nunca asyncio.run() por requisição.

Strawberry

from orfeu import Level
from orfeu.adapters.strawberry import Orfeu, operation

orfeu = Orfeu(authorizer_url="...", realm_id="...", app_id="...", secret="...")
schema = strawberry.Schema(query=Query, extensions=[orfeu.extension])
# no lifespan: await orfeu.startup_async()   (ou orfeu.startup() síncrono) — idempotentes
# no fim do lifespan: await orfeu.shutdown_async()   (para a recarga das permissões, faz o flush)

No resolver: info.context["orfeu_user"]; orfeu_headers(info.context) para repassar o contexto. Em subscriptions (graphql-ws) as credenciais vêm do payload de connection_init ({"x-access-token": "..."}), e as tags em routeId/clientId. Um campo negado vira erro GraphQL com extensions = {"code": "UNAUTHENTICATED", "status": 401} (OrfeuUnauthorizedError) ou {"code": "FORBIDDEN", "status": 403} (OrfeuForbiddenError). Cada campo guardado é uma operação (log e auditoria). Funciona em execução assíncrona e com execute_sync.

Recarga das permissões

O catálogo de operações (com os grupos liberados de cada operação protected) é buscado no startup() e, por padrão, revalidado a cada 60 s (±20% por instância): liberar, revogar ou excluir um grupo no Argos passa a valer em até ~1 min, sem reiniciar o serviço. A revalidação manda If-None-Match com o último ETag (Argos sem ETag: funciona igual, só não economiza o corpo); 304 não troca nada, 200 troca o catálogo de uma vez, 401 renova o token do app e repete uma vez.

Se o Argos estiver fora (rede/5xx/401 repetido), o catálogo anterior continua valendo (nunca libera acesso por falha) e sai um aviso no logger orfeu.permissions, no máximo um a cada 5 min. Quando o catálogo muda, sai um INFO em orfeu.permissions só com contagens (operations, operations_changed, groups_changed), sem ids de usuário.

from orfeu import PermissionsRefreshOptions

Orfeu(app, ..., permissions_refresh=PermissionsRefreshOptions(interval_s=30))  # ajusta
Orfeu(app, ..., permissions_refresh=False)                                      # desliga (teste/caso excepcional)
# Django: ORFEU_PERMISSIONS_REFRESH = False  (ou um PermissionsRefreshOptions)

Para o /health: client.permissions_stale (True quando o último sucesso tem 5 × o intervalo ou mais — o serviço segue decidindo com o catálogo anterior) e client.last_permissions_sync_at (datetime UTC).

client.refresh_operations() revalida na hora (nunca lança; retorna True em sucesso) e client.fetch_operations() continua público (lança OrfeuError em falha).

Credenciais

Quem chama manda a credencial crua num destes headers (nunca Authorization/Bearer, que é ignorado):

Tokens opacos passam por POST /realms/{realm}/tokens/introspect autenticado com o token do próprio backend; a resposta fica em cache por token (60 s; inativo, 10 s). Um Argos sem esse endpoint (404) faz o x-api-token cair no POST /apps/tokens/validate antigo e o x-app-token ser recusado.

Opções de segurança do OrfeuClient

Valem no OrfeuClient e nos adapters FastAPI, Flask e Strawberry (argumentos keyword-only); no Django, pelas settings ORFEU_* acima. Sem o argumento, cada uma lê a variável de ambiente indicada.

Validação do x-access-token

O JWT é validado localmente contra o JWKS do realm e recusado (401) quando: o algoritmo não é RS256; assinatura ou kid não conferem; falta exp, sub ou iss, ou venceu; o iss não é o do realm do token (com issuer_url, exatamente {issuer_url}/realms/{realm_id}, como o Argos emite — SERVER_URL/realms/{realm}; sem, uma URL http(s) terminada em /realms/{realm_id}; o authorizer_url costuma ser endereço interno e por isso não é usado); o realm não é o do app (modo estrito); é um token de aplicação (type: "app", client_credentials), que não vale como usuário; ou allowed_clients está ligado e o client_id/azp não está na lista.

Falha ao falar com o Argos (JWKS, /apps/token, introspecção) nunca vira 500: em public a chamada segue anônima; nos outros níveis, 401.

allowed_clients — quais fronts podem chamar este backend. O access token do Argos é do realm inteiro e traz os grupos do usuário no client que fez o login. Sem a opção, um usuário com qualquer grupo em outro client do realm passa nas operações private (elas só exigem algum grupo). Apps com front próprio — o Hefesto, por exemplo — devem ligar (allowed_clients=["hefesto-ui"] ou ORFEU_ALLOWED_CLIENTS=hefesto-ui). O padrão segue desligado porque há backends que atendem qualquer app do realm de propósito (o admin do Argos). Vale também para o x-app-token.

Revogação (check_revocation). A validação local não enxerga revogação: um access token revogado (logout, /revoke, sessão encerrada) continua valendo até o exp — no Argos, até 10 min (CORE_TOKEN_LIFE). Para operações sensíveis, check_revocation=["users:delete"] confere o token na introspecção RFC 7662 (POST /realms/{realm}/protocol/openid-connect/token/introspect, com o token do app) antes de liberar; check_revocation=True confere em toda validação. Cache por token de 10 s (ativo) / 60 s (inativo); Argos fora do ar → 401 (fail-closed). Com accept_any_realm, só tokens do realm do app podem ser conferidos.

Regra de decisão (igual em FastAPI HTTP/WebSocket, Django, Flask e Strawberry — e no @orfeu/node)

1. public: nunca bloqueia. Se vier credencial (mesma ordem abaixo) e ela for válida, o usuário ou o token fica disponível para o handler como nos outros níveis (sem exigir permissão); credencial inválida é ignorada — o handler roda anônimo, sem 401 e sem vazar nada.
2. Ordem: x-access-token → x-app-token → x-api-token. A primeira presente é a que vale; as outras nem são lidas.
3. Credencial presente e inválida (expirada, revogada, de outro realm, sem permissão no caso do api-token) → 401. Nunca vira anônimo.
4. optional sem nenhuma credencial → o handler roda anônimo.
5. private/protected sem credencial → 401; usuário válido sem permissão → 403.
6. O corpo do 401 é genérico — {"status": 401, "message": "Invalid or missing credentials.", "data": null, "error": "unauthorized", "detail": "..."}; o motivo real vai só para o log e a auditoria. O 403 diz "Insufficient permissions.".

Operações: Level e Method

Use as constantes da lib (como o status.HTTP_404_NOT_FOUND do FastAPI). São StrEnum com exatamente os valores aceitos pelo Argos, então as strings iguais também funcionam; qualquer outro valor lança OrfeuConfigError já no registro (no @operation, em tempo de import, e no register_operation).

Sem method=, o @operation infere pelo nome da função: get_/list_/fetch_/read_ → READ; delete_/remove_/destroy_ → DELETE; contém stream/subscribe/watch/listen → STREAM; o resto → WRITE. Para fixar: @operation(id="tags:sync", level=Level.PROTECTED, method=Method.WRITE). Um (identifier, method) repetido é registrado uma vez só.

Trace (W3C traceparent)

Cada chamada guardada tem um OrfeuTrace(trace_id, span_id, entry, parent_span_id, flags):

• chegou com traceparent válido → continua o trace (trace_id igual, o span de quem chamou vira parent_span_id, entry=False, o bit sampled é propagado);
• chegou sem, ou com um inválido para o W3C (maiúsculas, versão ff, ids zerados, campos extras na versão 00) → trace novo, entry=True. Versões futuras são lidas pelos 4 primeiros campos.

parse_traceparent, format_traceparent e resolve_trace são públicos. O SDK não olha span ativo de biblioteca nenhuma.

IP do cliente (só em eventos entry): o primeiro valor do x-forwarded-for nunca é usado (quem chama escreve o que quiser nele). Conexão direta de um endereço fora de trusted_proxies → o IP do socket (cabeçalhos ignorados); vinda de proxy confiável (ou socket que não é IP) → a entrada mais à direita do x-forwarded-for que não é proxy confiável (todas confiáveis: a mais à esquerda); sem ele, x-real-ip; senão o socket. O padrão (orfeu.trace.DEFAULT_TRUSTED_PROXIES) confia em loopback e redes privadas; se os usuários também chegam por rede privada sem proxy, configure trusted_proxies só com os IPs dos proxies.

Auditoria

Toda chamada guardada, em todos os adapters, vira um evento em POST /realms/{realm}/audit/events, enviado em lote por uma thread em segundo plano (1 s ou 200 eventos; buffer de 10 mil; backoff até 60 s; 404 pausa 10 min). Formato idêntico ao do @orfeu/node; o campo sdk é python/<versão do orfeu>. No handler, annotate_operation(target_type=..., target_id=..., details=...) acrescenta o alvo. O flush roda no shutdown (FastAPI) e no atexit; client.flush_audit() força.

O núcleo do evento — quem (user_id/token_id/actor_type), o quê (operation.identifier) e onde (route_id/client_id) — vai sempre que disponível (ids só no formato GUID, que é o que o Argos guarda). O resto é complementar e nunca derruba o evento: client_ip inválido é omitido; duration_ms, message (500), target_type/target_id (100/200), service_instance (200) e status ficam dentro dos limites; details acima de 4 KB vira {"truncated": true, "original_bytes": n}.

Log de operação

Uma linha por chamada guardada no logger orfeu.operation (todos os adapters), via logging padrão. Os campos vão como atributos do LogRecord, todos serializáveis em JSON; opcionais só aparecem quando conhecidos:

Sucesso: INFO "operação executada"; falha: ERROR com a mensagem do erro. Para uma linha JSON por registro, sem biblioteca extra:

import logging
from orfeu import JsonLogFormatter

handler = logging.StreamHandler()
handler.setFormatter(JsonLogFormatter())
logging.getLogger("orfeu").addHandler(handler)
logging.getLogger("orfeu").setLevel(logging.INFO)

{"timestamp": "2026-10-04T12:00:00.123456Z", "level": "INFO", "logger": "orfeu.operation", "message": "operação executada", "operation": "items.list", "operation_id": "5b1e…9c2a", "operation_method": "read", "operation_level": "private", "agent_id": "…", "actor_type": "user", "permission_result": "allowed", "duration_ms": 3.21, "user": "…", "trace_id": "4bf9…4736", "span_id": "00f0…02b7"}

Outros loggers: orfeu.audit (falhas de entrega da auditoria, prefixo [orfeu]), orfeu.envelope (exceção inesperada convertida em 500) e orfeu.guard (rota que a guarda não conseguiu resolver).

Envelope { status, message, data }

install_envelope(app) (FastAPI) faz todo erro responder no envelope: EnvelopeError (BadRequest 400, Unauthorized 401, Forbidden 403, NotFound 404, Conflict 409, Unprocessable 422, Unavailable 503) com a mensagem dele; HTTPException com o detail; 422 de validação com as mensagens unidas por ; ; o resto vira 500 "Erro inesperado no serviço.". ok(data, message, status) monta o sucesso; Envelope[T] documenta o contrato no OpenAPI. Os 401/403 do próprio middleware também trazem status/message/data (mais error/detail, por compatibilidade).

Cliente avulso

from orfeu import Level, Method, OperationDescriptor, OrfeuClient

client = OrfeuClient("https://argos.exemplo/api", "meu-realm", "<app_id>", "<segredo>")
client.register_operation(OperationDescriptor("items.list", Method.READ, Level.PRIVATE))
await client.startup()                     # token do app → JWKS → sync → operações (+ recarga periódica)
claims = await client.validate_token(raw_jwt)
allowed = client.check_permission(claims, "items.list", "read")
await client.close()                       # para a recarga e faz o flush da auditoria

Licença

Licença Orfeu 1.0: uso livre, inclusive comercial e dentro de aplicações, sem modificar nem redistribuir — ver LICENSE.

Vindo do caronte-sdk

O orfeu 0.9.9 é o caronte-sdk renomeado, sem OpenTelemetry e com as falhas corrigidas. Troque pip install caronte-sdk por orfeu, import caronte por import orfeu e os nomes Caronte*/caronte_*/CARONTE_* por Orfeu*/orfeu_*/ORFEU_*; remova otel_collector_url. A regra de autorização foi unificada (a ordem dos headers e o tratamento de token inválido/optional mudaram em alguns adapters). Tabela completa e o que muda sem OTel: MIGRATION.md.