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]==0.9.9"      # 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 o flush da auditoria roda no shutdown. 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) e audit=True (False desliga a auditoria; AuditOptions(...) ajusta lote/buffer). O mesmo vale para Flask e Strawberry; no Django, os settings ORFEU_TLS_VERIFY e ORFEU_AUDIT.

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(app, authorizer_url="...", realm_id="...", app_id="...", secret="...")  # faz o startup na hora

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)

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 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.

Credenciais

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

Header	Quem	Permissão	Validação
x-access-token	usuário logado numa interface	grupos do usuário (claim do JWT)	local: JWT RS256 + JWKS (kid obrigatório, realm igual ao do backend)
x-app-token	usuário fora da interface (MCP, script), token pessoal	grupos do usuário no client + nível escolhido na criação	introspecção
x-api-token	outro backend	só as operações concedidas a esse token NESTE backend	introspecção

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.

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).

Constante	Valor	Quem acessa / o que é
Level.PUBLIC	public	todos, sem token
Level.OPTIONAL	optional	como PRIVATE, mas sem credencial o handler roda anônimo
Level.PRIVATE	private	qualquer usuário autenticado com ao menos um grupo
Level.PROTECTED	protected	usuários cujos grupos cruzam os allowed_groups da operação
Method.READ	read	leitura
Method.WRITE	write	criação/alteração
Method.DELETE	delete	remoção
Method.STREAM	stream	assinatura/fluxo contínuo

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.

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:

Campo	Tipo	Conteúdo
operation	str	identificador da operação (@operation(id=...))
operation_id	str?	UUID da operação no Argos (auth.operations), quando o SDK já o conhece (busca no startup, depois do sync)
operation_method / operation_level	str	Method / Level da operação
agent_id	str	app_id deste backend
actor_type	str	user, app_token, api_token, public, unauthenticated
permission_result	str	allowed, forbidden, unauthorized, public
duration_ms	float	duração, 2 casas
user	str?	sub do usuário
token_id	str?	id do token opaco
route_id / client_id	str?	x-route-id / x-client-id como vieram
trace_id / span_id	str?	ids W3C da chamada

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
claims = await client.validate_token(raw_jwt)
allowed = client.check_permission(claims, "items.list", "read")

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.