Metadata-Version: 2.4
Name: koda-mcp-access
Version: 1.0.1
Summary: Reusable FastMCP access-control middleware (PEP) + identity helpers for KODA-ecosystem MCP servers.
Project-URL: Homepage, https://bitbucket.org/kushki/koda-mcp-access
Project-URL: Repository, https://bitbucket.org/kushki/koda-mcp-access
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-access

Reusable [FastMCP](https://github.com/jlowin/fastmcp) **access-control** middleware
for KODA-ecosystem MCP servers, plus identity helpers for tools. It is a pure
**Policy Enforcement Point (PEP)**: it reads identity + an already-resolved
permission subset from proxy-injected headers and allows / denies / filters
MCP tool calls 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 + KodaAccessMiddleware
                          (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-access
# or with the optional standalone Okta verifier (local-dev):
uv add "koda-mcp-access[okta]"
# or: pip install koda-mcp-access
```

Pin it in your project as:

```toml
# pyproject.toml
koda-mcp-access>=1.0,<2
```

## Use

```python
from fastmcp import FastMCP
from koda_mcp_access import KodaAccessMiddleware

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

With an audit hook for observability (every allow/deny decision):

```python
from koda_mcp_access import GuardConfig, KodaAccessMiddleware

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

mcp.add_middleware(KodaAccessMiddleware(GuardConfig(audit_hook=audit)))
```

> **Only tools are enforced.** The middleware filters `tools/list` and gates
> `tools/call` against the caller's `tools` grant. Skills are authorized
> upstream in the registry, not here, so this package does not gate `prompts/*`
> or skill names. Resources pass through (see [Behavior](#behavior)).

Read identity from inside a tool:

```python
from koda_mcp_access 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):

```python
from koda_mcp_access 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(KodaAccessMiddleware())` and **no local `src/auth/`**. If a tool
> needs identity, import it from this package. See
> [`docs/MCP_ACCESS_STANDARD.md`](docs/MCP_ACCESS_STANDARD.md).

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

```python
from koda_mcp_access.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 — the server-scoped `tools` plus the
global catalogs `skills` / `agents`:

```json
{ "tools": ["my_profile"],
  "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`, `resources/list`, `resources/templates/list` | — | passthrough (not role-gated), still fail-closed without identity |
| any | `None` (stdio / discovery) | passthrough (no filter) |
| any | header missing/invalid + not stdio/discovery | `AuthorizationError` (fail-closed) |

> **Only tools are role-gated.** Resources pass through for any authenticated
> caller (the permission model has no resource dimension); the fail-closed check
> still applies (no identity → denied). Skills are authorized upstream in the
> registry, and prompts are not gated — this package does not touch `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 |
| `audit_hook` | `None` | `Callable[[AccessDecision], 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` drives enforcement, `is_skill_allowed` / `is_agent_action_allowed` are exported helpers for consumers |
| `context.py` | contextvar holding the current `Identity` for the request |
| `config.py` | `GuardConfig`, `AccessDecision` |
| `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 → `call_next` → reset
identity (always).

## Develop

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