Metadata-Version: 2.4
Name: koda-mcp-authz-middleware
Version: 0.1.0
Summary: Reusable FastMCP authorization middleware (PEP) for KODA-ecosystem MCP servers.
Project-URL: Homepage, https://bitbucket.org/kushki/koda-mcp-authz-middleware
Project-URL: Repository, https://bitbucket.org/kushki/koda-mcp-authz-middleware
Author: Kushki
License-Expression: MIT
License-File: LICENSE
Keywords: authorization,fastmcp,koda,mcp,middleware,rbac
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: fastmcp>=3.2.4
Provides-Extra: okta
Requires-Dist: pyjwt[crypto]>=2.9.0; extra == 'okta'
Description-Content-Type: text/markdown

# koda-mcp-authz-middleware

Reusable [FastMCP](https://github.com/jlowin/fastmcp) **authorization** middleware
for KODA-ecosystem MCP servers. It is a pure **Policy Enforcement Point (PEP)**:
it reads identity + an already-resolved permission subset from proxy-injected
headers and allows / denies / filters every MCP operation by pattern matching.

It does **not** know about roles, servers, Okta, or the `grants` permission
model — that intelligence lives upstream (usrv-koda = data, koda-proxy = subset
resolution). Any MCP behind a proxy that honors the header contract can use it
unchanged.

## Install

```bash
uv add koda-mcp-authz-middleware
# or with the optional standalone Okta verifier (local-dev):
uv add "koda-mcp-authz-middleware[okta]"
# or: pip install koda-mcp-authz-middleware
```

Pin it in your project as:

```toml
# pyproject.toml
koda-mcp-authz-middleware>=0.1,<0.2
```

## Use

```python
from fastmcp import FastMCP
from koda_mcp_authz_middleware import KodaAuthzMiddleware

mcp = FastMCP("My MCP")
mcp.add_middleware(KodaAuthzMiddleware())   # defaults: AgentCore prefix, fail-closed
```

With configuration (observability + prompts enforcement):

```python
from koda_mcp_authz_middleware import GuardConfig, KodaAuthzMiddleware

def audit(d):
    logger.info("authz action=%s target=%s decision=%s roles=%s",
                d.action, d.target, d.decision, d.identity.roles if d.identity else [])

mcp.add_middleware(KodaAuthzMiddleware(GuardConfig(
    enforce_prompts=True,
    audit_hook=audit,
)))
```

Read identity from inside a tool:

```python
from koda_mcp_authz_middleware import get_current_identity

@mcp.tool()
async def whoami() -> dict:
    ident = get_current_identity()
    return {"sub": ident.sub, "roles": ident.roles, "tenant": ident.tenant_id}
```

Standalone local-dev (validate the Okta JWT itself, requires `[okta]` extra):

```python
from koda_mcp_authz_middleware.verifiers.okta import OktaTokenVerifier
from fastmcp.server.auth import RemoteAuthProvider

if IS_LOCAL:
    mcp.auth = RemoteAuthProvider(
        token_verifier=OktaTokenVerifier(),
        authorization_servers=[OKTA_ISSUER],
        base_url=MCP_BASE_URL,
    )
```

## Header contract

The upstream proxy injects, under a configurable prefix (default
`x-amzn-bedrock-agentcore-runtime-custom-koda-`):

| Header suffix | Meaning |
|---|---|
| `sub` | caller subject (required; absent → no identity) |
| `email` | caller email (defaults to `sub`) |
| `roles` | space-separated roles (resolved upstream) |
| `auth-groups` | space-separated Okta groups |
| `tenant-id` | tenant |
| `authorization` | raw Okta JWT (for tools calling backends) |
| `permissions` | base64(JSON) — the **already-resolved subset for this MCP** |
| `discovery` | `"true"` → registry catalogue scan, no filtering |

`permissions` decodes to a flat object — already server-scoped (tools/components)
plus the global catalogs (skills/agents):

```json
{ "tools": ["my_profile"], "components": ["ui://koda/*"],
  "skills": ["pm_challenge"], "agents": { "roadmap-agent": { "actions": ["*"] } } }
```

## Behavior

| Operation | perms | Result |
|---|---|---|
| `tools/call` | match in `tools` | execute / `AuthorizationError` |
| `tools/list` | `tools` | filtered list |
| `resources/read` | per `resource_gate` | serve / `AuthorizationError` |
| `resources/list` | per `resource_gate` | filtered list |
| any | `None` (stdio / discovery) | passthrough (no filter) |
| any | header missing/invalid + not stdio/discovery | `AuthorizationError` (fail-closed) |

`resource_gate`: `by_mcp_access` (default — any MCP access grants resources),
`by_component_list` (strict URI match against `components`), or `open`.

## Develop

```bash
uv sync --group dev
uv run pytest
uv run ruff check src/ tests/
```
