Metadata-Version: 2.4
Name: koda-mcp-authz-middleware
Version: 0.1.2
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 `permissions` 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.

## How it fits

```
client → registry/nginx → koda-proxy → AgentCore runtime → your MCP + KodaAuthzMiddleware
                          (resolves &                        (this package:
                           injects X-Koda-* )                 reads & enforces)
```

- **Decision** (who are you, what may you do) happens **upstream** in the
  koda-proxy: it calls usrv-koda, resolves the per-server subset, and injects
  it as base64-JSON in the `permissions` header.
- **Enforcement** (allow / deny / filter each call) happens **here**. This
  package never validates the JWT, never queries a DB, never knows roles.

## 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.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 + skill 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(
    # Tools that carry a skill name in their args: the middleware authorizes
    # BOTH the tool (against `tools`) AND the named skill (against `skills`).
    skill_arg_tools={
        "install_skill": "skill_name",
        "get_skill_content": "skill_name",
        "uninstall_skill": "skill_name",
    },
    audit_hook=audit,
)))
```

### Skill authorization

KODA does not expose skills as MCP prompts — they are consumed through **tools**
that take a skill name as an argument (e.g. `install_skill(skill_name=...)`).
So skills are authorized at the tool-call boundary, not via `prompts/*`:

- List `skill_arg_tools` as `{tool_name: arg_name}`.
- On `tools/call`, if the tool is skill-bearing, the middleware reads
  `arguments[arg_name]` and checks it against the caller's `skills` patterns
  (same matcher as tools). The tool grant gets you to the tool; the skill grant
  lets you act on that specific skill.
- If the arg is absent, only the tool grant is checked (nothing to gate).
- Leave `skill_arg_tools` empty (default) to disable skill-arg gating.

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}
```

Tools that call a backend **on the caller's behalf** can get the raw JWT as a
FastMCP `AccessToken` (or a compact user dict) — no local adapter needed
(`>=0.1.2`):

```python
from koda_mcp_authz_middleware import get_current_access_token, get_current_user

@mcp.tool()
async def list_my_initiatives() -> list:
    token = get_current_access_token()        # None if unauthenticated
    return await backend.get("/initiatives", bearer=token.token)
```

> **Standard:** every KODA-ecosystem MCP authorizes with a single
> `add_middleware(KodaAuthzMiddleware())` and **no local `src/auth/`**. If a tool
> needs identity, import it from this package. See
> [`docs/MCP_AUTHZ_STANDARD.md`](docs/MCP_AUTHZ_STANDARD.md).

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/call` (skill-bearing) | `tools` **and** named skill in `skills` | execute / `AuthorizationError` |
| `tools/list` | `tools` | filtered list |
| `resources/read` | per `resource_gate` | serve / `AuthorizationError` |
| `resources/list` | per `resource_gate` | filtered list |
| `resources/templates/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`.

> **Prompts are not enforced.** KODA does not use MCP prompts; skill
> authorization runs through `skill_arg_tools` (see above), not `prompts/*`.

### Pattern matching

Every grant list is matched the same way (`:` and `.` are interchangeable
separators). An empty / missing list always **denies** (fail-closed):

| Pattern | Matches |
|---|---|
| `*` | anything |
| `jira_*` | prefix (`jira_create`, `jira_get`, …) |
| `my_profile` | exact |

`agents` is a map `{name: {actions: [...]}}`; `is_agent_action_allowed` matches
the agent key (exact / `*` / prefix), then the action against that entry's list.

## Configuration (`GuardConfig`)

| Field | Default | Purpose |
|---|---|---|
| `header_prefix` | AgentCore custom prefix | header namespace to read |
| `fail_mode` | `"closed"` | `closed` denies when identity/perms absent; `open` passes through |
| `bypass_transports` | `{"stdio"}` | transports that skip filtering (local dev) |
| `discovery_suffix` | `"discovery"` | header suffix that marks a catalogue scan |
| `resource_gate` | `"by_mcp_access"` | how `resources/*` are gated |
| `skill_arg_tools` | `{}` | `{tool: arg}` map for skill-bearing tools |
| `audit_hook` | `None` | `Callable[[AuthzDecision], None]` for every decision |

## Internals

One file per concern — all role/server-agnostic:

| Module | Responsibility |
|---|---|
| `middleware.py` | the `Middleware` hooks (`on_call_tool`, `on_list_tools`, `on_*_resource*`); binds identity per request, enforces, resets |
| `headers.py` | `HeaderContract` — builds header names from the prefix, reads them, base64-decodes `permissions` |
| `identity.py` | `Identity` dataclass + `identity_from_headers` |
| `permissions.py` | `MCPPermissions` shape + `parse_permissions` (tolerant of malformed input; ignores legacy `prompts`) |
| `matcher.py` | the pure pattern matchers (`is_tool_allowed`, `is_skill_allowed`, `is_resource_allowed`, `is_agent_action_allowed`) |
| `context.py` | contextvar holding the current `Identity` for the request |
| `config.py` | `GuardConfig`, `AuthzDecision` |
| `verifiers/okta.py` | optional standalone Okta JWT verifier (local dev only) |

Request flow inside `on_call_tool`: resolve identity + perms from headers →
bypass if stdio/discovery → deny if tool not granted → if skill-bearing, deny if
named skill not granted → `call_next` → reset identity (always).

## Develop

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