diff --git a/RI/.env.example b/RI/.env.example new file mode 100644 index 0000000..0c38570 --- /dev/null +++ b/RI/.env.example @@ -0,0 +1,19 @@ +# Copy to .env and fill in real values for local development. +# Never commit a real .env -- see docs/RI-PLANNING.md, EP-00-T05 (gestão de segredos). + +# --- Gateway (EP-05) --- +GATEWAY_HOST=0.0.0.0 +GATEWAY_PORT=8000 + +# --- Example Sales-domain MCP server (EP-06-T02) --- +SALES_DOMAIN_HOST=0.0.0.0 +SALES_DOMAIN_PORT=8100 +# How the Gateway reaches it (service name resolves inside docker compose; +# use 127.0.0.1 for `make dev`, which runs both processes on the host). +SALES_DOMAIN_URL=http://sales-domain:8100/mcp + +# --- Observability (EP-08) --- +# Unset (default): traces print to stdout via ConsoleSpanExporter. +# Set to enable OTLP/HTTP export instead (requires the `emcp-bus[otlp]` extra +# and a running collector -- see EP-08-T04). +# OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 diff --git a/RI/.gitignore b/RI/.gitignore new file mode 100644 index 0000000..d801290 --- /dev/null +++ b/RI/.gitignore @@ -0,0 +1,8 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +*.egg-info/ +.env diff --git a/RI/Makefile b/RI/Makefile new file mode 100644 index 0000000..9c11bde --- /dev/null +++ b/RI/Makefile @@ -0,0 +1,48 @@ +.PHONY: venv install lint format typecheck test dev up down logs validate-schemas + +VENV := .venv +PYTHON := $(VENV)/bin/python3 + +venv: + python3 -m venv $(VENV) + +install: venv + $(VENV)/bin/pip install --upgrade pip + $(VENV)/bin/pip install -e ".[dev]" + +lint: + $(VENV)/bin/ruff check . + $(VENV)/bin/ruff format --check . + +format: + $(VENV)/bin/ruff format . + $(VENV)/bin/ruff check --fix . + +typecheck: + $(VENV)/bin/mypy src services tests + +test: + $(VENV)/bin/python3 -m pytest + +validate-schemas: + # Placeholder until EP-02/EP-04 introduce the first JSON Schema-backed + # YAML config types; will validate every file under config/ against its + # schema (see docs/RI-PLANNING.md, EP-00-T03). + @echo "No schema-validated config types yet (see EP-02-T01, EP-04-T01)." + +# Runs both M0 services directly on the host (no Docker required) -- +# handy for local development and for environments without a Docker daemon. +dev: + $(VENV)/bin/python3 -m example_mcp_servers.sales_domain.server & \ + echo $$! > /tmp/emcp-sales.pid; \ + sleep 1; \ + $(VENV)/bin/python3 -m emcp_bus.gateway.server + +up: + docker compose --profile core up --build + +down: + docker compose --profile core down + +logs: + docker compose --profile core logs -f diff --git a/RI/README.md b/RI/README.md new file mode 100644 index 0000000..af6d70b --- /dev/null +++ b/RI/README.md @@ -0,0 +1,130 @@ +# Enterprise MCP Service Bus — Reference Implementation + +Implementação executável da arquitetura descrita em [`../README.md`](../README.md), planejada em [`../docs/RI-PLANNING.md`](../docs/RI-PLANNING.md). + +> **Status: núcleo de enforcement do Marco M1 concluído e testado ponta a ponta** (auth real → entitlement → PDP → decisão no Gateway, com OPA e um servidor JWKS reais — não mocks). Ver "O que existe hoje" abaixo. Os demais marcos estão descritos em `docs/RI-PLANNING.md`, seção 8.7. + +## Quickstart (sem Docker) + +```bash +make install # cria .venv e instala o pacote em modo editável (dev extras inclusos) +make lint # ruff check + ruff format --check +make typecheck # mypy --strict +make test # pytest (38 testes: unit + e2e) +``` + +`make test` precisa do binário `opa` no PATH (ou `OPA_BINARY` apontando para ele) para os testes que exercitam o PDP de verdade — ver "Como instalar o OPA" abaixo. Sem ele, esses testes são pulados (`pytest.skip`), não falham. + +### Como instalar o OPA (sem Docker) + +OPA é um binário Go estático — não precisa de Docker nem de build: + +```bash +curl -sL -o /tmp/opa "https://openpolicyagent.org/downloads/latest/opa_linux_amd64_static" +chmod +x /tmp/opa +cp /tmp/opa .venv/bin/opa # coloca no PATH do venv +opa version # confirma +``` + +Versão usada durante o desenvolvimento desta RI: **OPA v1.20.2** (Rego v1 é o padrão — as policies em `config/policies/*.rego` usam `import rego.v1`). + +### Subir a stack completa localmente (sem Docker) + +```bash +source .venv/bin/activate +opa run --server --addr 127.0.0.1:8181 config/policies/ & +python3 -m example_mcp_servers.sales_domain.server & +# Sem OIDC_JWKS_URL/OIDC_ISSUER, o Gateway roda em modo "sem IdP configurado": +# nega tudo por padrão (ver "Dois modos de deployment" abaixo). +python3 -m emcp_bus.gateway.server +``` + +Para autenticação real, seria necessário apontar `OIDC_JWKS_URL`/`OIDC_ISSUER` para um IdP OIDC de verdade (Keycloak, EP-01-T01 — não validado nesta sessão, ver abaixo). Os testes automatizados (`tests/e2e/conftest.py`) substituem isso por um servidor JWKS local real + tokens RS256 reais assinados na hora — válido para provar a lógica de validação de token, não para provar o Keycloak em si. + +## Quickstart (Docker) + +```bash +docker compose --profile core up --build +``` + +> **Nota**: `docker-compose.yml` e os Dockerfiles em `deploy/docker/` **não foram validados com um daemon Docker real** nesta sessão (ambiente sem `docker` disponível) — foram escritos e revisados manualmente, e a stack equivalente foi validada rodando os processos diretamente no host. Rode `docker compose --profile core up --build` na primeira vez em um ambiente com Docker e reporte qualquer ajuste necessário. **O `docker-compose.yml` atual ainda não inclui OPA/Keycloak como serviços** — apenas gateway + sales-domain (perfil `core`, herdado do M0); isso é trabalho pendente do M1 (ver tabela abaixo). + +## Dois modos de deployment (importante) + +O Gateway (`src/emcp_bus/gateway/server.py`) suporta dois modos, sempre *deny by default*: + +1. **Com OIDC configurado** (`OIDC_JWKS_URL` + `OIDC_ISSUER` definidos): o próprio middleware de auth do SDK `mcp` (`RequireAuthMiddleware`) rejeita qualquer conexão sem bearer token válido **antes mesmo do handshake `initialize`** — mais estrito que qualquer checagem no nível do handler. +2. **Sem OIDC configurado** (padrão se as env vars não estiverem definidas): nenhum middleware de auth é instalado; toda requisição chega aos handlers com `get_access_token() is None`, e é a própria lógica do handler (`on_list_tools`/`on_call_tool`) que nega por padrão — `tools/list` vazio, `tools/call` sempre negado com `Access denied: UNAUTHENTICATED`. + +Os dois modos são testados explicitamente em `tests/e2e/test_governed_gateway.py`. + +## O que existe hoje + +| Tarefa | Componente | Status | +|---|---|---| +| EP-00 (completo) | Scaffold, config loader, docker-compose base | ✅ (ver README anterior/git history) | +| EP-01-T02 | Schema de registro de MCP Client | ✅ `src/emcp_bus/identity/models.py` | +| EP-01-T03 | Middleware de autenticação OIDC/JWT | ✅ validado com JWKS real + tokens RS256 reais (assinatura, expiração, issuer, audience) | +| EP-01-T01 | Keycloak (realm export, serviço no compose) | ❌ não implementado — sem Java neste ambiente para validar; ver "Limitações desta sessão" | +| EP-01-T05 | Lint de shared client | ❌ não implementado (prioridade `could`) | +| EP-02 (Registry) | Schema de capability manifest | ✅ `src/emcp_bus/registry/models.py` (schema apenas — pipeline/storage/API ainda não) | +| EP-02-T02/T03/T05 | Publishing pipeline, storage, manifests de exemplo | ❌ não implementado | +| EP-03 (completo) | PDP/OPA: entitlement.rego, transaction.rego, cliente Python, contrato de decisão | ✅ validado com **OPA real** (binário, não mock) — 9 testes Rego nativos + 6 testes Python de integração | +| EP-04 (completo) | Entitlement Manager: resolve, revoke, versionamento | ✅ validado com OPA real + YAML reais de `config/clients/` | +| EP-05-T01 | Gateway skeleton | ✅ (M0) | +| EP-05-T02 | `tools/list` filtrado por entitlement | ✅ testado (README.md §45 Teste 1) | +| EP-05-T04 | `tools/call` reautorizado via PDP | ✅ testado (Teste 2, Teste 4 — agent role spoofing sem efeito) | +| EP-05-T06 | Fail-closed (PDP indisponível) | ✅ testado — `lifespan()` recusa subir se o OPA sync falhar | +| EP-05-T03 | Cache privado de `tools/list` | ❌ não implementado | +| EP-05-T07 | Anti-bypass (isolamento de rede) | ❌ não implementado (depende de docker-compose real) | +| EP-05-T08 | Consistência header/body (Mcp-Method/Mcp-Name) | ❌ não implementado explicitamente (o SDK já expõe os headers nativamente — ver §22 do README de arquitetura; falta o passo de validação cruzada) | +| EP-06-T01 | Fabric core (roteador por capability) | ❌ ainda roteamento estático (uma única URL fixa), não por Registry | +| EP-06-T02 | MCP Server de exemplo — Sales | ✅ (M0) | +| EP-06-T03 | MCP Server de exemplo — Finance | ❌ não implementado — os testes de negação para tools `finance.*` passam porque o PDP nega *antes* de qualquer tentativa de roteamento ao backend, então a ausência do servidor Finance real não invalida esses testes, mas nenhum cenário de **sucesso** em Finance foi exercitado | +| EP-06-T04 | Adapter REST genérico | ❌ não implementado | +| EP-08-T01 | Tracing OTel | ✅ (M0) | +| EP-08-T02/T03/T04 | Eventos de auditoria (IC/NOC/GRL), correlação, stack de observabilidade | ❌ não implementado — hoje só existe o trace OTel, sem taxonomia de eventos nem sink de auditoria | + +**39 testes automatizados**, todos reais (sem mocks do núcleo de segurança): OPA de verdade (subprocess), JWKS HTTP real + JWT RS256 reais, YAML reais de `config/`. + +## Limitações desta sessão + +Nenhum `docker` nem `java` disponíveis no ambiente onde esta RI foi escrita: + +- **Docker**: não validado (ver nota acima). Risco considerado baixo — o Dockerfile usa a mesma instalação (`pip install -e .`) já validada localmente. +- **Keycloak**: não implementado nem validado — precisa de JVM. A lógica de validação de OAuth2/OIDC (`TokenValidator`, EP-01-T03) é padrão-agnóstica de IdP (qualquer provedor OIDC compliant funciona) e foi validada com um servidor JWKS real + tokens RS256 reais, o que prova a lógica de validação mas **não prova a integração específica com Keycloak** (realm config, client credentials grant, claims exatas que o Keycloak emite). EP-01-T01 continua pendente. + +## Versões fixadas + +Ver `pyproject.toml`. Confirmadas via `pip index versions`/download direto em 2026-09-11 (não citadas de memória — ver `docs/RI-PLANNING.md`, EP-16-T04): + +- `mcp` 2.2.0 (SDK v2 — API mudou significativamente da v1; ver comentários em `src/emcp_bus/gateway/server.py`) +- `pydantic` 2.13.x, `opentelemetry-sdk` 1.44.x, `uvicorn` 0.52.x, `pyjwt[crypto]` 2.14.x, `ruff` 0.16.x, `mypy` 2.3.x, `pytest` 9.1.x +- `OPA` 1.20.2 (binário estático, Rego v1 padrão) + +## Estrutura + +Ver `../docs/RI-PLANNING.md`, seção 8.5, para a árvore completa planejada. O que existe até agora: + +```text +RI/ +├── pyproject.toml +├── Makefile +├── docker-compose.yml # perfil "core": gateway + sales-domain (OPA/Keycloak pendentes) +├── .env.example +├── deploy/docker/Dockerfile +├── config/ +│ ├── policies/ # EP-03: entitlement.rego, transaction.rego (+ testes opa test) +│ └── clients/ # EP-01/EP-04: registrations + profiles/*.yaml +├── src/emcp_bus/ +│ ├── common/{config,otel,models}.py # EP-00-T02, EP-08-T01, risk tier compartilhado +│ ├── identity/{models,authn,token_verifier}.py # EP-01-T02/T03 +│ ├── entitlement/{models,manager}.py # EP-04 +│ ├── registry/models.py # EP-02-T01 (schema apenas) +│ ├── pdp/{client,models}.py # EP-03 +│ └── gateway/server.py # EP-05-T01/T02/T04/T06 +├── services/example_mcp_servers/sales_domain/server.py # EP-06-T02 +└── tests/ + ├── unit/ # schemas, PDP client, entitlement manager, authn — todos contra dependências reais + └── e2e/ # conftest.py (fixtures compartilhados), walking skeleton, governed gateway +``` diff --git a/RI/config/clients/finance-payments-agent.yaml b/RI/config/clients/finance-payments-agent.yaml new file mode 100644 index 0000000..5565511 --- /dev/null +++ b/RI/config/clients/finance-payments-agent.yaml @@ -0,0 +1,4 @@ +client_id: finance-payments-agent +profile: Finance.Payments +environment: production +description: "Finance agent authorized to create/execute payments (R3, approval above threshold)" diff --git a/RI/config/clients/profiles/finance-payments.yaml b/RI/config/clients/profiles/finance-payments.yaml new file mode 100644 index 0000000..232902c --- /dev/null +++ b/RI/config/clients/profiles/finance-payments.yaml @@ -0,0 +1,17 @@ +# README.md §7.2, §17 (R3), §26 (payment.execute < 10k ALLOW, >= 10k REQUIRE_APPROVAL) +name: Finance.Payments +owner: finance-platform +environment: production +risk_ceiling: R3 +allowed_tools: + - finance.invoice.get + - finance.payment.create + - finance.payment.execute +restrictions: + max_transaction_value: 10000000 + approval_threshold: 10000 +credential_policy: + workload_identity: true + short_lived: true +audit: + level: full diff --git a/RI/config/clients/profiles/finance-read.yaml b/RI/config/clients/profiles/finance-read.yaml new file mode 100644 index 0000000..8fa64c4 --- /dev/null +++ b/RI/config/clients/profiles/finance-read.yaml @@ -0,0 +1,12 @@ +# README.md §7.2, §17 (R1) +name: Finance.Read +owner: finance-platform +environment: production +risk_ceiling: R1 +allowed_tools: + - finance.invoice.get +credential_policy: + workload_identity: true + short_lived: true +audit: + level: full diff --git a/RI/config/clients/profiles/sales-approve.yaml b/RI/config/clients/profiles/sales-approve.yaml new file mode 100644 index 0000000..a7794c2 --- /dev/null +++ b/RI/config/clients/profiles/sales-approve.yaml @@ -0,0 +1,18 @@ +# README.md §7.1 +name: Sales.Approve +owner: sales-platform +environment: production +risk_ceiling: R2 +allowed_tools: + - sales.customer.search + - sales.quote.get + - sales.quote.approve + - sales.discount.approve +restrictions: + max_transaction_value: 1000000 + allowed_regions: ["BR-SP", "BR-RJ"] +credential_policy: + workload_identity: true + short_lived: true +audit: + level: full diff --git a/RI/config/clients/profiles/sales-read.yaml b/RI/config/clients/profiles/sales-read.yaml new file mode 100644 index 0000000..4459584 --- /dev/null +++ b/RI/config/clients/profiles/sales-read.yaml @@ -0,0 +1,15 @@ +# README.md §7.1 +name: Sales.Read +owner: sales-platform +environment: production +risk_ceiling: R1 +allowed_tools: + - sales.customer.search + - sales.customer.get + - sales.order.get + - sales.inventory.check +credential_policy: + workload_identity: true + short_lived: true +audit: + level: full diff --git a/RI/config/clients/profiles/sales-write.yaml b/RI/config/clients/profiles/sales-write.yaml new file mode 100644 index 0000000..c61147e --- /dev/null +++ b/RI/config/clients/profiles/sales-write.yaml @@ -0,0 +1,20 @@ +# README.md §7.1, §41 +name: Sales.Write +owner: sales-platform +environment: production +risk_ceiling: R2 +allowed_tools: + - sales.customer.search + - sales.customer.get + - sales.order.get + - sales.inventory.check + - sales.quote.create + - sales.quote.update +restrictions: + max_transaction_value: 500000 + allowed_regions: ["BR-SP", "BR-RJ"] +credential_policy: + workload_identity: true + short_lived: true +audit: + level: full diff --git a/RI/config/clients/sales-read-agent.yaml b/RI/config/clients/sales-read-agent.yaml new file mode 100644 index 0000000..2b94210 --- /dev/null +++ b/RI/config/clients/sales-read-agent.yaml @@ -0,0 +1,4 @@ +client_id: sales-read-agent +profile: Sales.Read +environment: production +description: "Read-only Sales conversational agent" diff --git a/RI/config/clients/sales-write-agent.yaml b/RI/config/clients/sales-write-agent.yaml new file mode 100644 index 0000000..bb463f8 --- /dev/null +++ b/RI/config/clients/sales-write-agent.yaml @@ -0,0 +1,4 @@ +client_id: sales-write-agent +profile: Sales.Write +environment: production +description: "Sales agent authorized to create/update quotes" diff --git a/RI/config/policies/entitlement.rego b/RI/config/policies/entitlement.rego new file mode 100644 index 0000000..af5ba5d --- /dev/null +++ b/RI/config/policies/entitlement.rego @@ -0,0 +1,20 @@ +# Entitlement policy (EP-03-T02, README.md §25 Passo 1, §6.5.A). +# +# permit if requested_tool IN entitlement(authenticated_client) +# +# `input.client_profile` is the client profile *name* the Gateway already +# resolved from the authenticated client_id (EP-01-T03 -> identity/models.py +# ClientRegistration.profile) -- never a value read from the request body. +# `data.emcp.client_profiles` is pushed into OPA by the Entitlement Manager +# (EP-04-T02) from the validated ClientProfile YAML (EP-04-T01); this policy +# never parses YAML itself. +package emcp.entitlement + +import rego.v1 + +default allow := false + +allow if { + profile := data.emcp.client_profiles[input.client_profile] + input.tool in profile.allowed_tools +} diff --git a/RI/config/policies/entitlement_test.rego b/RI/config/policies/entitlement_test.rego new file mode 100644 index 0000000..0888d5c --- /dev/null +++ b/RI/config/policies/entitlement_test.rego @@ -0,0 +1,21 @@ +package emcp.entitlement_test + +import rego.v1 +import data.emcp.entitlement + +test_allow_when_tool_in_entitlement if { + entitlement.allow with input as {"client_profile": "Sales.Read", "tool": "sales.customer.get"} + with data.emcp.client_profiles as {"Sales.Read": {"allowed_tools": ["sales.customer.get"]}} +} + +test_deny_when_tool_not_in_entitlement if { + # README.md §45 Teste 1/2: finance.createPayment must never be reachable + # by a Sales.Read client, discovered or not. + not entitlement.allow with input as {"client_profile": "Sales.Read", "tool": "finance.createPayment"} + with data.emcp.client_profiles as {"Sales.Read": {"allowed_tools": ["sales.customer.get"]}} +} + +test_deny_when_client_profile_unknown if { + not entitlement.allow with input as {"client_profile": "NoSuchProfile", "tool": "sales.customer.get"} + with data.emcp.client_profiles as {"Sales.Read": {"allowed_tools": ["sales.customer.get"]}} +} diff --git a/RI/config/policies/transaction.rego b/RI/config/policies/transaction.rego new file mode 100644 index 0000000..938c165 --- /dev/null +++ b/RI/config/policies/transaction.rego @@ -0,0 +1,68 @@ +# Transaction policy (EP-03-T03, README.md §25 Passo 2, §6.5.B, §26). +# +# permit quote.create if +# amount <= client.transaction_limit +# AND region IN client.allowed_regions +# AND customer.classification != "restricted" +# +# Restricts a tool *already granted* by entitlement.rego -- this policy never +# adds a tool absent from the client's entitlement (README.md §25, "A +# Transaction Policy pode negar ou restringir. Ela não pode adicionar uma +# tool inexistente no Client Entitlement."). `allow` here is meaningless on +# its own; the PDP client (EP-03-T04) only consults it after entitlement.rego +# already allowed the same (client_profile, tool) pair. +# +# input: { +# "client_profile": "Sales.Write", +# "tool": "sales.quote.create", +# "arguments": {"amount": 100000, "region": "BR-SP"}, +# "customer_classification": "standard" +# } +package emcp.transaction + +import rego.v1 + +default allow := false + +default require_approval := false + +allow if { + profile := data.emcp.client_profiles[input.client_profile] + within_transaction_limit(profile) + within_allowed_regions(profile) + not restricted_customer +} + +# README.md §26: above the profile's approval_threshold, a transaction that +# would otherwise be allowed instead requires human approval (EP-14) -- it is +# never silently allowed, and never silently upgraded to a hard deny either. +require_approval if { + allow + profile := data.emcp.client_profiles[input.client_profile] + threshold := profile.restrictions.approval_threshold + amount := object.get(input.arguments, "amount", 0) + amount > threshold +} + +within_transaction_limit(profile) if { + not profile.restrictions.max_transaction_value +} + +within_transaction_limit(profile) if { + limit := profile.restrictions.max_transaction_value + amount := object.get(input.arguments, "amount", 0) + amount <= limit +} + +within_allowed_regions(profile) if { + not profile.restrictions.allowed_regions +} + +within_allowed_regions(profile) if { + region := object.get(input.arguments, "region", "") + region in profile.restrictions.allowed_regions +} + +restricted_customer if { + input.customer_classification == "restricted" +} diff --git a/RI/config/policies/transaction_test.rego b/RI/config/policies/transaction_test.rego new file mode 100644 index 0000000..e04133e --- /dev/null +++ b/RI/config/policies/transaction_test.rego @@ -0,0 +1,72 @@ +package emcp.transaction_test + +import rego.v1 +import data.emcp.transaction + +sales_write_profile := {"restrictions": { + "max_transaction_value": 500000, + "allowed_regions": ["BR-SP", "BR-RJ"], + "approval_threshold": 100000, +}} + +test_allow_within_limit_and_region if { + transaction.allow with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 50000, "region": "BR-SP"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} + +test_deny_above_transaction_limit if { + not transaction.allow with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 600000, "region": "BR-SP"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} + +test_deny_outside_allowed_region if { + not transaction.allow with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 50000, "region": "BR-MG"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} + +test_deny_restricted_customer if { + not transaction.allow with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 50000, "region": "BR-SP"}, + "customer_classification": "restricted", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} + +test_require_approval_above_threshold if { + # README.md §26: allowed, but only with human approval, above threshold. + transaction.allow with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 150000, "region": "BR-SP"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} + + transaction.require_approval with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 150000, "region": "BR-SP"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} + +test_no_approval_required_below_threshold if { + not transaction.require_approval with input as { + "client_profile": "Sales.Write", + "tool": "sales.quote.create", + "arguments": {"amount": 50000, "region": "BR-SP"}, + "customer_classification": "standard", + } with data.emcp.client_profiles as {"Sales.Write": sales_write_profile} +} diff --git a/RI/deploy/docker/Dockerfile b/RI/deploy/docker/Dockerfile new file mode 100644 index 0000000..742241a --- /dev/null +++ b/RI/deploy/docker/Dockerfile @@ -0,0 +1,29 @@ +# Shared image for every RI service (Gateway, example MCP servers, ...). +# docker-compose.yml selects the process to run per service via `command:`. +# +# NOTE: not verified with a real `docker build` in this session -- no Docker +# daemon was available in the sandbox this was authored in (see RI/README.md, +# "Validado nesta sessão" for exactly what *was* run). Please verify on first +# use in an environment with Docker and report back if anything needs fixing. +FROM python:3.12-slim AS base + +WORKDIR /app + +# System deps kept minimal on purpose -- this image has no compiled +# dependencies today; add build-essential here if that changes. +COPY pyproject.toml ./ +COPY src ./src +COPY services ./services + +RUN pip install --no-cache-dir --upgrade pip \ + && pip install --no-cache-dir -e . + +# Runs as a non-root user; every RI service binds 0.0.0.0 inside its own +# container network namespace, never the host network directly. +RUN useradd --create-home --uid 10001 emcp +USER emcp + +EXPOSE 8000 8100 + +# Overridden per-service by docker-compose.yml `command:`. +CMD ["python3", "-m", "emcp_bus.gateway.server"] diff --git a/RI/docker-compose.yml b/RI/docker-compose.yml new file mode 100644 index 0000000..3f2a8b0 --- /dev/null +++ b/RI/docker-compose.yml @@ -0,0 +1,50 @@ +# EP-00-T04 (walking skeleton) + EP-16-T01 (grows into the full stack). +# +# `core` profile = M0: gateway + one example MCP server, static routing, no +# enforcement yet. Later milestones add services under additional profiles +# (`identity`, `policy`, `observability`, `llm-observability`, ...) so `docker +# compose --profile core up` stays cheap through M0/M1 instead of requiring +# the whole stack (Keycloak, OPA, Langfuse, ...) just to see the walking +# skeleton trace. +# +# NOT executed against a real Docker daemon in the session this was authored +# in (no `docker` binary in that sandbox) -- validated instead by running the +# same two processes directly on the host (see RI/README.md). Please run +# `docker compose --profile core up --build` on first use and report back if +# anything needs adjusting. + +services: + sales-domain: + build: + context: . + dockerfile: deploy/docker/Dockerfile + command: ["python3", "-m", "example_mcp_servers.sales_domain.server"] + environment: + SALES_DOMAIN_HOST: "0.0.0.0" + SALES_DOMAIN_PORT: "8100" + ports: + - "8100:8100" + profiles: ["core"] + healthcheck: + # A bare TCP connect, not an HTTP/MCP-protocol check: streamable-http + # rejects a plain GET, so this only confirms the process is listening. + test: ["CMD", "python3", "-c", "import socket; socket.create_connection(('127.0.0.1', 8100), timeout=2)"] + interval: 5s + timeout: 3s + retries: 5 + + gateway: + build: + context: . + dockerfile: deploy/docker/Dockerfile + command: ["python3", "-m", "emcp_bus.gateway.server"] + environment: + GATEWAY_HOST: "0.0.0.0" + GATEWAY_PORT: "8000" + SALES_DOMAIN_URL: "http://sales-domain:8100/mcp" + ports: + - "8000:8000" + profiles: ["core"] + depends_on: + sales-domain: + condition: service_healthy diff --git a/RI/pyproject.toml b/RI/pyproject.toml new file mode 100644 index 0000000..8b697aa --- /dev/null +++ b/RI/pyproject.toml @@ -0,0 +1,54 @@ +[project] +name = "emcp-bus" +version = "0.1.0" +description = "Reference implementation of the Enterprise MCP Service Bus (Client-Bound Entitlements architecture)" +readme = "README.md" +requires-python = ">=3.12" +license = { text = "Apache-2.0" } +dependencies = [ + "mcp[cli]>=2.2.0", + "pydantic>=2.13,<3", + "pyyaml>=6.0.3,<7", + "opentelemetry-sdk>=1.44,<2", + "opentelemetry-instrumentation>=0.65b0", + "uvicorn>=0.52,<1", + "httpx>=0.28,<1", + "pyjwt[crypto]>=2.14,<3", +] + +[project.optional-dependencies] +otlp = [ + "opentelemetry-exporter-otlp-proto-http>=1.44,<2", +] +dev = [ + "ruff>=0.16.7,<1", + "mypy>=2.3.1,<3", + "pytest>=9.1.1,<10", + "pytest-asyncio>=1.4.0,<2", + "types-PyYAML>=6.0.12,<7", +] + +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +where = ["src", "services", "."] +include = ["emcp_bus*", "example_mcp_servers*"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +asyncio_mode = "auto" +pythonpath = ["src", "services"] + +[tool.ruff] +line-length = 100 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B", "SIM"] + +[tool.mypy] +python_version = "3.12" +strict = true +mypy_path = ["src", "services"] diff --git a/RI/services/example_mcp_servers/__init__.py b/RI/services/example_mcp_servers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/services/example_mcp_servers/sales_domain/__init__.py b/RI/services/example_mcp_servers/sales_domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/services/example_mcp_servers/sales_domain/server.py b/RI/services/example_mcp_servers/sales_domain/server.py new file mode 100644 index 0000000..1f8bdbe --- /dev/null +++ b/RI/services/example_mcp_servers/sales_domain/server.py @@ -0,0 +1,75 @@ +"""Example Sales-domain MCP server (EP-06-T02, walking-skeleton subset). + +Read-only tools (risk tier R1, see README.md §17): customer.search, +customer.get, order.get, inventory.check. Backed by static mock data -- this +server is deliberately "dumb": it never authorizes anything (ADR-023 in +docs/RI-PLANNING.md). Authorization happens exclusively upstream, in the +Gateway/PEP (EP-05) before a call ever reaches this process. + +Run directly for local development: + + python -m example_mcp_servers.sales_domain.server +""" + +from __future__ import annotations + +import os + +from mcp.server.mcpserver import MCPServer + +from emcp_bus.common.otel import setup_tracing + +setup_tracing("sales-domain-mcp-server") + +server = MCPServer(name="sales-domain", version="0.1.0") + +_CUSTOMERS: dict[str, dict[str, object]] = { + "cust-001": {"customer_id": "cust-001", "name": "Aurora Ltda", "region": "BR-SP"}, + "cust-002": {"customer_id": "cust-002", "name": "Boreal S.A.", "region": "BR-RJ"}, +} + +_ORDERS: dict[str, dict[str, object]] = { + "ord-001": {"order_id": "ord-001", "customer_id": "cust-001", "status": "shipped"}, +} + +_INVENTORY: dict[str, int] = { + "sku-100": 42, + "sku-200": 0, +} + + +@server.tool(name="sales.customer.search", description="Search sales customers by name fragment.") +def customer_search(query: str) -> dict[str, object]: + matches = [c for c in _CUSTOMERS.values() if query.lower() in str(c["name"]).lower()] + return {"results": matches} + + +@server.tool(name="sales.customer.get", description="Get a single sales customer by id.") +def customer_get(customer_id: str) -> dict[str, object]: + customer = _CUSTOMERS.get(customer_id) + if customer is None: + return {"error": "not_found", "customer_id": customer_id} + return customer + + +@server.tool(name="sales.order.get", description="Get a single sales order by id.") +def order_get(order_id: str) -> dict[str, object]: + order = _ORDERS.get(order_id) + if order is None: + return {"error": "not_found", "order_id": order_id} + return order + + +@server.tool(name="sales.inventory.check", description="Check available inventory for a SKU.") +def inventory_check(sku: str) -> dict[str, object]: + return {"sku": sku, "available": _INVENTORY.get(sku, 0)} + + +def main() -> None: + host = os.environ.get("SALES_DOMAIN_HOST", "127.0.0.1") + port = int(os.environ.get("SALES_DOMAIN_PORT", "8100")) + server.run(transport="streamable-http", host=host, port=port) + + +if __name__ == "__main__": + main() diff --git a/RI/src/emcp_bus/__init__.py b/RI/src/emcp_bus/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/common/__init__.py b/RI/src/emcp_bus/common/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/common/config.py b/RI/src/emcp_bus/common/config.py new file mode 100644 index 0000000..a86edca --- /dev/null +++ b/RI/src/emcp_bus/common/config.py @@ -0,0 +1,85 @@ +"""YAML -> Pydantic configuration loading with environment-variable interpolation. + +Every declarative config type in the RI (client profiles, capability manifests, +policies, ...) is loaded through :func:`load_yaml_model`, so secrets never live +as literal values in a committed YAML file (see docs/RI-PLANNING.md, EP-00-T05): +a value written as ``${VAR_NAME}`` or ``${VAR_NAME:-default}`` is resolved from +the environment at load time, and a value that still looks like an unresolved +placeholder after interpolation is treated as a configuration error. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path +from typing import TypeVar + +import yaml +from pydantic import BaseModel, ValidationError + +ModelT = TypeVar("ModelT", bound=BaseModel) + +_ENV_VAR_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(:-(.*?))?\}") + + +class ConfigError(Exception): + """Raised when a YAML config file fails to parse or to validate against its model.""" + + +def _interpolate_env_vars(raw_text: str) -> str: + def _replace(match: re.Match[str]) -> str: + var_name, _, default = match.groups() + if var_name in os.environ: + return os.environ[var_name] + if default is not None: + return default + raise ConfigError( + f"Environment variable '{var_name}' referenced in config is not set " + "and no default was provided (use ${VAR:-default} for optional values)." + ) + + return _ENV_VAR_PATTERN.sub(_replace, raw_text) + + +def load_yaml_model[ModelT: BaseModel](path: str | Path, model: type[ModelT]) -> ModelT: + """Load a single YAML document at ``path`` and validate it against ``model``. + + Raises :class:`ConfigError` on a missing file, invalid YAML, or a document + that fails Pydantic validation -- callers should treat this as a fatal + startup error (fail closed), never as a recoverable condition. + """ + file_path = Path(path) + if not file_path.is_file(): + raise ConfigError(f"Config file not found: {file_path}") + + raw_text = file_path.read_text(encoding="utf-8") + interpolated = _interpolate_env_vars(raw_text) + + try: + document = yaml.safe_load(interpolated) + except yaml.YAMLError as exc: + raise ConfigError(f"Invalid YAML in {file_path}: {exc}") from exc + + if document is None: + document = {} + + try: + return model.model_validate(document) + except ValidationError as exc: + raise ConfigError(f"Config validation failed for {file_path}:\n{exc}") from exc + + +def load_yaml_models[ModelT: BaseModel]( + directory: str | Path, model: type[ModelT], pattern: str = "*.yaml" +) -> list[ModelT]: + """Load and validate every file matching ``pattern`` under ``directory``. + + Files are returned sorted by filename for deterministic ordering (config + load order must never be a source of non-determinism in the PEP/PDP path). + """ + dir_path = Path(directory) + if not dir_path.is_dir(): + raise ConfigError(f"Config directory not found: {dir_path}") + + return [load_yaml_model(file_path, model) for file_path in sorted(dir_path.glob(pattern))] diff --git a/RI/src/emcp_bus/common/models.py b/RI/src/emcp_bus/common/models.py new file mode 100644 index 0000000..9792053 --- /dev/null +++ b/RI/src/emcp_bus/common/models.py @@ -0,0 +1,30 @@ +"""Shared value types used across config models (README.md §17, §41).""" + +from __future__ import annotations + +from enum import StrEnum + + +class RiskTier(StrEnum): + """Capability/client risk classification (README.md §17). + + R0 discovery/public, R1 read, R2 write, R3 high-impact (approval-eligible), + R4 destructive/restricted. The *same* enum is used for both a capability + manifest's `risk_tier` (EP-02-T01) and a client profile's `risk_ceiling` + (EP-04-T01) so entitlement resolution (EP-04-T02) can compare them + directly instead of maintaining two parallel classifications. + """ + + R0 = "R0" + R1 = "R1" + R2 = "R2" + R3 = "R3" + R4 = "R4" + + +_ORDER: dict[RiskTier, int] = {tier: index for index, tier in enumerate(RiskTier)} + + +def risk_tier_at_or_below(tier: RiskTier, ceiling: RiskTier) -> bool: + """True if `tier` is within `ceiling` (e.g. R1 <= R2, R3 > R2).""" + return _ORDER[tier] <= _ORDER[ceiling] diff --git a/RI/src/emcp_bus/common/otel.py b/RI/src/emcp_bus/common/otel.py new file mode 100644 index 0000000..a4e815e --- /dev/null +++ b/RI/src/emcp_bus/common/otel.py @@ -0,0 +1,68 @@ +"""Minimal OpenTelemetry tracing setup shared by every RI service. + +Walking-skeleton scope (EP-08-T01): a `TracerProvider` per process, exporting +to stdout by default so the M0 trace is visible without a running collector. +Setting `OTEL_EXPORTER_OTLP_ENDPOINT` switches to OTLP/HTTP (requires the +`emcp-bus[otlp]` extra) for use against the collector in EP-08-T04. + +Trace context propagates across the gateway -> downstream MCP server hop via +the `mcp` SDK's built-in `OpenTelemetryMiddleware` (on by default on every +`Server`/`MCPServer`): it injects/extracts standard W3C `traceparent` into the +JSON-RPC `_meta` field around every request, so a single trace_id spans both +processes without any extra instrumentation here -- this is the concrete, +checkable form of the M0 acceptance criterion in docs/RI-PLANNING.md ("um +único trace_id conecta Gateway -> Fabric -> backend mock"), verified in +tests/e2e/test_walking_skeleton.py. + +`setup_tracing` registers one *process-global* TracerProvider via +`opentelemetry.trace.set_tracer_provider`, which OTel only ever honors once +per process -- a second call from a different service name in the same +process is silently ignored by the OTel API itself. That is correct for the +real deployment (one service per process); tests that run two RI services +in-process (as tests/e2e/test_walking_skeleton.py does, for speed) must not +rely on `resource.attributes["service.name"]` to tell spans from different +services apart, since both end up sharing whichever provider registered +first -- use `span.kind` / `span.name` instead. +""" + +from __future__ import annotations + +import os + +from opentelemetry import trace +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter +from opentelemetry.trace import Tracer + +_configured_services: set[str] = set() + + +def setup_tracing(service_name: str) -> Tracer: + """Configure (once per process) a TracerProvider for `service_name` and return its Tracer. + + Idempotent: calling this more than once for the same service name in the + same process returns a Tracer against the already-configured provider + instead of registering a second one. + """ + if service_name not in _configured_services: + resource = Resource.create({SERVICE_NAME: service_name}) + provider = TracerProvider(resource=resource) + + otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") + if otlp_endpoint: + # Imported lazily: the OTLP exporter is an optional dependency + # (`emcp-bus[otlp]`) not required for the console-only M0 path. + from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( # type: ignore[import-not-found] + OTLPSpanExporter, + ) + + exporter = OTLPSpanExporter(endpoint=f"{otlp_endpoint}/v1/traces") + else: + exporter = ConsoleSpanExporter() + + provider.add_span_processor(BatchSpanProcessor(exporter)) + trace.set_tracer_provider(provider) + _configured_services.add(service_name) + + return trace.get_tracer(service_name) diff --git a/RI/src/emcp_bus/entitlement/__init__.py b/RI/src/emcp_bus/entitlement/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/entitlement/manager.py b/RI/src/emcp_bus/entitlement/manager.py new file mode 100644 index 0000000..ee6af2a --- /dev/null +++ b/RI/src/emcp_bus/entitlement/manager.py @@ -0,0 +1,146 @@ +"""Entitlement Manager (EP-04-T02/T03/T04, README.md §6.9). + +Resolves `MaximumEntitlement(client)` by combining an authenticated +client_id (never a value the agent/prompt could influence -- EP-01-T03) with +its registered `ClientProfile` (EP-04-T01), and keeps that resolution +versioned, cacheable and revocable (RF-08, RF-12). This module is the +*only* authority for `MaximumEntitlement` in the RI: it does not rank +relevance, does not compute NBA, and does not trust anything the LLM +declared (README.md §6.9, Axiom 2). +""" + +from __future__ import annotations + +import hashlib + +from pydantic import BaseModel + +from emcp_bus.common.models import RiskTier +from emcp_bus.entitlement.models import ClientProfile +from emcp_bus.identity.models import ClientRegistration +from emcp_bus.pdp.client import PDPClient + + +class UnknownClientError(Exception): + """`client_id` has no registration, or its registration points at an unknown + profile. Callers (the Gateway) must treat this identically to DENY -- + never as "grant the default/broadest profile".""" + + +class ResolvedEntitlement(BaseModel): + client_id: str + profile_name: str + entitlement_version: str + """Changes whenever the client profile OR the PDP policies change (EP-04-T04) -- + a Gateway cache keyed partly on this value self-invalidates on either (EP-05-T03).""" + allowed_tools: frozenset[str] + risk_ceiling: RiskTier + provenance: str + """Which rule produced this entitlement -- trivial in the RI (one client + profile, no additional dynamic rule composition yet), but the field + exists so a richer Entitlement Manager can report it without a contract + change (README.md §6.9, "manter provenance").""" + + model_config = {"frozen": True} + + +class EntitlementManager: + def __init__(self, pdp_client: PDPClient) -> None: + self._pdp = pdp_client + self._registrations: dict[str, ClientRegistration] = {} + self._profiles: dict[str, ClientProfile] = {} + self._cache: dict[str, ResolvedEntitlement] = {} + + def load_registrations(self, registrations: list[ClientRegistration]) -> None: + self._registrations = {r.client_id: r for r in registrations} + + def load_profiles(self, profiles: list[ClientProfile]) -> None: + self._profiles = {p.name: p for p in profiles} + + async def sync_profiles_to_pdp(self) -> None: + """Push every loaded client profile into the PDP's data document. + + Must run (and complete) before the Gateway starts accepting + `tools/list`/`tools/call` -- a profile the PDP has never seen + resolves every tool to DENY, which is the correct fail-closed + behavior (README.md §38), never a reason to skip the sync. + """ + for profile in self._profiles.values(): + # `exclude_none`: Rego's `not X` only negates an *undefined* or + # literal-`false` X -- a pushed `null` is a defined value, so + # `not profile.restrictions.max_transaction_value` would NOT + # treat an explicit `null` as "no restriction" the way it treats + # a genuinely absent key. Dropping None fields keeps "unset" + # meaning "undefined in Rego" (see config/policies/transaction.rego). + await self._pdp.push_client_profile( + profile.name, + profile.model_dump(mode="json", exclude={"name"}, exclude_none=True), + ) + + def resolve(self, client_id: str) -> ResolvedEntitlement: + """Resolve and cache `MaximumEntitlement(client_id)`. + + Cache lookups are keyed on `client_id` only, but `entitlement_version` + inside the cached value is computed from the *current* profile + content and PDP policy_version every time -- so a stale cache entry + is always self-describing (a caller comparing versions can tell it's + stale) even though this method does not itself re-fetch policy state + from OPA on every call (see `revoke_client`/`revoke_profile` for the + actual invalidation triggers, EP-04-T03). + """ + registration = self._registrations.get(client_id) + if registration is None: + raise UnknownClientError(f"No ClientRegistration for client_id={client_id!r}") + + profile = self._profiles.get(registration.profile) + if profile is None: + raise UnknownClientError( + f"ClientRegistration for {client_id!r} references unknown profile " + f"{registration.profile!r}" + ) + + resolved = ResolvedEntitlement( + client_id=client_id, + profile_name=profile.name, + entitlement_version=self._compute_version(profile), + allowed_tools=frozenset(profile.allowed_tools), + risk_ceiling=profile.risk_ceiling, + provenance=f"client_profile:{profile.name}", + ) + self._cache[client_id] = resolved + return resolved + + def cached(self, client_id: str) -> ResolvedEntitlement | None: + return self._cache.get(client_id) + + def revoke_client(self, client_id: str) -> None: + """Immediately invalidate one client's entitlement (RF-08). + + The *next* `resolve()` for this client_id will raise + `UnknownClientError` until/unless it is re-registered -- there is no + grace period and no fallback to a cached value. + """ + self._registrations.pop(client_id, None) + self._cache.pop(client_id, None) + + def revoke_profile(self, profile_name: str) -> None: + """Invalidate the cache for every client currently on `profile_name`. + + Used when a profile itself changes (not just one client) -- + `policy_changed` in the audit taxonomy (README.md §27). Does not + remove the profile or its registrations; the next `resolve()` call + recomputes against whatever profile is loaded now. + """ + stale_client_ids = [ + client_id + for client_id, resolved in self._cache.items() + if resolved.profile_name == profile_name + ] + for client_id in stale_client_ids: + del self._cache[client_id] + + def _compute_version(self, profile: ClientProfile) -> str: + digest = hashlib.sha256() + digest.update(self._pdp.policy_version.encode("utf-8")) + digest.update(profile.model_dump_json().encode("utf-8")) + return digest.hexdigest()[:12] diff --git a/RI/src/emcp_bus/entitlement/models.py b/RI/src/emcp_bus/entitlement/models.py new file mode 100644 index 0000000..7e03089 --- /dev/null +++ b/RI/src/emcp_bus/entitlement/models.py @@ -0,0 +1,83 @@ +"""Client profile schema (EP-04-T01, README.md §41). + +A client profile is the governed object that defines a MCP Client's +`MaximumEntitlement` ceiling (§7, §17, ADR-008: granularity = domain + risk +tier). It is *not* itself the entitlement decision -- that is resolved by +the Entitlement Manager (EP-04-T02) combining this profile with the PDP's +entitlement policy (EP-03-T02). A profile only ever narrows what a client +may be granted; nothing here can widen it at runtime (Axiom 4). +""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field, field_validator + +from emcp_bus.common.models import RiskTier + + +class TransactionRestrictions(BaseModel): + """Argument/resource-level bounds enforced by the Transaction Policy (EP-03-T03). + + These are *hints* carried alongside the profile for the PDP's Rego data + document -- the PDP, not this model, is the authority that actually + enforces them (README.md §6.5.B, §25 Passo 2). + """ + + max_transaction_value: int | None = Field(default=None, ge=0) + allowed_regions: list[str] | None = None + approval_threshold: int | None = Field( + default=None, + ge=0, + description=( + "Above this `arguments.amount`, the transaction policy returns " + "REQUIRE_APPROVAL instead of ALLOW (README.md §26 example: " + "payment.execute < 10k ALLOW, >= 10k REQUIRE_APPROVAL)." + ), + ) + + +class CredentialPolicy(BaseModel): + workload_identity: bool = True + short_lived: bool = True + + +class AuditPolicy(BaseModel): + level: Literal["none", "standard", "full"] = "full" + + +class ClientProfile(BaseModel): + """A governed client profile (README.md §41 example).""" + + name: str = Field(description='e.g. "Sales.Read", "Finance.Payments" (README.md §7.1).') + owner: str + environment: Literal["dev", "ci", "production"] = "production" + risk_ceiling: RiskTier + allowed_tools: list[str] = Field( + min_length=1, + description="Canonical . tool names (README.md §39).", + ) + restrictions: TransactionRestrictions = Field(default_factory=TransactionRestrictions) + credential_policy: CredentialPolicy = Field(default_factory=CredentialPolicy) + audit: AuditPolicy = Field(default_factory=AuditPolicy) + + @field_validator("name") + @classmethod + def _name_is_domain_dot_qualifier(cls, value: str) -> str: + if "." not in value: + raise ValueError( + f"client profile name {value!r} must be '.' " + "(e.g. 'Sales.Read') per README.md §7.1/§39" + ) + return value + + @field_validator("allowed_tools") + @classmethod + def _tools_are_domain_dot_capability(cls, tools: list[str]) -> list[str]: + for tool in tools: + if "." not in tool: + raise ValueError( + f"tool name {tool!r} must be '.' per README.md §39" + ) + return tools diff --git a/RI/src/emcp_bus/gateway/__init__.py b/RI/src/emcp_bus/gateway/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/gateway/server.py b/RI/src/emcp_bus/gateway/server.py new file mode 100644 index 0000000..7ea51ce --- /dev/null +++ b/RI/src/emcp_bus/gateway/server.py @@ -0,0 +1,260 @@ +"""MCP Gateway / PEP (EP-05-T01/T02/T04). + +Enforces the two invariants of README.md §9/§10: +- `tools/list` returns the intersection of the downstream catalog and the + authenticated client's MaximumEntitlement (filtered discovery, RF-03). +- `tools/call` is re-authorized independently of whatever `tools/list` + returned -- the PDP is consulted on every call, never skipped because a + tool was (or wasn't) visible in a prior discovery response (RF-04, Axiom 5). + +No identity, no discovery, no execution (README.md §5): a request with no +valid access token gets an empty tool list and every `tools/call` denied -- +this is the correct behavior when EP-01-T01 (Keycloak or any OIDC IdP) is +not configured for a given deployment, not a bug to work around. + +Fail-closed (README.md §38, EP-05-T06): a PDP that cannot be reached is a +DENY, not an ALLOW -- see the `PDPUnavailableError` branch in `on_call_tool`. + +Configuration is read from the environment lazily, inside `lifespan()`/ +`build_auth()`, rather than into module-level constants at import time: +every RI service is one process per deployment in production, so this is +purely about correctness (env vars can differ across a Starlette app's +lifespan runs within one Python process, which the test suite relies on to +exercise multiple configurations without re-importing this module). + +Tracing is automatic (see `common/otel.py`): the SDK's `OpenTelemetryMiddleware` +correlates every hop without any manual span code here. +""" + +from __future__ import annotations + +import os +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager +from dataclasses import dataclass +from pathlib import Path + +import mcp_types as types +from mcp.client.session import ClientSession +from mcp.client.streamable_http import streamable_http_client +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import TokenVerifier +from mcp.server.auth.settings import AuthSettings +from mcp.server.context import ServerRequestContext +from mcp.server.lowlevel import Server +from pydantic import AnyHttpUrl + +from emcp_bus.common.config import load_yaml_models +from emcp_bus.common.otel import setup_tracing +from emcp_bus.entitlement.manager import EntitlementManager, UnknownClientError +from emcp_bus.entitlement.models import ClientProfile +from emcp_bus.identity.authn import TokenValidator +from emcp_bus.identity.models import ClientRegistration +from emcp_bus.identity.token_verifier import MCPTokenVerifier +from emcp_bus.pdp.client import PDPClient, PDPUnavailableError, compute_policy_version +from emcp_bus.pdp.models import PDPOutcome + +setup_tracing("emcp-gateway") + +_REPO_ROOT = Path(__file__).resolve().parents[3] + + +@dataclass(frozen=True) +class GatewayConfig: + downstream_url: str + opa_url: str + config_dir: Path + oidc_jwks_url: str + oidc_issuer: str + oidc_audience: str + + @classmethod + def from_env(cls) -> GatewayConfig: + return cls( + downstream_url=os.environ.get("SALES_DOMAIN_URL", "http://127.0.0.1:8100/mcp"), + opa_url=os.environ.get("OPA_URL", "http://127.0.0.1:8181"), + config_dir=Path(os.environ.get("EMCP_CONFIG_DIR", str(_REPO_ROOT / "config"))), + oidc_jwks_url=os.environ.get("OIDC_JWKS_URL", ""), + oidc_issuer=os.environ.get("OIDC_ISSUER", ""), + oidc_audience=os.environ.get("OIDC_AUDIENCE", "emcp-gateway"), + ) + + @property + def clients_dir(self) -> Path: + return self.config_dir / "clients" + + @property + def profiles_dir(self) -> Path: + return self.clients_dir / "profiles" + + @property + def policies_dir(self) -> Path: + return self.config_dir / "policies" + + +@dataclass +class GatewayState: + entitlement_manager: EntitlementManager + pdp: PDPClient + downstream_url: str + + +@asynccontextmanager +async def lifespan(server: Server[GatewayState]) -> AsyncIterator[GatewayState]: + config = GatewayConfig.from_env() + pdp = PDPClient( + base_url=config.opa_url, policy_version=compute_policy_version(config.policies_dir) + ) + manager = EntitlementManager(pdp) + manager.load_registrations(load_yaml_models(config.clients_dir, ClientRegistration)) + manager.load_profiles(load_yaml_models(config.profiles_dir, ClientProfile)) + # Deliberately not try/excepted: a PDP the Gateway cannot sync entitlements + # to at startup must not come up looking healthy (README.md §38). + await manager.sync_profiles_to_pdp() + try: + yield GatewayState( + entitlement_manager=manager, pdp=pdp, downstream_url=config.downstream_url + ) + finally: + await pdp.aclose() + + +async def _with_downstream_session[ResultT]( + downstream_url: str, + coro_factory: Callable[[ClientSession], Awaitable[ResultT]], +) -> ResultT: + """Open a fresh downstream connection for one request and run `coro_factory(session)`. + + A connection per call is simpler and safer than a shared long-lived + session under concurrent requests; the Enterprise MCP Fabric (EP-06) is + where connection pooling/reuse per backend would be introduced later. + """ + async with ( + streamable_http_client(downstream_url) as (read_stream, write_stream), + ClientSession(read_stream, write_stream) as session, + ): + await session.initialize() + return await coro_factory(session) + + +def _denied(reason_code: str) -> types.CallToolResult: + """A deterministic denial the calling agent can observe and react to + (README.md §10/§14: "Operação bloqueada") -- a structured tool-result, + not a raised protocol-level error, matching real MCP tool-failure shape. + """ + return types.CallToolResult( + is_error=True, + content=[types.TextContent(type="text", text=f"Access denied: {reason_code}")], + ) + + +async def on_list_tools( + ctx: ServerRequestContext[GatewayState], + params: types.PaginatedRequestParams | None, +) -> types.ListToolsResult: + access_token = get_access_token() + if access_token is None: + return types.ListToolsResult(tools=[]) + + state = ctx.lifespan_context + try: + resolved = state.entitlement_manager.resolve(access_token.client_id) + except UnknownClientError: + return types.ListToolsResult(tools=[]) + + downstream_result = await _with_downstream_session( + state.downstream_url, lambda session: session.list_tools(params=params) + ) + filtered_tools = [ + tool for tool in downstream_result.tools if tool.name in resolved.allowed_tools + ] + return downstream_result.model_copy(update={"tools": filtered_tools}) + + +async def on_call_tool( + ctx: ServerRequestContext[GatewayState], + params: types.CallToolRequestParams, +) -> types.CallToolResult: + access_token = get_access_token() + if access_token is None: + return _denied("UNAUTHENTICATED") + + state = ctx.lifespan_context + try: + resolved = state.entitlement_manager.resolve(access_token.client_id) + except UnknownClientError: + return _denied("UNKNOWN_CLIENT") + + try: + decision = await state.pdp.evaluate( + client_profile=resolved.profile_name, + tool=params.name, + arguments=dict(params.arguments or {}), + ) + except PDPUnavailableError: + return _denied("PDP_UNAVAILABLE") + + if decision.outcome is PDPOutcome.DENY: + return _denied(decision.reason_code) + if decision.outcome is PDPOutcome.REQUIRE_APPROVAL: + # EP-14 (human approval workflow) does not exist yet in this milestone + # -- fail closed rather than silently upgrading to ALLOW. + return _denied(f"REQUIRE_APPROVAL:{decision.reason_code}") + + result = await _with_downstream_session( + state.downstream_url, lambda session: session.call_tool(params.name, params.arguments or {}) + ) + if isinstance(result, types.CallToolResult): + return result + raise TypeError(f"Unexpected downstream result type for tools/call: {type(result)!r}") + + +server = Server( + "emcp-gateway", + version="0.1.0", + instructions="Enterprise MCP Service Bus Gateway / PEP.", + lifespan=lifespan, + on_list_tools=on_list_tools, + on_call_tool=on_call_tool, +) + + +def build_auth( + config: GatewayConfig | None = None, +) -> tuple[AuthSettings | None, TokenVerifier | None]: + """Wires up OIDC bearer-token auth when EP-01-T01 (Keycloak or any OIDC + IdP) is configured; otherwise both are `None` and every request is + treated as unauthenticated (deny by default, see module docstring) -- a + deliberate, safe default, not a placeholder to silence. + """ + config = config or GatewayConfig.from_env() + if not (config.oidc_jwks_url and config.oidc_issuer): + return None, None + + token_validator = TokenValidator( + jwks_url=config.oidc_jwks_url, issuer=config.oidc_issuer, audience=config.oidc_audience + ) + token_verifier: TokenVerifier = MCPTokenVerifier(token_validator, resource=config.oidc_audience) + # `TokenValidator` already checks `aud` against `oidc_audience` itself + # (README.md §45 Teste 8), so the SDK's own RFC 8707 resource check is + # left off here rather than requiring `oidc_audience` to also be a URL. + auth_settings = AuthSettings( + issuer_url=AnyHttpUrl(config.oidc_issuer), + resource_server_url=AnyHttpUrl(f"https://emcp-bus.internal/{config.oidc_audience}"), + validate_token_resource=False, + ) + return auth_settings, token_verifier + + +def main() -> None: + import uvicorn + + host = os.environ.get("GATEWAY_HOST", "127.0.0.1") + port = int(os.environ.get("GATEWAY_PORT", "8000")) + auth_settings, token_verifier = build_auth() + app = server.streamable_http_app(host=host, auth=auth_settings, token_verifier=token_verifier) + uvicorn.run(app, host=host, port=port, log_level="info") + + +if __name__ == "__main__": + main() diff --git a/RI/src/emcp_bus/identity/__init__.py b/RI/src/emcp_bus/identity/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/identity/authn.py b/RI/src/emcp_bus/identity/authn.py new file mode 100644 index 0000000..83f8a8d --- /dev/null +++ b/RI/src/emcp_bus/identity/authn.py @@ -0,0 +1,73 @@ +"""OAuth2/OIDC access token validation (EP-01-T03, README.md §5, §19, §20). + +Validates a bearer token against the Identity Provider's JWKS (Keycloak, +EP-01-T01, in production; any standards-compliant OIDC IdP works, since this +speaks the standard, not a Keycloak-specific protocol) -- signature, +expiry, issuer and audience. The verified `azp`/`client_id` claim is the +*only* source of client identity this module trusts; nothing about +`clientInfo` or any other agent-declared field is ever consulted here +(README.md §5, "Observação crítica sobre clientInfo", Axiom 2). +""" + +from __future__ import annotations + +import jwt +from jwt import PyJWKClient +from pydantic import BaseModel + + +class AuthenticationError(Exception): + """Token missing/malformed/expired/wrong audience-issuer/bad signature. + + Callers (the Gateway, EP-05-T02/T04) must treat this identically to + DENY -- fail closed (README.md §38), never fall back to an unauthenticated + or agent-declared identity. + """ + + +class AuthenticatedClient(BaseModel): + client_id: str + """From the token's `azp` (OIDC 'authorized party') or `client_id` claim -- + never from clientInfo or any other unauthenticated, agent-controlled field.""" + + issuer: str + expires_at: int + + +class TokenValidator: + def __init__( + self, + jwks_url: str, + issuer: str, + audience: str, + leeway_seconds: int = 5, + ) -> None: + self._jwks_client = PyJWKClient(jwks_url) + self._issuer = issuer + self._audience = audience + self._leeway_seconds = leeway_seconds + + def validate(self, token: str) -> AuthenticatedClient: + try: + signing_key = self._jwks_client.get_signing_key_from_jwt(token) + claims = jwt.decode( + token, + signing_key.key, + algorithms=["RS256"], + audience=self._audience, + issuer=self._issuer, + leeway=self._leeway_seconds, + options={"require": ["exp", "iss", "aud"]}, + ) + except jwt.PyJWTError as exc: + raise AuthenticationError(str(exc)) from exc + + client_id = claims.get("azp") or claims.get("client_id") + if not isinstance(client_id, str) or not client_id: + raise AuthenticationError("Token has neither a usable 'azp' nor 'client_id' claim") + + return AuthenticatedClient( + client_id=client_id, + issuer=str(claims["iss"]), + expires_at=int(claims["exp"]), + ) diff --git a/RI/src/emcp_bus/identity/models.py b/RI/src/emcp_bus/identity/models.py new file mode 100644 index 0000000..2dff905 --- /dev/null +++ b/RI/src/emcp_bus/identity/models.py @@ -0,0 +1,31 @@ +"""MCP Client registration schema (EP-01-T02, README.md §6.3, §19). + +Maps an authenticated OAuth2 client identity (the `client_id`/`azp` claim a +validated access token carries -- see EP-01-T03) to exactly one client +profile (EP-04). This indirection is what lets `docs/RI-PLANNING.md`'s §18 +rule ("share a client only if all sharing agents can legitimately carry the +same ceiling") be expressed as data instead of code: two client_ids pointing +at the same profile are an explicit, auditable sharing decision. +""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel + + +class ClientRegistration(BaseModel): + client_id: str + """OAuth2 client_id as issued by the Identity Provider (Keycloak, EP-01-T01). + + This is the identity the Gateway trusts (from a verified token's `azp` + claim, EP-01-T03) -- never a value read from the MCP request body or + from `clientInfo` (README.md §5, "Observação crítica sobre clientInfo"). + """ + + profile: str + """Name of the ClientProfile (EP-04) this client_id resolves to.""" + + environment: Literal["dev", "ci", "production"] = "production" + description: str | None = None diff --git a/RI/src/emcp_bus/identity/token_verifier.py b/RI/src/emcp_bus/identity/token_verifier.py new file mode 100644 index 0000000..a7e825a --- /dev/null +++ b/RI/src/emcp_bus/identity/token_verifier.py @@ -0,0 +1,38 @@ +"""Adapts `TokenValidator` (EP-01-T03) to the `mcp` SDK's own `TokenVerifier` +protocol, so the Gateway can use the SDK's built-in bearer-auth middleware +(`AuthenticationMiddleware` + `AuthContextMiddleware`) instead of parsing the +`Authorization` header by hand. +""" + +from __future__ import annotations + +import asyncio + +from mcp.server.auth.provider import AccessToken + +from emcp_bus.identity.authn import AuthenticationError, TokenValidator + + +class MCPTokenVerifier: + def __init__(self, token_validator: TokenValidator, resource: str) -> None: + self._token_validator = token_validator + self._resource = resource + + async def verify_token(self, token: str) -> AccessToken | None: + """Returns `None` on any validation failure -- the SDK's bearer-auth + middleware treats that as "no authenticated user", never as an + exception to swallow-and-allow. `TokenValidator.validate` does a + blocking JWKS fetch/signature check (PyJWT), so it runs off the + event loop thread.""" + try: + authenticated = await asyncio.to_thread(self._token_validator.validate, token) + except AuthenticationError: + return None + + return AccessToken( + token=token, + client_id=authenticated.client_id, + scopes=[], + expires_at=authenticated.expires_at, + resource=self._resource, + ) diff --git a/RI/src/emcp_bus/pdp/__init__.py b/RI/src/emcp_bus/pdp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/pdp/client.py b/RI/src/emcp_bus/pdp/client.py new file mode 100644 index 0000000..f34326d --- /dev/null +++ b/RI/src/emcp_bus/pdp/client.py @@ -0,0 +1,148 @@ +"""PDP client (EP-03-T01/T04): wraps OPA's REST API behind the two-step +entitlement + transaction model (README.md §25). + +Fail-closed by construction (README.md §38, RNF-Disponibilidade): every +method that talks to OPA either returns a `PDPDecision` or raises +`PDPUnavailableError` -- there is no code path that turns "OPA didn't +answer" into an implicit ALLOW. Callers (the Gateway, EP-05-T06) are +expected to treat `PDPUnavailableError` as DENY. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +from typing import Any + +import httpx + +from emcp_bus.pdp.models import PDPDecision, PDPOutcome + + +class PDPUnavailableError(Exception): + """OPA could not be reached or returned an unexpected response. + + Deliberately a distinct type from a `PDPDecision(outcome=DENY)`: an + unreachable PDP is an *availability* incident (README.md §38), not a + policy decision, and the two must never be logged/audited identically. + """ + + +def compute_policy_version(policy_dir: str | Path) -> str: + """Deterministic version id for the Rego policies currently on disk. + + A SHA-256 over the sorted `*.rego` (non-test) file contents under + `policy_dir`. This is a pragmatic stand-in for real OPA bundle revision + tracking (which needs a bundle server) -- sufficient for the RI to prove + `policy_version` flows end-to-end into every PDP decision and audit + event (EP-08-T02/T03); replacing it with bundle revisions is a natural + M2+ extension, not a contract change. + """ + digest = hashlib.sha256() + policy_files = sorted( + p for p in Path(policy_dir).glob("*.rego") if not p.name.endswith("_test.rego") + ) + for path in policy_files: + digest.update(path.name.encode("utf-8")) + digest.update(path.read_bytes()) + return digest.hexdigest()[:12] + + +class PDPClient: + def __init__(self, base_url: str, policy_version: str, timeout_seconds: float = 2.0) -> None: + self._http = httpx.AsyncClient(base_url=base_url, timeout=timeout_seconds) + self._policy_version = policy_version + + @property + def policy_version(self) -> str: + return self._policy_version + + async def aclose(self) -> None: + await self._http.aclose() + + async def push_client_profile(self, profile_name: str, profile_data: dict[str, Any]) -> None: + """Push a validated ClientProfile (EP-04-T01) into OPA's data document. + + Called by the Entitlement Manager (EP-04-T02) at startup/reload -- + this client never reads client profile YAML itself, only the + already-validated dict a caller hands it. + """ + try: + response = await self._http.put( + f"/v1/data/emcp/client_profiles/{profile_name}", json=profile_data + ) + response.raise_for_status() + except httpx.HTTPError as exc: + raise PDPUnavailableError( + f"Failed to push client profile {profile_name!r}: {exc}" + ) from exc + + async def evaluate( + self, + *, + client_profile: str, + tool: str, + arguments: dict[str, Any] | None = None, + customer_classification: str | None = None, + ) -> PDPDecision: + """Run the two-step model (README.md §25) and return one combined decision. + + Step 1 (entitlement.rego) is authoritative and short-circuits: a + tool absent from entitlement is DENY before any argument is ever + evaluated (Axiom 5 -- discovery/execution filtering, not argument + policy, is the first line of defense). + """ + entitlement_input = {"client_profile": client_profile, "tool": tool} + entitled = await self._query_bool("/v1/data/emcp/entitlement/allow", entitlement_input) + if not entitled: + return PDPDecision( + outcome=PDPOutcome.DENY, + policy_version=self._policy_version, + reason_code="NOT_IN_ENTITLEMENT", + ) + + transaction_input = { + "client_profile": client_profile, + "tool": tool, + "arguments": arguments or {}, + "customer_classification": customer_classification, + } + transaction_result = await self._query("/v1/data/emcp/transaction", transaction_input) + if not transaction_result.get("allow", False): + return PDPDecision( + outcome=PDPOutcome.DENY, + policy_version=self._policy_version, + reason_code="TRANSACTION_POLICY_DENIED", + ) + if transaction_result.get("require_approval", False): + return PDPDecision( + outcome=PDPOutcome.REQUIRE_APPROVAL, + policy_version=self._policy_version, + reason_code="APPROVAL_THRESHOLD_EXCEEDED", + ) + return PDPDecision( + outcome=PDPOutcome.ALLOW, + policy_version=self._policy_version, + reason_code="ALLOWED", + ) + + async def _query(self, path: str, input_document: dict[str, Any]) -> dict[str, Any]: + try: + response = await self._http.post(path, json={"input": input_document}) + response.raise_for_status() + body: dict[str, Any] = response.json() + except httpx.HTTPError as exc: + raise PDPUnavailableError(f"PDP query to {path} failed: {exc}") from exc + result = body.get("result") + if not isinstance(result, dict): + raise PDPUnavailableError(f"PDP query to {path} returned an unexpected shape: {body!r}") + return result + + async def _query_bool(self, path: str, input_document: dict[str, Any]) -> bool: + try: + response = await self._http.post(path, json={"input": input_document}) + response.raise_for_status() + body: dict[str, Any] = response.json() + except httpx.HTTPError as exc: + raise PDPUnavailableError(f"PDP query to {path} failed: {exc}") from exc + return bool(body.get("result", False)) diff --git a/RI/src/emcp_bus/pdp/models.py b/RI/src/emcp_bus/pdp/models.py new file mode 100644 index 0000000..8fb3143 --- /dev/null +++ b/RI/src/emcp_bus/pdp/models.py @@ -0,0 +1,28 @@ +"""PDP decision contract (EP-03-T04). + +Every response from the PDP is one of these three outcomes, never a bare +boolean -- `policy_version` and `reason_code` are mandatory so the Gateway's +audit trail (EP-08-T02) and end-to-end correlation (EP-08-T03, RF-17) always +have something to point at, and so a REQUIRE_APPROVAL is never collapsed +into ALLOW/DENY by a caller that only checks truthiness. +""" + +from __future__ import annotations + +from enum import StrEnum + +from pydantic import BaseModel + + +class PDPOutcome(StrEnum): + ALLOW = "ALLOW" + DENY = "DENY" + REQUIRE_APPROVAL = "REQUIRE_APPROVAL" + + +class PDPDecision(BaseModel): + outcome: PDPOutcome + policy_version: str + reason_code: str + """Machine-readable reason, e.g. NOT_IN_ENTITLEMENT, TRANSACTION_POLICY_DENIED, + APPROVAL_THRESHOLD_EXCEEDED, ALLOWED, PDP_UNAVAILABLE (README.md §27).""" diff --git a/RI/src/emcp_bus/registry/__init__.py b/RI/src/emcp_bus/registry/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/src/emcp_bus/registry/models.py b/RI/src/emcp_bus/registry/models.py new file mode 100644 index 0000000..3766bff --- /dev/null +++ b/RI/src/emcp_bus/registry/models.py @@ -0,0 +1,104 @@ +"""Capability manifest schema (EP-02-T01, README.md §6.8, §23, §39, §40). + +Reconciles the two illustrative shapes in README.md: the publishing-pipeline +manifest example (§6.8, camelCase, `riskTier: medium`) and the Tool Registry +metadata example (§23, snake_case, `risk_tier: R2`). This RI standardizes on +the §23 field names and the `RiskTier` R0-R4 enum, because that is the same +enum a `ClientProfile.risk_ceiling` uses (EP-04-T01) -- one risk vocabulary, +not two, is what lets entitlement resolution compare a tool's risk against a +client's ceiling directly (see `emcp_bus.common.models.risk_tier_at_or_below`). +""" + +from __future__ import annotations + +from datetime import date +from enum import StrEnum +from typing import Literal + +from pydantic import BaseModel, Field, ValidationInfo, field_validator + +from emcp_bus.common.models import RiskTier + + +class DataClassification(StrEnum): + PUBLIC = "public" + INTERNAL = "internal" + CONFIDENTIAL = "confidential" + RESTRICTED = "restricted" + + +class LifecycleStatus(StrEnum): + """Registry governance lifecycle (README.md §23).""" + + NOMINATE = "nominate" + REVIEW = "review" + REGISTER = "register" + ACTIVE = "active" + DEPRECATED = "deprecated" + RETIRED = "retired" + + +class BackendRef(BaseModel): + """Where the Fabric (EP-06-T01) routes a capability once authorized.""" + + type: Literal["rest", "mcp", "grpc"] = "mcp" + service: str = Field( + description="Adapter/service id the Fabric router resolves, e.g. 'sales-domain'." + ) + operation: str | None = Field( + default=None, description="Informational, e.g. 'GET /v2/customers/{id}' (README.md §6.8)." + ) + + +class ApprovalPolicy(BaseModel): + required: bool = False + + +class ObservabilityPolicy(BaseModel): + audit_level: Literal["none", "standard", "full"] = "full" + + +class LifecycleInfo(BaseModel): + status: LifecycleStatus = LifecycleStatus.NOMINATE + sunset_after: date | None = None + + +class CapabilityManifest(BaseModel): + name: str = Field(description="Canonical '.' name (README.md §39).") + version: str = Field(description="SemVer, e.g. '1.2.0' (README.md §40).") + domain: str + owner: str + risk_tier: RiskTier + side_effects: bool + idempotent: bool + data_classification: list[DataClassification] = Field( + default_factory=lambda: [DataClassification.INTERNAL] + ) + entitlements: list[str] = Field( + min_length=1, + description="ClientProfile names required to see/call this capability (EP-04).", + ) + backend: BackendRef + approval: ApprovalPolicy = Field(default_factory=ApprovalPolicy) + observability: ObservabilityPolicy = Field(default_factory=ObservabilityPolicy) + lifecycle: LifecycleInfo = Field(default_factory=LifecycleInfo) + + @field_validator("name") + @classmethod + def _name_is_canonical(cls, value: str) -> str: + if "." not in value: + raise ValueError( + f"capability name {value!r} must be '.' (README.md §39)" + ) + return value + + @field_validator("domain") + @classmethod + def _domain_matches_name_prefix(cls, value: str, info: ValidationInfo) -> str: + name = info.data.get("name") + if isinstance(name, str) and not name.startswith(f"{value}."): + raise ValueError( + f"domain {value!r} does not match the prefix of name {name!r} " + "(README.md §39: name must be '.')" + ) + return value diff --git a/RI/tests/__init__.py b/RI/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/tests/e2e/__init__.py b/RI/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/tests/e2e/conftest.py b/RI/tests/e2e/conftest.py new file mode 100644 index 0000000..aa93569 --- /dev/null +++ b/RI/tests/e2e/conftest.py @@ -0,0 +1,273 @@ +"""Shared e2e fixtures: a real OPA process, a real JWKS HTTP endpoint + RS256 +token signing, and helpers to run the Gateway/Sales servers in-process +against them. No Keycloak available in this environment (no Java) -- see +RI/README.md for exactly what that does and does not validate. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import socket +import subprocess +import threading +import time +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +import jwt +import pytest +import uvicorn +from cryptography.hazmat.primitives.asymmetric import rsa +from jwt.algorithms import RSAAlgorithm +from mcp.client.session import ClientSession +from mcp.client.streamable_http import ( # type: ignore[attr-defined] + create_mcp_http_client, + streamable_http_client, +) +from mcp.shared.exceptions import MCPError + +REPO_ROOT = Path(__file__).parents[2] +POLICIES_DIR = REPO_ROOT / "config" / "policies" + +ISSUER = "https://keycloak.example/realms/emcp" +AUDIENCE = "emcp-gateway" +KID = "test-key-1" + +GATEWAY_PORT = 18000 +SALES_PORT = 18100 + + +def free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return int(s.getsockname()[1]) + + +# --- Real OPA process (EP-03) -------------------------------------------- + + +@pytest.fixture(scope="session") +def opa_url() -> Iterator[str]: + opa_binary = shutil.which(os.environ.get("OPA_BINARY", "opa")) + if opa_binary is None: + pytest.skip( + "opa binary not found on PATH (set OPA_BINARY or install it -- see RI/README.md)" + ) + + port = free_port() + process = subprocess.Popen( + [opa_binary, "run", "--server", "--addr", f"127.0.0.1:{port}", str(POLICIES_DIR)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + base_url = f"http://127.0.0.1:{port}" + try: + import httpx + + for _ in range(100): + try: + httpx.get(f"{base_url}/health", timeout=1) + break + except httpx.HTTPError: + time.sleep(0.05) + else: + raise RuntimeError("opa server did not become healthy in time") + yield base_url + finally: + process.terminate() + process.wait(timeout=5) + + +# --- Real JWKS HTTP endpoint + RS256 signing (EP-01-T03) ------------------ + + +class _JWKSHandler(BaseHTTPRequestHandler): + jwks_document: dict[str, object] = {} + + def do_GET(self) -> None: # noqa: N802 + body = json.dumps(self.jwks_document).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + pass + + +@pytest.fixture(scope="session") +def rsa_keypair() -> tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return private_key, private_key.public_key() + + +@pytest.fixture(scope="session") +def jwks_url(rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]) -> Iterator[str]: + _, public_key = rsa_keypair + jwk = RSAAlgorithm.to_jwk(public_key, as_dict=True) + jwk.update({"kid": KID, "use": "sig", "alg": "RS256"}) + _JWKSHandler.jwks_document = {"keys": [jwk]} + + server = HTTPServer(("127.0.0.1", 0), _JWKSHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}/jwks" + finally: + server.shutdown() + thread.join(timeout=5) + + +@pytest.fixture +def sign_token( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], +) -> Callable[..., str]: + private_key, _ = rsa_keypair + + def _sign( + client_id: str, + *, + issuer: str = ISSUER, + audience: str = AUDIENCE, + expires_in_seconds: int = 300, + signing_key: rsa.RSAPrivateKey | None = None, + ) -> str: + now = int(time.time()) + claims = { + "iss": issuer, + "aud": audience, + "sub": f"service-account-{client_id}", + "azp": client_id, + "iat": now, + "exp": now + expires_in_seconds, + } + return jwt.encode( + claims, signing_key or private_key, algorithm="RS256", headers={"kid": KID} + ) + + return _sign + + +# --- Running Gateway + Sales servers, wired to the fixtures above -------- + + +class RunningServer: + def __init__(self, config: uvicorn.Config) -> None: + self.server = uvicorn.Server(config) + self._task: asyncio.Task[None] | None = None + + async def start(self) -> None: + self._task = asyncio.create_task(self.server.serve()) + while not self.server.started: + await asyncio.sleep(0.01) + # See test_walking_skeleton.py's original note: give the ASGI app's + # own lifespan (session manager task group) one extra tick. + await asyncio.sleep(0.05) + + async def stop(self) -> None: + self.server.should_exit = True + assert self._task is not None + await self._task + + +@pytest.fixture +async def governed_stack( + monkeypatch: pytest.MonkeyPatch, opa_url: str, jwks_url: str +) -> AsyncIterator[None]: + """Real OPA + real JWKS-backed OIDC auth, wired into fresh Gateway/Sales + instances for one test (EP-01/EP-03/EP-04/EP-05 all real, only Keycloak + itself is substituted by the JWKS server -- see RI/README.md).""" + monkeypatch.setenv("SALES_DOMAIN_URL", f"http://127.0.0.1:{SALES_PORT}/mcp") + monkeypatch.setenv("OPA_URL", opa_url) + monkeypatch.setenv("OIDC_JWKS_URL", jwks_url) + monkeypatch.setenv("OIDC_ISSUER", ISSUER) + monkeypatch.setenv("OIDC_AUDIENCE", AUDIENCE) + + import example_mcp_servers.sales_domain.server as sales_module + + sales_config = uvicorn.Config( + sales_module.server.streamable_http_app(host="127.0.0.1"), + host="127.0.0.1", + port=SALES_PORT, + log_level="warning", + ) + sales_server = RunningServer(sales_config) + await sales_server.start() + + import emcp_bus.gateway.server as gateway_module + + auth_settings, token_verifier = gateway_module.build_auth() + gateway_config = uvicorn.Config( + gateway_module.server.streamable_http_app( + host="127.0.0.1", auth=auth_settings, token_verifier=token_verifier + ), + host="127.0.0.1", + port=GATEWAY_PORT, + log_level="warning", + ) + gateway_server = RunningServer(gateway_config) + await gateway_server.start() + + try: + yield + finally: + await gateway_server.stop() + await sales_server.stop() + + +async def run_against_gateway[ResultT]( + operation: Callable[[ClientSession], Awaitable[ResultT]], + *, + token: str | None = None, +) -> ResultT: + """Connect to the gateway (optionally authenticated) and run `operation(session)`. + + Retries on `MCPError` ("SSE stream ended without a response" has been + observed, rarely, under this in-process dual-uvicorn test setup under + scheduler contention) -- a test-harness robustness concern, not a + Gateway defect: production deployments run one process per service. + """ + headers = {"Authorization": f"Bearer {token}"} if token else None + url = f"http://127.0.0.1:{GATEWAY_PORT}/mcp" + last_error: MCPError | None = None + for attempt in range(3): + if attempt: + await asyncio.sleep(0.1 * attempt) + try: + async with ( + create_mcp_http_client(headers=headers) as http_client, + streamable_http_client(url, http_client=http_client) as (read, write), + ClientSession(read, write) as session, + ): + await session.initialize() + return await operation(session) + except MCPError as exc: + last_error = exc + except BaseExceptionGroup as group: + # Each nested anyio TaskGroup (`streamable_http_client`'s, then + # `ClientSession`'s) wraps the one below it in its own + # ExceptionGroup, so a handler-raised MCPError (e.g. an auth + # rejection during `initialize`) can arrive doubly-wrapped. + # Unwrap fully so callers can `pytest.raises(MCPError)` + # uniformly, same as for one raised directly. + unwrapped = _unwrap_single_mcp_error(group) + if unwrapped is not None: + last_error = unwrapped + else: + raise + assert last_error is not None + raise last_error + + +def _unwrap_single_mcp_error(exc: BaseException) -> MCPError | None: + """If `exc` is an MCPError, or a (possibly nested) ExceptionGroup wrapping + exactly one, return it; otherwise `None`.""" + if isinstance(exc, MCPError): + return exc + if isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1: + return _unwrap_single_mcp_error(exc.exceptions[0]) + return None diff --git a/RI/tests/e2e/test_governed_gateway.py b/RI/tests/e2e/test_governed_gateway.py new file mode 100644 index 0000000..a25d0ea --- /dev/null +++ b/RI/tests/e2e/test_governed_gateway.py @@ -0,0 +1,242 @@ +"""EP-05-T02/T04 governed Gateway: the actual security invariants of +README.md §45 ("Critérios de aceite de segurança"), run against the real +chain -- real OIDC token validation (JWKS server standing in for Keycloak), +real OPA-backed PDP, real EntitlementManager loading the real config/ +YAML files. +""" + +from __future__ import annotations + +from collections.abc import Callable + +import pytest +from mcp.shared.exceptions import MCPError + +from tests.e2e.conftest import run_against_gateway + +SALES_READ_CLIENT_ID = "sales-read-agent" +SALES_WRITE_CLIENT_ID = "sales-write-agent" +FINANCE_PAYMENTS_CLIENT_ID = "finance-payments-agent" + + +async def test_teste1_unauthorized_discovery( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §45 Teste 1: finance.createPayment MUST NOT be returned to Sales.Read.""" + token = sign_token(SALES_READ_CLIENT_ID) + + result = await run_against_gateway(lambda session: session.list_tools(), token=token) + + tool_names = {tool.name for tool in result.tools} + assert "finance.payment.execute" not in tool_names + assert "sales.quote.create" not in tool_names # Sales.Write-only, not Sales.Read + + +async def test_teste2_direct_unauthorized_call( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §45 Teste 2: calling a known-but-unentitled tool directly, without + ever having seen it in tools/list, must still DENY.""" + token = sign_token(SALES_READ_CLIENT_ID) + + result = await run_against_gateway( + lambda session: session.call_tool("finance.payment.execute", {"amount": 1_000_000}), + token=token, + ) + + assert result.is_error is True + assert "Access denied" in _text(result) + + +async def test_teste4_agent_role_spoofing_has_no_effect( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §45 Teste 4: nothing the caller declares in tool arguments + (an agent-shaped 'role' claim in this case) can expand entitlement.""" + token = sign_token(SALES_READ_CLIENT_ID) + + result = await run_against_gateway( + lambda session: session.call_tool( + "finance.payment.execute", + {"amount": 100, "role": "finance-admin", "client_profile": "Finance.Payments"}, + ), + token=token, + ) + + assert result.is_error is True + + +async def test_unauthenticated_request_sees_no_tools_and_is_denied( + governed_stack: None, +) -> None: + """README.md §5: 'No entitlement -> no discovery -> no execution'. + + With OIDC configured, the SDK's own `RequireAuthMiddleware` rejects a + connection with no bearer token before the MCP session handshake + (`initialize`) ever completes -- stricter than this Gateway's own + handler-level `access_token is None` checks (`on_list_tools`/ + `on_call_tool`), which exist as the deny-by-default backstop for the + *other* supported deployment mode: no OIDC configured at all (see + `GatewayConfig`/`build_auth` in emcp_bus/gateway/server.py), where every + request reaches the handlers unauthenticated and those checks are what + enforces "no discovery, no execution" instead. + """ + with pytest.raises(MCPError): + await run_against_gateway(lambda session: session.list_tools()) + + +async def test_expired_token_is_denied( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §45 Teste 3 (adjacent)/§46: expired token, adversarial scenario.""" + token = sign_token(SALES_READ_CLIENT_ID, expires_in_seconds=-3600) + + with pytest.raises(MCPError): + await run_against_gateway(lambda session: session.list_tools(), token=token) + + +async def test_wrong_audience_token_is_denied( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §45 Teste 8: a token issued for a different resource server.""" + token = sign_token(SALES_READ_CLIENT_ID, audience="some-other-service") + + with pytest.raises(MCPError): + await run_against_gateway(lambda session: session.list_tools(), token=token) + + +async def test_forged_signature_token_is_denied( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §46: stolen/forged token, signed by a key not in the JWKS.""" + from cryptography.hazmat.primitives.asymmetric import rsa + + forged_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + token = sign_token(SALES_READ_CLIENT_ID, signing_key=forged_key) + + with pytest.raises(MCPError): + await run_against_gateway(lambda session: session.list_tools(), token=token) + + +async def test_transaction_policy_denies_over_limit_even_though_entitled( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §25 Passo 2: sales.quote.create IS in Sales.Write's entitlement, + but the transaction policy still denies an amount above the profile's limit.""" + token = sign_token(SALES_WRITE_CLIENT_ID) + + result = await run_against_gateway( + lambda session: session.call_tool( + "sales.quote.create", {"amount": 999_999_999, "region": "BR-SP"} + ), + token=token, + ) + + assert result.is_error is True + assert "TRANSACTION_POLICY_DENIED" in _text(result) + + +async def test_require_approval_is_denied_pending_ep14( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """README.md §26: Finance.Payments allows payment.execute below the approval + threshold (README.md §7.2/§17 profile) but must never silently upgrade a + REQUIRE_APPROVAL to ALLOW -- and EP-14 (approval workflow) does not exist + yet in this milestone, so the only safe outcome is DENY.""" + token = sign_token(FINANCE_PAYMENTS_CLIENT_ID) + + result = await run_against_gateway( + lambda session: session.call_tool("finance.payment.execute", {"amount": 50_000}), + token=token, + ) + + assert result.is_error is True + assert "REQUIRE_APPROVAL" in _text(result) + + +async def test_gateway_lifespan_fails_closed_when_pdp_unreachable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """README.md §38: PDP indisponível -> fail closed, never ALLOW-by-omission. + + A Gateway that cannot sync entitlements to its PDP at startup must not + come up looking healthy (EP-05-T06): its `lifespan` must raise, not swallow + the error and yield a state built on an empty/stale entitlement sync. + """ + from tests.e2e.conftest import free_port + + monkeypatch.setenv("OPA_URL", f"http://127.0.0.1:{free_port()}") # nothing listening here + + import emcp_bus.gateway.server as gateway_module + from emcp_bus.pdp.client import PDPUnavailableError + + with pytest.raises(PDPUnavailableError): + async with gateway_module.lifespan(gateway_module.server): + pytest.fail("lifespan must not yield a state when the PDP sync failed") + + +async def test_deny_by_default_when_no_oidc_provider_is_configured( + monkeypatch: pytest.MonkeyPatch, opa_url: str +) -> None: + """README.md §5: the *other* supported deployment mode (no OIDC provider + configured at all -- `OIDC_JWKS_URL`/`OIDC_ISSUER` unset). No + `RequireAuthMiddleware` is installed in this mode (see `build_auth`), so + every request reaches the handlers with `get_access_token() is None`, + and it is the Gateway's own handler-level checks that must deny by + default -- this is the code path Teste 1-style discovery filtering and + Teste 2-style execution denial fall back to without an IdP. + """ + import uvicorn + + from tests.e2e.conftest import GATEWAY_PORT, SALES_PORT, RunningServer, run_against_gateway + + monkeypatch.setenv("SALES_DOMAIN_URL", f"http://127.0.0.1:{SALES_PORT}/mcp") + monkeypatch.setenv("OPA_URL", opa_url) + monkeypatch.delenv("OIDC_JWKS_URL", raising=False) + monkeypatch.delenv("OIDC_ISSUER", raising=False) + + import example_mcp_servers.sales_domain.server as sales_module + + sales_server = RunningServer( + uvicorn.Config( + sales_module.server.streamable_http_app(host="127.0.0.1"), + host="127.0.0.1", + port=SALES_PORT, + log_level="warning", + ) + ) + await sales_server.start() + + import emcp_bus.gateway.server as gateway_module + + auth_settings, token_verifier = gateway_module.build_auth() + assert (auth_settings, token_verifier) == (None, None) # confirms the mode under test + + gateway_server = RunningServer( + uvicorn.Config( + gateway_module.server.streamable_http_app( + host="127.0.0.1", auth=auth_settings, token_verifier=token_verifier + ), + host="127.0.0.1", + port=GATEWAY_PORT, + log_level="warning", + ) + ) + await gateway_server.start() + try: + list_result = await run_against_gateway(lambda session: session.list_tools()) + assert list_result.tools == [] + + call_result = await run_against_gateway( + lambda session: session.call_tool("sales.customer.get", {"customer_id": "cust-001"}) + ) + assert call_result.is_error is True + assert "UNAUTHENTICATED" in _text(call_result) + finally: + await gateway_server.stop() + await sales_server.stop() + + +def _text(result: object) -> str: + content = getattr(result, "content", []) + return " ".join(getattr(item, "text", "") for item in content) diff --git a/RI/tests/e2e/test_walking_skeleton.py b/RI/tests/e2e/test_walking_skeleton.py new file mode 100644 index 0000000..850db61 --- /dev/null +++ b/RI/tests/e2e/test_walking_skeleton.py @@ -0,0 +1,105 @@ +"""M0 walking skeleton, now exercised through M1's real enforcement chain +(docs/RI-PLANNING.md, Marco M0 + M1: EP-01/EP-03/EP-04/EP-05). + +Client -> Gateway (real OIDC auth + real OPA-backed entitlement) -> (static +routing) -> example Sales MCP server -> response, over real HTTP +(streamable-http), with a single OTel trace_id correlating spans across both +processes. Authenticates as `sales-read-agent` (Sales.Read, README.md §7.1) +calling only tools within its own entitlement -- the denial paths (no token, +wrong tool, expired/forged token, PDP down) are tests/e2e/test_governed_gateway.py. +""" + +from __future__ import annotations + +from collections.abc import Callable + +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import SpanKind + +from tests.e2e.conftest import run_against_gateway + +SALES_READ_CLIENT_ID = "sales-read-agent" + + +async def test_tools_list_is_filtered_to_sales_read_entitlement( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + token = sign_token(SALES_READ_CLIENT_ID) + + result = await run_against_gateway(lambda session: session.list_tools(), token=token) + + # README.md §7.1: Sales.Read sees exactly these four -- not + # sales.quote.create/update (Sales.Write-only) and not any Finance tool. + assert {tool.name for tool in result.tools} == { + "sales.customer.search", + "sales.customer.get", + "sales.order.get", + "sales.inventory.check", + } + + +async def test_tools_call_reaches_the_backend_and_returns_its_result( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + token = sign_token(SALES_READ_CLIENT_ID) + + result = await run_against_gateway( + lambda session: session.call_tool("sales.customer.get", {"customer_id": "cust-001"}), + token=token, + ) + + assert result.is_error is not True + assert result.structured_content == { + "customer_id": "cust-001", + "name": "Aurora Ltda", + "region": "BR-SP", + } + + +async def test_a_single_trace_id_correlates_gateway_and_backend_spans( + governed_stack: None, sign_token: Callable[..., str] +) -> None: + """Both hops of one `tools/call` must land on the same trace_id. + + This test runs both server processes in-process (in-process uvicorn, for + speed), so they share one OTel global TracerProvider/resource -- unlike a + real multi-process deployment (verified manually: each process gets its + own `service.name` resource), so `resource.attributes["service.name"]` + cannot be used here to tell the two hops apart. `SpanKind.SERVER` can: + the gateway receiving the call from the client, and the Sales server + receiving it from the gateway, are each a distinct SERVER-kind + `tools/call ...` span -- exactly two (or more, under a rare transport + retry -- see `run_against_gateway`) of them sharing one trace_id proves + the request was correlated across both hops. + """ + from opentelemetry import trace + + exporter = InMemorySpanExporter() + trace.get_tracer_provider().add_span_processor(SimpleSpanProcessor(exporter)) # type: ignore[attr-defined] + + token = sign_token(SALES_READ_CLIENT_ID) + await run_against_gateway( + lambda session: session.call_tool("sales.customer.get", {"customer_id": "cust-001"}), + token=token, + ) + + spans: tuple[ReadableSpan, ...] = exporter.get_finished_spans() + server_tool_call_spans = [ + span + for span in spans + if span.name.startswith("tools/call") and span.kind == SpanKind.SERVER + ] + trace_ids = {span.context.trace_id for span in server_tool_call_spans} + + assert len(trace_ids) == 1, ( + "Expected exactly one trace_id shared by both hops, got " + f"{[f'{t:#x}' for t in trace_ids]} across spans {[s.name for s in server_tool_call_spans]}" + ) + assert len(server_tool_call_spans) >= 2, ( + "Expected at least one SERVER 'tools/call' span for the gateway hop and one " + f"for the backend hop, got {len(server_tool_call_spans)}: " + f"{[s.name for s in server_tool_call_spans]} -- see Marco M0 acceptance " + "criterion in docs/RI-PLANNING.md" + ) diff --git a/RI/tests/unit/__init__.py b/RI/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/RI/tests/unit/test_authn.py b/RI/tests/unit/test_authn.py new file mode 100644 index 0000000..3874549 --- /dev/null +++ b/RI/tests/unit/test_authn.py @@ -0,0 +1,180 @@ +"""TokenValidator against a real JWKS HTTP endpoint and real RS256-signed +tokens (EP-01-T03) -- no Keycloak available in this environment (no Java), +so a local JWKS server + self-issued tokens exercise the actual OIDC +validation logic (signature, expiry, issuer, audience) that Keycloak would +otherwise be the source of. See RI/README.md for what this does and does +not prove. +""" + +from __future__ import annotations + +import json +import threading +import time +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, HTTPServer + +import jwt +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa +from jwt.algorithms import RSAAlgorithm + +from emcp_bus.identity.authn import AuthenticationError, TokenValidator + +ISSUER = "https://keycloak.example/realms/emcp" +AUDIENCE = "emcp-gateway" +KID = "test-key-1" + + +class _JWKSHandler(BaseHTTPRequestHandler): + jwks_document: dict[str, object] = {} + + def do_GET(self) -> None: # noqa: N802 -- BaseHTTPRequestHandler's naming convention + body = json.dumps(self.jwks_document).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 -- stdlib signature + pass # keep test output quiet + + +@pytest.fixture(scope="module") +def rsa_keypair() -> tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return private_key, private_key.public_key() + + +@pytest.fixture(scope="module") +def jwks_server( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], +) -> Iterator[str]: + _, public_key = rsa_keypair + jwk = RSAAlgorithm.to_jwk(public_key, as_dict=True) + jwk.update({"kid": KID, "use": "sig", "alg": "RS256"}) + _JWKSHandler.jwks_document = {"keys": [jwk]} + + server = HTTPServer(("127.0.0.1", 0), _JWKSHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}/jwks" + finally: + server.shutdown() + thread.join(timeout=5) + + +def _sign_token( + private_key: rsa.RSAPrivateKey, + *, + issuer: str = ISSUER, + audience: str = AUDIENCE, + client_id: str = "sales-read-agent", + expires_in_seconds: int = 300, + client_id_claim: str = "azp", +) -> str: + now = int(time.time()) + claims = { + "iss": issuer, + "aud": audience, + "sub": "service-account-sales-read-agent", + client_id_claim: client_id, + "iat": now, + "exp": now + expires_in_seconds, + } + return jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": KID}) + + +def test_valid_token_resolves_the_authenticated_client_id( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + private_key, _ = rsa_keypair + token = _sign_token(private_key, client_id="sales-write-agent") + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + authenticated = validator.validate(token) + + assert authenticated.client_id == "sales-write-agent" + assert authenticated.issuer == ISSUER + + +def test_client_id_claim_fallback_when_azp_absent( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + private_key, _ = rsa_keypair + token = _sign_token(private_key, client_id="sales-write-agent", client_id_claim="client_id") + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + authenticated = validator.validate(token) + + assert authenticated.client_id == "sales-write-agent" + + +def test_expired_token_is_rejected( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + private_key, _ = rsa_keypair + token = _sign_token(private_key, expires_in_seconds=-3600) + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + with pytest.raises(AuthenticationError): + validator.validate(token) + + +def test_wrong_audience_is_rejected( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + """README.md §45 Teste 8: a token issued for a different resource server must be rejected.""" + private_key, _ = rsa_keypair + token = _sign_token(private_key, audience="some-other-resource-server") + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + with pytest.raises(AuthenticationError): + validator.validate(token) + + +def test_wrong_issuer_is_rejected( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + private_key, _ = rsa_keypair + token = _sign_token(private_key, issuer="https://attacker.example/realms/fake") + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + with pytest.raises(AuthenticationError): + validator.validate(token) + + +def test_token_signed_by_an_untrusted_key_is_rejected(jwks_server: str) -> None: + """A different keypair than the one in the JWKS -- e.g. a stolen/forged token.""" + forged_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + token = _sign_token(forged_key) + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + with pytest.raises(AuthenticationError): + validator.validate(token) + + +def test_agent_declared_role_claim_is_never_consulted( + rsa_keypair: tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey], jwks_server: str +) -> None: + """README.md §35.1 anti-pattern / Axiom 2: a bonus claim on an otherwise + valid token must have zero effect on the resolved client_id.""" + private_key, _ = rsa_keypair + now = int(time.time()) + claims = { + "iss": ISSUER, + "aud": AUDIENCE, + "sub": "service-account-sales-read-agent", + "azp": "sales-read-agent", + "role": "finance-admin", # attacker-controlled-shaped claim, never trusted + "iat": now, + "exp": now + 300, + } + token = jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": KID}) + + validator = TokenValidator(jwks_url=jwks_server, issuer=ISSUER, audience=AUDIENCE) + authenticated = validator.validate(token) + + assert authenticated.client_id == "sales-read-agent" + assert not hasattr(authenticated, "role") diff --git a/RI/tests/unit/test_entitlement_manager.py b/RI/tests/unit/test_entitlement_manager.py new file mode 100644 index 0000000..e849d47 --- /dev/null +++ b/RI/tests/unit/test_entitlement_manager.py @@ -0,0 +1,165 @@ +"""EntitlementManager against real OPA (EP-04-T02/T03/T04), loading the +actual YAML client profiles/registrations checked into config/ -- not +hand-built fixtures, so a schema or naming mistake in the YAML itself would +fail these tests too. +""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import socket +import subprocess +from collections.abc import AsyncIterator +from pathlib import Path + +import pytest + +from emcp_bus.common.config import load_yaml_model, load_yaml_models +from emcp_bus.common.models import RiskTier +from emcp_bus.entitlement.manager import EntitlementManager, UnknownClientError +from emcp_bus.entitlement.models import ClientProfile +from emcp_bus.identity.models import ClientRegistration +from emcp_bus.pdp.client import PDPClient, compute_policy_version + +REPO_ROOT = Path(__file__).parents[2] +POLICIES_DIR = REPO_ROOT / "config" / "policies" +PROFILES_DIR = REPO_ROOT / "config" / "clients" / "profiles" +CLIENTS_DIR = REPO_ROOT / "config" / "clients" + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return int(s.getsockname()[1]) + + +@pytest.fixture +async def entitlement_manager() -> AsyncIterator[EntitlementManager]: + opa_binary = shutil.which(os.environ.get("OPA_BINARY", "opa")) + if opa_binary is None: + pytest.skip( + "opa binary not found on PATH (set OPA_BINARY or install it -- see RI/README.md)" + ) + + port = _free_port() + process = subprocess.Popen( + [opa_binary, "run", "--server", "--addr", f"127.0.0.1:{port}", str(POLICIES_DIR)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + pdp = PDPClient( + base_url=f"http://127.0.0.1:{port}", + policy_version=compute_policy_version(POLICIES_DIR), + ) + try: + for _ in range(100): + try: + await pdp._http.get("/health") + break + except Exception: # noqa: BLE001 + await asyncio.sleep(0.05) + else: + raise RuntimeError("opa server did not become healthy in time") + + manager = EntitlementManager(pdp) + manager.load_registrations( + [ + load_yaml_model(path, ClientRegistration) + for path in sorted(CLIENTS_DIR.glob("*.yaml")) + ] + ) + manager.load_profiles(load_yaml_models(PROFILES_DIR, ClientProfile)) + await manager.sync_profiles_to_pdp() + yield manager + finally: + await pdp.aclose() + process.terminate() + process.wait(timeout=5) + + +async def test_resolves_sales_read_exactly_as_readme_7_1( + entitlement_manager: EntitlementManager, +) -> None: + resolved = entitlement_manager.resolve("sales-read-agent") + + assert resolved.profile_name == "Sales.Read" + assert resolved.risk_ceiling is RiskTier.R1 + assert resolved.allowed_tools == frozenset( + { + "sales.customer.search", + "sales.customer.get", + "sales.order.get", + "sales.inventory.check", + } + ) + assert resolved.entitlement_version + + +async def test_sales_write_does_not_include_approve_tools( + entitlement_manager: EntitlementManager, +) -> None: + resolved = entitlement_manager.resolve("sales-write-agent") + + assert "sales.quote.create" in resolved.allowed_tools + assert "sales.discount.approve" not in resolved.allowed_tools + assert "finance.createPayment" not in resolved.allowed_tools + + +async def test_unknown_client_id_raises() -> None: + pdp = PDPClient(base_url="http://127.0.0.1:1", policy_version="dev") + manager = EntitlementManager(pdp) + try: + with pytest.raises(UnknownClientError): + manager.resolve("no-such-client") + finally: + await pdp.aclose() + + +async def test_revoke_client_forces_re_resolution_to_fail( + entitlement_manager: EntitlementManager, +) -> None: + entitlement_manager.resolve("sales-read-agent") # populate cache + assert entitlement_manager.cached("sales-read-agent") is not None + + entitlement_manager.revoke_client("sales-read-agent") + + assert entitlement_manager.cached("sales-read-agent") is None + with pytest.raises(UnknownClientError): + entitlement_manager.resolve("sales-read-agent") + + +async def test_revoke_profile_invalidates_cache_for_all_its_clients( + entitlement_manager: EntitlementManager, +) -> None: + entitlement_manager.resolve("sales-read-agent") + assert entitlement_manager.cached("sales-read-agent") is not None + + entitlement_manager.revoke_profile("Sales.Read") + + assert entitlement_manager.cached("sales-read-agent") is None + # Registration itself is untouched -- the *next* resolve() recomputes + # cleanly against whatever profile is currently loaded (RF-08). + resolved_again = entitlement_manager.resolve("sales-read-agent") + assert resolved_again.profile_name == "Sales.Read" + + +async def test_entitlement_version_changes_when_profile_content_changes( + entitlement_manager: EntitlementManager, +) -> None: + before = entitlement_manager.resolve("sales-read-agent").entitlement_version + + mutated = ClientProfile.model_validate( + { + "name": "Sales.Read", + "owner": "sales-platform", + "risk_ceiling": "R1", + "allowed_tools": ["sales.customer.get", "sales.order.get", "sales.inventory.check"], + } + ) + entitlement_manager.load_profiles([mutated]) + + after = entitlement_manager.resolve("sales-read-agent").entitlement_version + + assert before != after diff --git a/RI/tests/unit/test_pdp_client.py b/RI/tests/unit/test_pdp_client.py new file mode 100644 index 0000000..083ff85 --- /dev/null +++ b/RI/tests/unit/test_pdp_client.py @@ -0,0 +1,154 @@ +"""PDPClient against a *real* OPA process (EP-03-T01/T04) -- not a mock. + +Requires the `opa` binary on PATH (or `OPA_BINARY` pointing at it); skips +otherwise. See RI/README.md for install instructions -- OPA ships as a +single static binary, no Docker required. +""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import socket +import subprocess +from collections.abc import AsyncIterator +from pathlib import Path + +import pytest + +from emcp_bus.pdp.client import PDPClient, PDPUnavailableError, compute_policy_version +from emcp_bus.pdp.models import PDPOutcome + +POLICIES_DIR = Path(__file__).parents[2] / "config" / "policies" + + +def _opa_binary() -> str | None: + return shutil.which(os.environ.get("OPA_BINARY", "opa")) + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return int(s.getsockname()[1]) + + +@pytest.fixture +async def pdp_client() -> AsyncIterator[PDPClient]: + opa_binary = _opa_binary() + if opa_binary is None: + pytest.skip( + "opa binary not found on PATH (set OPA_BINARY or install it -- see RI/README.md)" + ) + + port = _free_port() + process = subprocess.Popen( + [opa_binary, "run", "--server", "--addr", f"127.0.0.1:{port}", str(POLICIES_DIR)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + client = PDPClient( + base_url=f"http://127.0.0.1:{port}", + policy_version=compute_policy_version(POLICIES_DIR), + ) + try: + for _ in range(100): + try: + await client._http.get("/health") + break + except Exception: # noqa: BLE001 -- retry loop, any error means "not ready yet" + await asyncio.sleep(0.05) + else: + raise RuntimeError("opa server did not become healthy in time") + yield client + finally: + await client.aclose() + process.terminate() + process.wait(timeout=5) + + +async def test_policy_version_is_stable_and_nonempty() -> None: + version = compute_policy_version(POLICIES_DIR) + assert version + assert version == compute_policy_version(POLICIES_DIR) + + +async def test_deny_when_tool_not_in_entitlement(pdp_client: PDPClient) -> None: + await pdp_client.push_client_profile("Sales.Read", {"allowed_tools": ["sales.customer.get"]}) + + decision = await pdp_client.evaluate(client_profile="Sales.Read", tool="finance.createPayment") + + assert decision.outcome is PDPOutcome.DENY + assert decision.reason_code == "NOT_IN_ENTITLEMENT" + assert decision.policy_version + + +async def test_allow_when_tool_in_entitlement_and_within_transaction_limits( + pdp_client: PDPClient, +) -> None: + await pdp_client.push_client_profile( + "Sales.Write", + { + "allowed_tools": ["sales.quote.create"], + "restrictions": {"max_transaction_value": 500000, "approval_threshold": 100000}, + }, + ) + + decision = await pdp_client.evaluate( + client_profile="Sales.Write", + tool="sales.quote.create", + arguments={"amount": 50000}, + ) + + assert decision.outcome is PDPOutcome.ALLOW + assert decision.reason_code == "ALLOWED" + + +async def test_deny_when_transaction_exceeds_limit_even_though_entitled( + pdp_client: PDPClient, +) -> None: + await pdp_client.push_client_profile( + "Sales.Write", + { + "allowed_tools": ["sales.quote.create"], + "restrictions": {"max_transaction_value": 500000}, + }, + ) + + decision = await pdp_client.evaluate( + client_profile="Sales.Write", + tool="sales.quote.create", + arguments={"amount": 999999}, + ) + + assert decision.outcome is PDPOutcome.DENY + assert decision.reason_code == "TRANSACTION_POLICY_DENIED" + + +async def test_require_approval_above_threshold(pdp_client: PDPClient) -> None: + await pdp_client.push_client_profile( + "Finance.Payments", + { + "allowed_tools": ["finance.payment.execute"], + "restrictions": {"max_transaction_value": 10_000_000, "approval_threshold": 10_000}, + }, + ) + + decision = await pdp_client.evaluate( + client_profile="Finance.Payments", + tool="finance.payment.execute", + arguments={"amount": 15_000}, + ) + + assert decision.outcome is PDPOutcome.REQUIRE_APPROVAL + assert decision.reason_code == "APPROVAL_THRESHOLD_EXCEEDED" + + +async def test_fails_closed_when_opa_is_unreachable() -> None: + unreachable_port = _free_port() # nothing listening here + client = PDPClient(base_url=f"http://127.0.0.1:{unreachable_port}", policy_version="dev") + try: + with pytest.raises(PDPUnavailableError): + await client.evaluate(client_profile="Sales.Read", tool="sales.customer.get") + finally: + await client.aclose() diff --git a/RI/tests/unit/test_schemas.py b/RI/tests/unit/test_schemas.py new file mode 100644 index 0000000..2bb801d --- /dev/null +++ b/RI/tests/unit/test_schemas.py @@ -0,0 +1,90 @@ +"""Sanity checks for the M1 config schemas (EP-01-T02, EP-02-T01, EP-04-T01).""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from emcp_bus.common.models import RiskTier, risk_tier_at_or_below +from emcp_bus.entitlement.models import ClientProfile +from emcp_bus.identity.models import ClientRegistration +from emcp_bus.registry.models import BackendRef, CapabilityManifest + + +def test_risk_tier_ordering() -> None: + assert risk_tier_at_or_below(RiskTier.R1, RiskTier.R2) + assert not risk_tier_at_or_below(RiskTier.R3, RiskTier.R2) + assert risk_tier_at_or_below(RiskTier.R0, RiskTier.R0) + + +def test_client_profile_from_readme_example() -> None: + profile = ClientProfile.model_validate( + { + "name": "Sales.Write", + "owner": "sales-platform", + "environment": "production", + "risk_ceiling": "R2", + "allowed_tools": [ + "sales.customer.get", + "sales.order.get", + "sales.quote.create", + "sales.quote.update", + ], + "restrictions": {"max_transaction_value": 500000}, + "credential_policy": {"workload_identity": True, "short_lived": True}, + "audit": {"level": "full"}, + } + ) + assert profile.risk_ceiling is RiskTier.R2 + assert "sales.quote.create" in profile.allowed_tools + + +def test_client_profile_rejects_flat_name() -> None: + with pytest.raises(ValidationError): + ClientProfile.model_validate( + { + "name": "SalesWrite", # missing "." + "owner": "sales-platform", + "risk_ceiling": "R2", + "allowed_tools": ["sales.quote.create"], + } + ) + + +def test_client_registration_references_a_profile_by_name() -> None: + registration = ClientRegistration.model_validate( + {"client_id": "sales-write-agent", "profile": "Sales.Write"} + ) + assert registration.profile == "Sales.Write" + + +def test_capability_manifest_domain_must_prefix_name() -> None: + manifest = CapabilityManifest.model_validate( + { + "name": "sales.quote.create", + "version": "2.1.0", + "domain": "sales", + "owner": "sales-platform", + "risk_tier": "R2", + "side_effects": True, + "idempotent": False, + "entitlements": ["Sales.Write", "Sales.Approve"], + "backend": {"type": "mcp", "service": "sales-domain"}, + } + ) + assert manifest.backend == BackendRef(type="mcp", service="sales-domain") + + with pytest.raises(ValidationError): + CapabilityManifest.model_validate( + { + "name": "sales.quote.create", + "version": "2.1.0", + "domain": "finance", # mismatched prefix + "owner": "sales-platform", + "risk_tier": "R2", + "side_effects": True, + "idempotent": False, + "entitlements": ["Sales.Write"], + "backend": {"type": "mcp", "service": "sales-domain"}, + } + )