diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..67f7b3f --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,71 @@ +name: ci + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + # The floor the package declares, and the newest supported release. + python-version: ["3.11", "3.13"] + # 3.10 is the declared floor but fastmcp needs 3.11+, so the floor is + # verified by the core-extra-only job below. + steps: + - uses: actions/checkout@v6 + + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + + - name: Install with the fastmcp extra + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[fastmcp]" pytest pytest-asyncio + + - name: Test + run: python -m pytest tests/ -q + + core-extra-only: + # The framework-agnostic core must import and work with no MCP framework + # installed, so `namoid[mcp]` stays usable outside FastMCP. Also runs on the + # declared minimum Python, which fastmcp itself does not support. + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.13"] + steps: + - uses: actions/checkout@v6 + + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + + - name: Install with the mcp extra only + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[mcp]" pytest pytest-asyncio + + - name: Core, Hosted Auth, and modularity tests + run: >- + python -m pytest tests/test_mcp_authorization.py tests/test_hosted_auth.py + tests/test_modularity.py -q + + - name: Assert fastmcp is absent + run: | + python - <<'PY' + import importlib.util, sys + if importlib.util.find_spec("fastmcp") is not None: + sys.exit("fastmcp must not be installed by the mcp extra") + import namoid.mcp + print("core imports without an MCP framework") + PY diff --git a/README.md b/README.md index aca6c81..5db9e0f 100644 --- a/README.md +++ b/README.md @@ -12,12 +12,235 @@ Python SDK for [NamoID](https://namoid.in) — enterprise identity for India (OA pip install namoid ``` -## Usage +Hosted Auth needs only the base install. Protecting an MCP server adds an extra: -See the integration guides and API reference at -[docs.namoid.in](https://docs.namoid.in), and runnable end-to-end examples in -[`namoid-examples`](https://github.com/namoidhq/namoid-examples) -(`python-fastapi`, `python-flask`). +```bash +pip install "namoid[fastmcp]" # protect a FastMCP server (Python 3.11+) +pip install "namoid[mcp]" # the same core, without FastMCP +``` + +Python 3.10 or newer. The FastMCP extra requires 3.11. + +## Pick what you need + +The two surfaces are independent. Take either, both, or neither — you pay only +for what you name. + +| You want | Install | Import | Pulled in | +|---|---|---|---| +| Hosted Auth | `namoid` | `namoid` | `httpx` | +| Protect an MCP server | `namoid[mcp]` | `namoid.mcp` | `httpx`, `joserfc` | +| …on FastMCP | `namoid[fastmcp]` | `namoid.mcp.fastmcp` | the above, `fastmcp` | + +`import namoid` loads neither surface. The top-level names resolve on first use +(PEP 562), so: + +- A Hosted Auth application never imports the MCP code and never needs its extra. +- An MCP server never imports the Hosted Auth client. +- The MCP core (`namoid.mcp`) imports no MCP framework at all — `fastmcp` appears + only inside `namoid.mcp.fastmcp`, so the core also backs the official MCP + Python SDK, a bare Starlette app, or a test. + +`dir(namoid)` and `namoid.__all__` still list the full surface, and type checkers +resolve every export, so laziness costs nothing in editor support. These +boundaries are enforced by tests in `tests/test_modularity.py`, each run in a +fresh interpreter, rather than left as an intention. + +Everything raises `NamoIDError`, including the MCP errors, so one `except` +catches both surfaces without importing the one you do not use. + +## Hosted Auth + +Hosted Auth redirects the user to a branded NamoID sign-in page and returns a +one-time code. The Client ID resolves the application, its environment, and its +Hosted Auth domain, so there is no issuer or application UUID to configure. + +```python +from namoid import NamoIDClient + +namoid = NamoIDClient( + client_id=os.environ["NAMOID_CLIENT_ID"], + client_secret=os.environ["NAMOID_CLIENT_SECRET"], # server-side only +) + +# 1. Start a state-bound transaction and keep the verifier in the user's session. +transaction = namoid.create_transaction() +session["namoid_state"] = transaction.state +session["namoid_verifier"] = transaction.code_verifier + +# 2. Send the browser to the application's own hosted sign-in page. +url = namoid.hosted_auth_url( + return_to="https://app.example/auth/callback", + state=transaction.state, + completion_mode="confidential", + code_challenge=transaction.code_challenge, +) + +# 3. On the callback, compare state, then exchange the code on the server. +tokens = namoid.exchange_code( + code=request.args["code"], + code_verifier=session.pop("namoid_verifier"), +) + +# 4. Confirm the token and create your own application session. +result = namoid.validate_access_token(tokens.access_token) +if not result.valid: + raise Unauthorized() + +# 5. On sign-out, revoke the NamoID session too. +namoid.revoke_session(access_token=tokens.access_token, refresh_token=tokens.refresh_token) +``` + +`AsyncNamoIDClient` has exactly the same methods with `await`, for FastAPI, +Starlette, or any async framework: + +```python +from namoid import AsyncNamoIDClient + +async with AsyncNamoIDClient(client_id=..., client_secret=...) as namoid: + tokens = await namoid.exchange_code(code=code, code_verifier=verifier) +``` + +Both accept an `http_client` if you want to supply your own configured +`httpx.Client` / `httpx.AsyncClient`, and cache the auth config after the first +fetch. + +For a browser-only public client, redirect with `completion_mode="public"` and +exchange with `confidential=False` — PKCE protects the flow and no secret is +involved. Never put a Client Secret anywhere a browser can reach. + +| Method | Endpoint | +|---|---| +| `get_auth_config()` | `GET /v1/auth/config` | +| `hosted_auth_url(...)` | builds the URL, no request | +| `exchange_code(...)` | `POST /v1/auth/hosted/exchange` | +| `refresh(...)` | `POST /v1/auth/refresh` | +| `validate_access_token(...)` | `POST /v1/auth/tokens/validate` | +| `revoke_session(...)` | `POST /v1/auth/logout` | + +Every failure raises `NamoIDError`, carrying `status`, `code` (the API's own +error code when present), and the parsed `detail`. + +`namoid.hosted_auth` exposes the pure pieces — `create_hosted_auth_transaction`, +`build_hosted_auth_url`, `build_configured_hosted_auth_url`, `pkce_challenge`, +`random_base64url` — if you would rather drive the flow yourself. + +## Protect an MCP server + +NamoID is the authorization server. Your MCP server is the protected resource. +An MCP host — Claude, ChatGPT, Cursor, VS Code — is the OAuth client, and the +signed-in human is the resource owner. NamoID authenticates that human, records +consent, and issues a short-lived token limited to your server and to the +actions approved; this package validates and enforces it. + +No NamoID credentials are needed: a resource server only consumes public +discovery metadata and JWKS. + +```python +from fastmcp import FastMCP +from namoid.mcp.fastmcp import create_namoid_auth, current_caller, require_namoid_scopes + +# Console -> Environment -> MCP Authorization -> Integration details. +# Discovery runs here, so a wrong issuer or resource fails at startup rather +# than as an opaque 401 on the first tool call. +auth = create_namoid_auth( + issuer="https://acme-test.id.namoid.in", + resource="https://mcp.acme.example/mcp", # exactly as registered as the audience + resource_name="Acme Finance MCP", + scopes_supported=["customers:read", "invoices:read"], +) + +mcp = FastMCP(name="acme-finance-mcp", auth=auth.provider) + + +@mcp.tool +@require_namoid_scopes(auth, "refunds:create") +def issue_refund(invoice_id: str, amount_minor: int) -> dict: + caller = current_caller(auth) # the NamoID user who consented + assert_refund_allowed(caller.subject, invoice_id, amount_minor) + return refund(invoice_id, amount_minor) + + +if __name__ == "__main__": + # The path comes from the resource URL, so the endpoint matches the audience. + mcp.run(transport="http", path=auth.mcp_path) +``` + +A scope is permission to **attempt** an action. `refunds:create` does not mean +this user may refund another organization's invoice or exceed your refund +policy. Ownership, limits, and every other business rule stay in the handler. + +When a scope is missing, the tool answers with an `insufficient_scope` result +naming the missing scopes, the resource, and the metadata URL — what a host +needs to start incremental authorization. The tool stays visible in +`tools/list`, because hiding it would leave the host unable to ask for access. +Use FastMCP's own `require_scopes` when hiding a capability is the goal. + +### What it validates + +Every token must satisfy all of: + +| Check | Why | +|---|---| +| `RS256` from the environment's JWKS | The only algorithm NamoID issues | +| Exact `iss` | A token from another issuer is not yours | +| Exact `aud` | A token minted for MCP server A must fail on server B | +| `exp` / `nbf`, 30s tolerance | Configurable via `clock_tolerance_seconds` | +| `token_use == "access"` | An ID token must never be an API token | +| `sub` and `client_id` present | Without `sub`, every caller is one identity | + +Discovery also checks that the issuer's metadata declares the issuer you asked +for — RFC 9700 mix-up defence — and reads `jwks_uri` from it rather than +hard-coding a key location. The key set is cached, and refetched when a token +arrives with an unrecognised `kid` so key rotation is picked up without a +restart. + +### Without FastMCP + +`namoid.mcp` imports no MCP framework, so it can back the official MCP Python +SDK, a bare Starlette app, or a test: + +```python +from namoid.mcp import create_namoid_mcp_auth, NamoIDMcpTokenError + +auth = await create_namoid_mcp_auth( + issuer="https://acme-test.id.namoid.in", + resource="https://mcp.acme.example/mcp", + scopes_supported=["invoices:read"], +) + +# Publish auth.protected_resource_metadata at auth.metadata_path (RFC 9728), +# and answer an unauthenticated call with a WWW-Authenticate challenge +# pointing at auth.metadata_url. +try: + caller = await auth.verify_access_token(bearer_token) +except NamoIDMcpTokenError: + ... # 401 + challenge +``` + +`create_namoid_mcp_auth_sync` is the blocking form, for servers built at module +import where there is no event loop to await on. + +### Client onboarding + +NamoID resolves MCP clients by pre-registration or Client ID Metadata Document +(CIMD). Dynamic Client Registration is not available for customer-owned MCP +resources, so a host that can only do DCR cannot connect yet. + +## Examples + +Complete runnable servers, including a TypeScript equivalent: +[namoid-examples/mcp-authorization](https://github.com/namoidhq/namoid-examples/tree/main/mcp-authorization). + +For the rest of the SDK, see the integration guides and API reference at +[docs.namoid.in](https://docs.namoid.in). + +## Develop + +```bash +python -m venv .venv && .venv/bin/pip install -e ".[fastmcp]" pytest pytest-asyncio +.venv/bin/python -m pytest tests/ -q +``` ## Links diff --git a/pyproject.toml b/pyproject.toml index 3b9e9f5..fff4208 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,15 +4,17 @@ build-backend = "hatchling.build" [project] name = "namoid" -version = "0.0.2" +version = "0.1.0" description = "Python SDK for NamoID, enterprise identity for India (OAuth 2.1 / OIDC)." readme = "README.md" -requires-python = ">=3.9" +requires-python = ">=3.10" license = "MIT" license-files = ["LICENSE"] authors = [{ name = "PolyMindsLabs Pvt. Ltd.", email = "hello@namoid.in" }] keywords = [ "namoid", + "mcp", + "model-context-protocol", "authentication", "oauth", "oauth2", @@ -32,10 +34,31 @@ classifiers = [ "Intended Audience :: Developers", "Operating System :: OS Independent", "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Topic :: Security", "Topic :: Software Development :: Libraries", ] +dependencies = [ + "httpx>=0.28", +] + +[project.optional-dependencies] +# Protect an MCP server. Framework-agnostic core: discovery, audience-bound +# token verification, and RFC 9728 metadata. +mcp = [ + "joserfc>=1.0", +] +# The same core wired into FastMCP. `fastmcp` itself requires a newer Python +# than this package's floor, so pip enforces that when the extra is installed. +fastmcp = [ + "joserfc>=1.0", + "fastmcp>=3.4.5,<4", +] + [project.urls] Homepage = "https://namoid.in" Documentation = "https://namoid.in" @@ -44,3 +67,11 @@ Issues = "https://github.com/namoidhq/namoid-python/issues" [tool.hatch.build.targets.wheel] packages = ["src/namoid"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/src/namoid/__init__.py b/src/namoid/__init__.py index 1708faa..f3fb7d4 100644 --- a/src/namoid/__init__.py +++ b/src/namoid/__init__.py @@ -1,8 +1,90 @@ -"""namoid: Python SDK for NamoID, enterprise identity for India (OAuth 2.1 / OIDC).""" +"""namoid: Python SDK for NamoID, enterprise identity for India (OAuth 2.1 / OIDC). + +Two independent surfaces. Take either, both, or neither — nothing is loaded +until you name it. + +**Hosted Auth** — redirect users to a branded NamoID sign-in page and exchange +the returned code for a session. :class:`NamoIDClient` and +:class:`AsyncNamoIDClient` here; pure helpers in :mod:`namoid.hosted_auth`. +Needs only the base install. + +**MCP authorization** — protect an MCP server so NamoID issues short-lived, +audience-bound tokens for it. In :mod:`namoid.mcp` (extra: ``mcp``) with a +FastMCP adapter in :mod:`namoid.mcp.fastmcp` (extra: ``fastmcp``). + +The two never import each other. Importing this package does not import either +one: the names below resolve on first use (PEP 562), so an MCP-only server never +loads the Hosted Auth client, and a Hosted Auth app never loads the MCP code or +needs its extra installed. +""" from __future__ import annotations -__version__ = "0.0.2" +from importlib import import_module +from typing import TYPE_CHECKING, Any + +__version__ = "0.1.0" __homepage__ = "https://namoid.in" -__all__ = ["__homepage__", "__version__"] +# Public name -> the module that defines it. Resolved on first attribute access +# so importing `namoid` stays free of httpx client construction and of any +# optional dependency. +_LAZY_EXPORTS = { + "AsyncNamoIDClient": "namoid._client", + "NamoIDClient": "namoid._client", + "NamoIDError": "namoid._errors", + "AuthConfig": "namoid.hosted_auth", + "HostedAuthTransaction": "namoid.hosted_auth", + "TokenResponse": "namoid.hosted_auth", + "TokenValidation": "namoid.hosted_auth", + "build_configured_hosted_auth_url": "namoid.hosted_auth", + "build_hosted_auth_url": "namoid.hosted_auth", + "create_hosted_auth_transaction": "namoid.hosted_auth", +} + +__all__ = [ + "AsyncNamoIDClient", + "AuthConfig", + "HostedAuthTransaction", + "NamoIDClient", + "NamoIDError", + "TokenResponse", + "TokenValidation", + "__homepage__", + "__version__", + "build_configured_hosted_auth_url", + "build_hosted_auth_url", + "create_hosted_auth_transaction", +] + + +def __getattr__(name: str) -> Any: + module_name = _LAZY_EXPORTS.get(name) + if module_name is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + value = getattr(import_module(module_name), name) + # Cache on the module so later lookups skip this path entirely. + globals()[name] = value + return value + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_LAZY_EXPORTS)) + + +if TYPE_CHECKING: + # Import eagerly for type checkers and IDEs, which do not run __getattr__. + from namoid._client import AsyncNamoIDClient as AsyncNamoIDClient + from namoid._client import NamoIDClient as NamoIDClient + from namoid._errors import NamoIDError as NamoIDError + from namoid.hosted_auth import AuthConfig as AuthConfig + from namoid.hosted_auth import HostedAuthTransaction as HostedAuthTransaction + from namoid.hosted_auth import TokenResponse as TokenResponse + from namoid.hosted_auth import TokenValidation as TokenValidation + from namoid.hosted_auth import ( + build_configured_hosted_auth_url as build_configured_hosted_auth_url, + ) + from namoid.hosted_auth import build_hosted_auth_url as build_hosted_auth_url + from namoid.hosted_auth import ( + create_hosted_auth_transaction as create_hosted_auth_transaction, + ) diff --git a/src/namoid/_client.py b/src/namoid/_client.py new file mode 100644 index 0000000..21916e3 --- /dev/null +++ b/src/namoid/_client.py @@ -0,0 +1,401 @@ +"""Sync and async NamoID clients. + +Both share one definition of every request and one response parser, so the two +flavours cannot drift apart. Only the transport differs. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +import httpx + +from namoid._errors import NamoIDError +from namoid.hosted_auth import ( + DEFAULT_API_BASE_URL, + AuthConfig, + HostedAuthTransaction, + TokenResponse, + TokenValidation, + build_configured_hosted_auth_url, + create_hosted_auth_transaction, +) + +__all__ = ["AsyncNamoIDClient", "NamoIDClient"] + +_DEFAULT_TIMEOUT_SECONDS = 10.0 + + +@dataclass(frozen=True) +class _Call: + """One HTTP call plus how to turn its body into a result.""" + + method: str + path: str + params: Mapping[str, Any] | None = None + json: Mapping[str, Any] | None = None + headers: Mapping[str, str] | None = None + expect_body: bool = True + failure_code: str = "namoid_request_failed" + failure_label: str = "NamoID request" + + +def _config_call(client_id: str) -> _Call: + return _Call( + "GET", + "/v1/auth/config", + params={"client_id": client_id}, + failure_code="auth_config_failed", + failure_label="Auth config request", + ) + + +def _exchange_call( + *, + code: str, + code_verifier: str | None, + client_id: str | None, + client_secret: str | None, + device_id: str | None, +) -> _Call: + return _Call( + "POST", + "/v1/auth/hosted/exchange", + json=_compact( + { + "code": code, + "code_verifier": code_verifier, + "device_id": device_id, + "client_id": client_id, + "client_secret": client_secret, + } + ), + failure_code="hosted_auth_exchange_failed", + failure_label="Hosted Auth exchange", + ) + + +def _refresh_call(refresh_token: str) -> _Call: + return _Call( + "POST", + "/v1/auth/refresh", + json={"refresh_token": refresh_token}, + failure_code="token_refresh_failed", + failure_label="Token refresh", + ) + + +def _validate_call(*, token: str, client_id: str | None, client_secret: str | None) -> _Call: + return _Call( + "POST", + "/v1/auth/tokens/validate", + json=_compact( + {"token": token, "client_id": client_id, "client_secret": client_secret} + ), + failure_code="token_validation_failed", + failure_label="Token validation", + ) + + +def _logout_call(*, access_token: str, refresh_token: str | None) -> _Call: + return _Call( + "POST", + "/v1/auth/logout", + json={"refresh_token": refresh_token}, + headers={"authorization": f"Bearer {access_token}"}, + expect_body=False, + failure_code="session_revocation_failed", + failure_label="Session revocation", + ) + + +class _ClientBase: + def __init__( + self, + *, + client_id: str, + client_secret: str | None = None, + api_base_url: str = DEFAULT_API_BASE_URL, + timeout: float = _DEFAULT_TIMEOUT_SECONDS, + ) -> None: + if not client_id: + raise NamoIDError("client_id is required", code="missing_client_id") + self._client_id = client_id + self._client_secret = client_secret + self._api_base_url = api_base_url.rstrip("/") + self._timeout = timeout + self._config: AuthConfig | None = None + + @property + def client_id(self) -> str: + return self._client_id + + @staticmethod + def create_transaction() -> HostedAuthTransaction: + """Fresh ``state`` and PKCE material for one sign-in attempt.""" + return create_hosted_auth_transaction() + + def _url(self, path: str) -> str: + return f"{self._api_base_url}{path}" + + def _require_secret(self, provided: str | None) -> str: + secret = provided or self._client_secret + if not secret: + raise NamoIDError( + "client_secret is required for this call; never place it in browser code", + code="missing_client_secret", + ) + return secret + + def _hosted_url(self, config: AuthConfig, kwargs: Mapping[str, Any]) -> str: + return build_configured_hosted_auth_url(config, **kwargs) + + +class NamoIDClient(_ClientBase): + """Blocking NamoID client. + + ``client_secret`` is only needed for confidential calls — code exchange with + a server-managed session, and token validation. Never pass it in code that + reaches a browser. + """ + + def __init__(self, *, http_client: httpx.Client | None = None, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._http = http_client + self._owns_http = http_client is None + + def __enter__(self) -> NamoIDClient: + return self + + def __exit__(self, *_exc: Any) -> None: + self.close() + + def close(self) -> None: + """Close the underlying HTTP client, if this instance created it.""" + if self._owns_http and self._http is not None: + self._http.close() + self._http = None + + def get_auth_config(self, *, refresh: bool = False) -> AuthConfig: + """Fetch and cache the application's browser-safe configuration.""" + if self._config is None or refresh: + self._config = AuthConfig.from_payload(self._send(_config_call(self._client_id))) + return self._config + + def hosted_auth_url(self, **kwargs: Any) -> str: + """Build the Hosted Auth URL for this application. + + Accepts the keyword arguments of + :func:`namoid.hosted_auth.build_configured_hosted_auth_url`. + """ + return self._hosted_url(self.get_auth_config(), kwargs) + + def exchange_code( + self, + *, + code: str, + code_verifier: str | None = None, + client_secret: str | None = None, + device_id: str | None = None, + confidential: bool = True, + ) -> TokenResponse: + """Exchange a one-time Hosted Auth code for a NamoID session. + + Args: + confidential: When true (the default) a Client Secret is required, + matching ``completion_mode="confidential"`` on the redirect. Pass + false for the browser-only PKCE flow. + """ + secret = self._require_secret(client_secret) if confidential else None + payload = self._send( + _exchange_call( + code=code, + code_verifier=code_verifier, + client_id=self._client_id, + client_secret=secret, + device_id=device_id, + ) + ) + return TokenResponse.from_payload(payload) + + def refresh(self, refresh_token: str) -> TokenResponse: + """Rotate a refresh token for a new session.""" + return TokenResponse.from_payload(self._send(_refresh_call(refresh_token))) + + def validate_access_token( + self, token: str, *, client_secret: str | None = None + ) -> TokenValidation: + """Validate an access token with this application's credentials. + + Server-side only. For an MCP server, verify tokens locally instead — see + :mod:`namoid.mcp`, which needs no credentials and no round trip. + """ + secret = self._require_secret(client_secret) + payload = self._send( + _validate_call(token=token, client_id=self._client_id, client_secret=secret) + ) + return TokenValidation.from_payload(payload) + + def revoke_session(self, *, access_token: str, refresh_token: str | None = None) -> None: + """Revoke the NamoID session behind ``access_token``. + + Omitting ``refresh_token`` revokes every session for the user. + """ + self._send(_logout_call(access_token=access_token, refresh_token=refresh_token)) + + def _send(self, call: _Call) -> Any: + if self._http is None: + self._http = httpx.Client(timeout=self._timeout) + self._owns_http = True + try: + response = self._http.request( + call.method, + self._url(call.path), + params=dict(call.params or {}) or None, + json=dict(call.json) if call.json is not None else None, + headers={"accept": "application/json", **(call.headers or {})}, + ) + except httpx.HTTPError as exc: + raise NamoIDError( + f"{call.failure_label} could not be sent: {exc}", code=call.failure_code + ) from exc + return _handle(response, call) + + +class AsyncNamoIDClient(_ClientBase): + """Async NamoID client. Mirrors :class:`NamoIDClient` exactly.""" + + def __init__(self, *, http_client: httpx.AsyncClient | None = None, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._http = http_client + self._owns_http = http_client is None + + async def __aenter__(self) -> AsyncNamoIDClient: + return self + + async def __aexit__(self, *_exc: Any) -> None: + await self.aclose() + + async def aclose(self) -> None: + """Close the underlying HTTP client, if this instance created it.""" + if self._owns_http and self._http is not None: + await self._http.aclose() + self._http = None + + async def get_auth_config(self, *, refresh: bool = False) -> AuthConfig: + if self._config is None or refresh: + self._config = AuthConfig.from_payload( + await self._send(_config_call(self._client_id)) + ) + return self._config + + async def hosted_auth_url(self, **kwargs: Any) -> str: + return self._hosted_url(await self.get_auth_config(), kwargs) + + async def exchange_code( + self, + *, + code: str, + code_verifier: str | None = None, + client_secret: str | None = None, + device_id: str | None = None, + confidential: bool = True, + ) -> TokenResponse: + secret = self._require_secret(client_secret) if confidential else None + payload = await self._send( + _exchange_call( + code=code, + code_verifier=code_verifier, + client_id=self._client_id, + client_secret=secret, + device_id=device_id, + ) + ) + return TokenResponse.from_payload(payload) + + async def refresh(self, refresh_token: str) -> TokenResponse: + return TokenResponse.from_payload(await self._send(_refresh_call(refresh_token))) + + async def validate_access_token( + self, token: str, *, client_secret: str | None = None + ) -> TokenValidation: + secret = self._require_secret(client_secret) + payload = await self._send( + _validate_call(token=token, client_id=self._client_id, client_secret=secret) + ) + return TokenValidation.from_payload(payload) + + async def revoke_session( + self, *, access_token: str, refresh_token: str | None = None + ) -> None: + await self._send(_logout_call(access_token=access_token, refresh_token=refresh_token)) + + async def _send(self, call: _Call) -> Any: + if self._http is None: + self._http = httpx.AsyncClient(timeout=self._timeout) + self._owns_http = True + try: + response = await self._http.request( + call.method, + self._url(call.path), + params=dict(call.params or {}) or None, + json=dict(call.json) if call.json is not None else None, + headers={"accept": "application/json", **(call.headers or {})}, + ) + except httpx.HTTPError as exc: + raise NamoIDError( + f"{call.failure_label} could not be sent: {exc}", code=call.failure_code + ) from exc + return _handle(response, call) + + +def _handle(response: httpx.Response, call: _Call) -> Any: + if response.is_success: + if not call.expect_body or response.status_code == 204 or not response.content: + return None + try: + return response.json() + except ValueError as exc: + raise NamoIDError( + f"{call.failure_label} returned a non-JSON body", + status=response.status_code, + code=call.failure_code, + ) from exc + + body = _safe_json(response) + raise NamoIDError( + _error_message(body) or f"{call.failure_label} failed with {response.status_code}", + status=response.status_code, + code=_error_code(body) or call.failure_code, + detail=body, + ) + + +def _safe_json(response: httpx.Response) -> Any: + try: + return response.json() + except ValueError: + return None + + +def _error_message(body: Any) -> str | None: + if isinstance(body, Mapping): + for key in ("message", "detail"): + value = body.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _error_code(body: Any) -> str | None: + if isinstance(body, Mapping): + value = body.get("error") + if isinstance(value, str) and value: + return value + return None + + +def _compact(values: Mapping[str, Any]) -> dict: + """Drop unset keys so the API sees an absent field, not an explicit null.""" + return {key: value for key, value in values.items() if value is not None} diff --git a/src/namoid/_errors.py b/src/namoid/_errors.py new file mode 100644 index 0000000..ca61a6c --- /dev/null +++ b/src/namoid/_errors.py @@ -0,0 +1,41 @@ +"""The error type every NamoID call raises.""" + +from __future__ import annotations + +from typing import Any + +__all__ = ["NamoIDError"] + + +class NamoIDError(Exception): + """A NamoID request failed, or was refused before being sent. + + Attributes: + status: HTTP status, or ``None`` when the call never reached the API. + code: The API's machine-readable ``error`` code where one was returned, + otherwise a local code such as ``"missing_client_id"``. + detail: The parsed response body, when there was one. + """ + + def __init__( + self, + message: str, + *, + status: int | None = None, + code: str | None = None, + detail: Any = None, + ) -> None: + super().__init__(message) + self.status = status + self.code = code + self.detail = detail + + def __str__(self) -> str: + base = super().__str__() + if self.code and self.status is not None: + return f"{base} (status={self.status}, code={self.code})" + if self.code: + return f"{base} (code={self.code})" + if self.status is not None: + return f"{base} (status={self.status})" + return base diff --git a/src/namoid/hosted_auth.py b/src/namoid/hosted_auth.py new file mode 100644 index 0000000..3dba29d --- /dev/null +++ b/src/namoid/hosted_auth.py @@ -0,0 +1,321 @@ +"""Hosted Auth: types, PKCE, and URL construction. + +Hosted Auth redirects a user to a branded NamoID sign-in page and returns a +one-time code to the application. The Client ID resolves the application, its +environment, and its Hosted Auth domain, so there is no issuer or application +UUID to copy into configuration. + +Everything in this module is pure — no I/O — so it is safe to use anywhere. +:class:`namoid.NamoIDClient` and :class:`namoid.AsyncNamoIDClient` perform the +requests. +""" + +from __future__ import annotations + +import base64 +import hashlib +import secrets +from dataclasses import dataclass, field +from typing import Any, Mapping, Sequence +from urllib.parse import urlencode, urlsplit, urlunsplit + +from namoid._errors import NamoIDError + +__all__ = [ + "AuthConfig", + "HostedAuthTransaction", + "TokenResponse", + "TokenValidation", + "build_configured_hosted_auth_url", + "build_hosted_auth_url", + "create_hosted_auth_transaction", + "pkce_challenge", + "random_base64url", +] + +DEFAULT_API_BASE_URL = "https://api.namoid.in" + +# RFC 7636 allows a 43-128 character verifier. 48 random bytes encodes to 64. +_VERIFIER_BYTES = 48 +_STATE_BYTES = 32 + +_MODE_PATHS = {"sign_in": "/sign-in", "sign_up": "/sign-up", "waitlist": "/waitlist"} + + +@dataclass(frozen=True) +class AuthConfig: + """Browser-safe configuration for an application, resolved from its Client ID.""" + + client_id: str + issuer: str + hosted_auth_base_url: str + hosted_auth_pages: Mapping[str, str] + """Enabled hosted pages, keyed by ``sign_in`` / ``sign_up`` / ``waitlist`` / ``account``.""" + + access_mode: str + waitlist_enabled: bool + signin_methods: Sequence[str] + mfa_mode: str + brand_logo_url: str | None = None + brand_primary_color: str | None = None + brand_accent_color: str | None = None + brand_dark_mode: bool = False + brand_locale_default: str = "en" + support_email: str | None = None + signup_tos_required: bool = False + signup_tos_url: str | None = None + signup_privacy_url: str | None = None + raw: Mapping[str, Any] = field(default_factory=dict) + """The unmodified payload, so a newly added field is never lost.""" + + @classmethod + def from_payload(cls, payload: Mapping[str, Any]) -> AuthConfig: + try: + return cls( + client_id=payload["client_id"], + issuer=payload["issuer"], + hosted_auth_base_url=payload["hosted_auth_base_url"], + hosted_auth_pages=dict(payload.get("hosted_auth_pages") or {}), + access_mode=payload.get("access_mode", "closed"), + waitlist_enabled=bool(payload.get("waitlist_enabled", False)), + signin_methods=list(payload.get("signin_methods") or []), + mfa_mode=payload.get("mfa_mode", "off"), + brand_logo_url=payload.get("brand_logo_url"), + brand_primary_color=payload.get("brand_primary_color"), + brand_accent_color=payload.get("brand_accent_color"), + brand_dark_mode=bool(payload.get("brand_dark_mode", False)), + brand_locale_default=payload.get("brand_locale_default", "en"), + support_email=payload.get("support_email"), + signup_tos_required=bool(payload.get("signup_tos_required", False)), + signup_tos_url=payload.get("signup_tos_url"), + signup_privacy_url=payload.get("signup_privacy_url"), + raw=dict(payload), + ) + except KeyError as exc: + raise NamoIDError( + f"auth config is missing {exc.args[0]!r}", code="invalid_auth_config" + ) from exc + + +@dataclass(frozen=True) +class HostedAuthTransaction: + """State and PKCE material binding one sign-in redirect to its callback. + + Keep ``code_verifier`` server-side (or in a session) and never put it in the + redirect. Only ``code_challenge`` travels to NamoID. + """ + + state: str + code_verifier: str + code_challenge: str + code_challenge_method: str = "S256" + + +@dataclass(frozen=True) +class TokenResponse: + """The NamoID session returned by a code exchange or a refresh.""" + + access_token: str + expires_in: int + token_type: str = "Bearer" # noqa: S105 - OAuth2 literal, not a credential + refresh_token: str | None = None + user_id: str | None = None + raw: Mapping[str, Any] = field(default_factory=dict) + + @classmethod + def from_payload(cls, payload: Mapping[str, Any]) -> TokenResponse: + access_token = payload.get("access_token") + if not isinstance(access_token, str) or not access_token: + raise NamoIDError( + "token response did not include an access_token", code="invalid_token_response" + ) + expires_in = payload.get("expires_in") + return cls( + access_token=access_token, + expires_in=int(expires_in) if isinstance(expires_in, (int, float)) else 0, + token_type=payload.get("token_type") or "Bearer", + refresh_token=payload.get("refresh_token"), + user_id=str(payload["user_id"]) if payload.get("user_id") is not None else None, + raw=dict(payload), + ) + + +@dataclass(frozen=True) +class TokenValidation: + """The result of validating an access token with application credentials.""" + + valid: bool + user_id: str | None = None + session_id: str | None = None + client_id: str | None = None + scopes: Sequence[str] = () + error: str | None = None + raw: Mapping[str, Any] = field(default_factory=dict) + + @classmethod + def from_payload(cls, payload: Mapping[str, Any]) -> TokenValidation: + return cls( + valid=bool(payload.get("valid", False)), + user_id=str(payload["user_id"]) if payload.get("user_id") is not None else None, + session_id=( + str(payload["session_id"]) if payload.get("session_id") is not None else None + ), + client_id=payload.get("client_id"), + scopes=list(payload.get("scopes") or []), + error=payload.get("error"), + raw=dict(payload), + ) + + +def random_base64url(num_bytes: int = 32) -> str: + """Cryptographically random, URL-safe, unpadded base64.""" + return _b64url(secrets.token_bytes(num_bytes)) + + +def pkce_challenge(verifier: str) -> str: + """The RFC 7636 S256 challenge for ``verifier``.""" + return _b64url(hashlib.sha256(verifier.encode("ascii")).digest()) + + +def create_hosted_auth_transaction() -> HostedAuthTransaction: + """Create fresh ``state`` and PKCE material for one sign-in attempt.""" + verifier = random_base64url(_VERIFIER_BYTES) + return HostedAuthTransaction( + state=random_base64url(_STATE_BYTES), + code_verifier=verifier, + code_challenge=pkce_challenge(verifier), + code_challenge_method="S256", + ) + + +def build_hosted_auth_url( + base_url: str, + *, + return_to: str, + state: str, + completion_mode: str, + mode: str = "sign_in", + code_challenge: str | None = None, + code_challenge_method: str | None = None, + extra_params: Mapping[str, Any] | None = None, +) -> str: + """Build a Hosted Auth URL from a Hosted Auth domain. + + Prefer :func:`build_configured_hosted_auth_url`, which uses the exact page + URLs the application has enabled. + + Args: + completion_mode: ``"confidential"`` when a server will exchange the code + with a Client Secret, ``"public"`` for a browser-only client. + """ + path = _MODE_PATHS.get(mode) + if path is None: + raise NamoIDError(f"unknown Hosted Auth mode: {mode}", code="unknown_hosted_auth_mode") + + parts = urlsplit(base_url) + if not parts.scheme or not parts.netloc: + raise NamoIDError( + f"hosted auth base URL must be absolute, received {base_url!r}", + code="invalid_hosted_auth_base_url", + ) + params = _hosted_auth_params( + return_to=return_to, + state=state, + completion_mode=completion_mode, + code_challenge=code_challenge, + code_challenge_method=code_challenge_method, + ) + for key, value in (extra_params or {}).items(): + if value is not None: + params[key] = _param_str(value) + return urlunsplit((parts.scheme, parts.netloc, path, urlencode(params), "")) + + +def build_configured_hosted_auth_url( + config: AuthConfig, + *, + return_to: str, + state: str, + completion_mode: str, + mode: str = "sign_in", + code_challenge: str | None = None, + code_challenge_method: str | None = None, + extra_params: Mapping[str, Any] | None = None, +) -> str: + """Build a Hosted Auth URL from the application's own configuration. + + The configured page URL already carries the parameters that identify the + application, so those are preserved and ``extra_params`` never overrides + them. + + Raises: + NamoIDError: If the requested page is not enabled for this application. + """ + page = config.hosted_auth_pages.get(mode) + if not page: + raise NamoIDError( + f"Hosted Auth page is not enabled: {mode}", code="hosted_auth_page_disabled" + ) + + parts = urlsplit(page) + existing = _parse_query(parts.query) + params = dict(existing) + params.update( + _hosted_auth_params( + return_to=return_to, + state=state, + completion_mode=completion_mode, + code_challenge=code_challenge, + code_challenge_method=code_challenge_method, + ) + ) + for key, value in (extra_params or {}).items(): + # Never let a caller overwrite what the configured page already sets. + if value is not None and key not in params: + params[key] = _param_str(value) + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(params), "")) + + +def _hosted_auth_params( + *, + return_to: str, + state: str, + completion_mode: str, + code_challenge: str | None, + code_challenge_method: str | None, +) -> dict: + if completion_mode not in ("public", "confidential"): + raise NamoIDError( + 'completion_mode must be "public" or "confidential"', + code="invalid_completion_mode", + ) + params = { + "return_to": return_to, + "state": state, + "completion_mode": completion_mode, + } + if code_challenge: + params["code_challenge"] = code_challenge + params["code_challenge_method"] = code_challenge_method or "S256" + elif code_challenge_method: + params["code_challenge_method"] = code_challenge_method + return params + + +def _parse_query(query: str) -> dict: + if not query: + return {} + from urllib.parse import parse_qsl + + return dict(parse_qsl(query, keep_blank_values=True)) + + +def _param_str(value: Any) -> str: + if isinstance(value, bool): + # Match the JS SDK, which stringifies booleans as "true"/"false". + return "true" if value else "false" + return str(value) + + +def _b64url(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") diff --git a/src/namoid/mcp/__init__.py b/src/namoid/mcp/__init__.py new file mode 100644 index 0000000..4a10db0 --- /dev/null +++ b/src/namoid/mcp/__init__.py @@ -0,0 +1,44 @@ +"""Protect an MCP server with NamoID. + +NamoID is the authorization server. Your MCP server is the protected resource. +An MCP host — Claude, ChatGPT, Cursor, VS Code — is the OAuth client, and the +signed-in human is the resource owner. NamoID authenticates that human, records +consent, and issues a short-lived token limited to your server and to the +actions approved; this package validates and enforces it. + +No NamoID credentials are needed: a resource server only consumes public +discovery metadata and JWKS. + +Nothing here imports an MCP framework. For FastMCP, use +:mod:`namoid.mcp.fastmcp`, which is available with the ``fastmcp`` extra. +""" + +from __future__ import annotations + +from namoid._errors import NamoIDError +from namoid.mcp._authorization import ( + McpCaller, + NamoIDMcpAuth, + NamoIDMcpConfigurationError, + NamoIDMcpTokenError, + caller_from_claims, + create_namoid_mcp_auth, + create_namoid_mcp_auth_sync, + insufficient_scope_message, + insufficient_scope_payload, + protected_resource_metadata_path, +) + +__all__ = [ + "McpCaller", + "NamoIDError", + "NamoIDMcpAuth", + "NamoIDMcpConfigurationError", + "NamoIDMcpTokenError", + "caller_from_claims", + "create_namoid_mcp_auth", + "create_namoid_mcp_auth_sync", + "insufficient_scope_message", + "insufficient_scope_payload", + "protected_resource_metadata_path", +] diff --git a/src/namoid/mcp/_authorization.py b/src/namoid/mcp/_authorization.py new file mode 100644 index 0000000..84439d2 --- /dev/null +++ b/src/namoid/mcp/_authorization.py @@ -0,0 +1,691 @@ +"""Protected-resource authorization for an MCP server, framework-agnostic. + +Nothing in this module imports an MCP framework, so it can back a FastMCP +server, the official MCP Python SDK, a bare Starlette app, or a test. +:mod:`namoid.mcp.fastmcp` adapts it to FastMCP. +""" + +from __future__ import annotations + +import asyncio +import time +from dataclasses import dataclass, field +from typing import Any, Mapping, Sequence +from urllib.parse import urlsplit, urlunsplit + +import httpx +from joserfc import jwk, jws, jwt +from joserfc.errors import JoseError + +# stdlib-only, so sharing the error base couples nothing. +from namoid._errors import NamoIDError + +__all__ = [ + "McpCaller", + "NamoIDMcpAuth", + "NamoIDMcpConfigurationError", + "NamoIDMcpTokenError", + "create_namoid_mcp_auth", + "create_namoid_mcp_auth_sync", + "insufficient_scope_payload", + "protected_resource_metadata_path", +] + +# NamoID signs every environment's tokens with RS256. Nothing else is accepted. +_ALLOWED_ALGORITHMS = ["RS256"] + +_DEFAULT_CLOCK_TOLERANCE_SECONDS = 30 +_DEFAULT_DISCOVERY_TIMEOUT_SECONDS = 5.0 + +# JWKS cache lifetime. The authorization server keeps retired public keys in the +# document during rotation, so a few minutes is safe. +_DEFAULT_JWKS_TTL_SECONDS = 300.0 + +# Floor between JWKS refetches triggered by an unrecognised `kid`, so a stream of +# tokens carrying invented key IDs cannot be used to hammer the issuer. +_MIN_JWKS_REFRESH_INTERVAL_SECONDS = 10.0 + +_LOCAL_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "::1"}) + + +class NamoIDMcpConfigurationError(NamoIDError): + """Configuration or discovery is wrong. Raised at startup, never per request.""" + + +class NamoIDMcpTokenError(NamoIDError): + """A bearer token was rejected. + + The message is deliberately short and free of token contents, so it is safe + to surface as an OAuth ``error_description``. + """ + + +@dataclass(frozen=True) +class McpCaller: + """Verified identity behind a single MCP request.""" + + subject: str + """NamoID user ID (``sub``) — the human who consented, never the MCP host.""" + + client_id: str + """The OAuth client: a preregistered Client ID or a CIMD document URL.""" + + scopes: frozenset + """Scopes actually granted, not the scopes the client requested.""" + + expires_at: int + """``exp``, seconds since the epoch.""" + + resource: str + """The audience this token was minted for: this MCP server.""" + + tenant_id: str | None = None + project_id: str | None = None + environment_id: str | None = None + token_id: str | None = None + """``jti``, useful for correlating your own audit records.""" + + claims: Mapping[str, Any] = field(default_factory=dict) + """All verified claims, for anything this dataclass does not surface.""" + + def has_scopes(self, *scopes: str) -> bool: + """True when every named scope was granted.""" + return all(scope in self.scopes for scope in scopes) + + def missing_scopes(self, *scopes: str) -> list[str]: + """The named scopes that were not granted, in the order given.""" + return [scope for scope in scopes if scope not in self.scopes] + + +@dataclass(frozen=True) +class NamoIDMcpAuth: + """Everything an MCP server needs to act as a NamoID protected resource.""" + + issuer: str + """The NamoID environment issuer, exactly as it spells itself.""" + + resource: str + """Canonical MCP URL, and the exact ``aud`` every accepted token carries.""" + + resource_origin: str + """Scheme and authority of :attr:`resource`.""" + + mcp_path: str + """Path component of :attr:`resource`; where the MCP endpoint belongs.""" + + metadata_path: str + """Where RFC 9728 metadata is published, derived from :attr:`resource`.""" + + metadata_url: str + """Absolute URL of that document, for ``WWW-Authenticate`` challenges.""" + + protected_resource_metadata: Mapping[str, Any] + """The RFC 9728 document to serve at :attr:`metadata_path`.""" + + authorization_server: Mapping[str, Any] + """NamoID's authorization-server metadata, fetched once at startup.""" + + scopes_supported: tuple + """Scopes advertised for an ordinary first connection.""" + + _verifier: _TokenVerifier + + async def verify_access_token(self, token: str) -> McpCaller: + """Verify a bearer token and return the caller behind it. + + Checks, in order: RS256 against the environment's JWKS, exact ``iss``, + exact ``aud``, ``exp``/``nbf`` within the clock tolerance, + ``token_use == "access"``, and the presence of ``sub`` and ``client_id``. + + Raises: + NamoIDMcpTokenError: If any check fails. + """ + return await self._verifier.verify(token) + + +def protected_resource_metadata_path(resource: str) -> str: + """Derive the RFC 9728 metadata path for a resource URL. + + The well-known segment goes between the authority and the resource path, so + ``https://mcp.acme.example/mcp`` publishes at + ``/.well-known/oauth-protected-resource/mcp``. + """ + path = urlsplit(resource).path + suffix = "" if path in ("", "/") else path.rstrip("/") + return f"/.well-known/oauth-protected-resource{suffix}" + + +def insufficient_scope_payload( + auth: NamoIDMcpAuth, + missing_scopes: Sequence[str], + granted_scopes: Sequence[str] = (), +) -> dict: + """Machine-readable body for a tool call that lacks a scope. + + The HTTP ``403`` + ``WWW-Authenticate`` challenge applies to the whole + endpoint. A single tool needing an elevated scope has to answer inside the + JSON-RPC response, so the same ``insufficient_scope`` code and + ``resource_metadata`` pointer travel here instead — the information a host + needs to start incremental authorization rather than give up. + """ + scope_list = " ".join(missing_scopes) + return { + "error": "insufficient_scope", + "required_scopes": list(missing_scopes), + "granted_scopes": list(granted_scopes), + "resource": auth.resource, + "resource_metadata": auth.metadata_url, + "www_authenticate": ( + f'Bearer error="insufficient_scope", scope="{scope_list}", ' + f'resource_metadata="{auth.metadata_url}"' + ), + } + + +def caller_from_claims(claims: Mapping[str, Any], *, resource: str) -> McpCaller: + """Build an :class:`McpCaller` from claims that have already been verified. + + The single construction path, shared by token verification and by framework + adapters that hold verified claims rather than a raw token. + + Raises: + NamoIDMcpTokenError: If ``sub``, ``client_id``, or ``exp`` is missing. + Without ``sub`` in particular, every caller would collapse into one + shared identity and per-user checks would silently pass. + """ + expires_at = claims.get("exp") + if not isinstance(expires_at, int): + raise NamoIDMcpTokenError("token has no expiration time") + + return McpCaller( + subject=_required_str(claims, "sub"), + client_id=_required_str(claims, "client_id"), + scopes=frozenset(_parse_scopes(claims.get("scope"))), + expires_at=expires_at, + resource=resource, + tenant_id=_optional_str(claims, "tid"), + project_id=_optional_str(claims, "pid"), + environment_id=_optional_str(claims, "eid"), + token_id=_optional_str(claims, "jti"), + claims=claims, + ) + + +def insufficient_scope_message(missing_scopes: Sequence[str]) -> str: + """Human-readable counterpart to :func:`insufficient_scope_payload`.""" + return ( + f"This action needs additional authorization. Missing scope(s): " + f"{' '.join(missing_scopes)}. Reconnect and approve the additional " + f"access to continue." + ) + + +async def create_namoid_mcp_auth( + *, + issuer: str, + resource: str, + scopes_supported: Sequence[str], + resource_name: str | None = None, + resource_documentation: str | None = None, + clock_tolerance_seconds: int = _DEFAULT_CLOCK_TOLERANCE_SECONDS, + discovery_timeout_seconds: float = _DEFAULT_DISCOVERY_TIMEOUT_SECONDS, + jwks_ttl_seconds: float = _DEFAULT_JWKS_TTL_SECONDS, + jwks_min_refresh_interval_seconds: float = _MIN_JWKS_REFRESH_INTERVAL_SECONDS, + http_client: httpx.AsyncClient | None = None, +) -> NamoIDMcpAuth: + """Verify NamoID discovery and build a protected-resource authorizer. + + Args: + issuer: The NamoID environment issuer, for example + ``https://acme-test.id.namoid.in``. Test and Live differ. + resource: The canonical public URL of this MCP server, exactly as + registered as the resource audience in NamoID. Tokens carry this + string in ``aud``, so any difference rejects every token. + scopes_supported: Scopes advertised for an initial connection. Keep this + minimal; request sensitive write scopes incrementally instead. + resource_name: Human-readable name shown during discovery and consent. + resource_documentation: Documentation URL to advertise. + clock_tolerance_seconds: Allowed skew when checking ``exp``/``nbf``. + discovery_timeout_seconds: Timeout for the startup discovery fetch. + http_client: Reused for discovery and JWKS instead of a per-fetch client. + + Raises: + NamoIDMcpConfigurationError: If the issuer or resource URL is unusable, + the issuer is unreachable, or its metadata declares a different + issuer. Raised at startup so a misconfiguration is not an opaque + 401 on the first tool call. + """ + plan = _plan( + issuer=issuer, + resource=resource, + scopes_supported=scopes_supported, + resource_name=resource_name, + resource_documentation=resource_documentation, + ) + + url = _discovery_url(plan.issuer) + if http_client is not None: + raw = await _get_json_async(http_client, url, discovery_timeout_seconds) + document = _parse_discovery(plan.issuer, url, raw) + else: + async with httpx.AsyncClient() as client: + raw = await _get_json_async(client, url, discovery_timeout_seconds) + document = _parse_discovery(plan.issuer, url, raw) + + return _assemble( + plan, + document, + clock_tolerance_seconds, + http_client, + jwks_ttl_seconds=jwks_ttl_seconds, + jwks_min_refresh_interval_seconds=jwks_min_refresh_interval_seconds, + ) + + +def create_namoid_mcp_auth_sync( + *, + issuer: str, + resource: str, + scopes_supported: Sequence[str], + resource_name: str | None = None, + resource_documentation: str | None = None, + clock_tolerance_seconds: int = _DEFAULT_CLOCK_TOLERANCE_SECONDS, + discovery_timeout_seconds: float = _DEFAULT_DISCOVERY_TIMEOUT_SECONDS, + jwks_ttl_seconds: float = _DEFAULT_JWKS_TTL_SECONDS, + jwks_min_refresh_interval_seconds: float = _MIN_JWKS_REFRESH_INTERVAL_SECONDS, +) -> NamoIDMcpAuth: + """Blocking form of :func:`create_namoid_mcp_auth`. + + MCP servers are usually constructed at module import, where there is no + running event loop to await discovery on. Token verification stays async. + """ + plan = _plan( + issuer=issuer, + resource=resource, + scopes_supported=scopes_supported, + resource_name=resource_name, + resource_documentation=resource_documentation, + ) + + url = _discovery_url(plan.issuer) + try: + response = httpx.get( + url, + headers={"accept": "application/json"}, + timeout=discovery_timeout_seconds, + follow_redirects=False, + ) + response.raise_for_status() + payload = response.json() + except httpx.HTTPError as exc: + raise NamoIDMcpConfigurationError(f"could not reach {url}: {exc}") from exc + except ValueError as exc: + raise NamoIDMcpConfigurationError(f"{url} did not return JSON") from exc + + return _assemble( + plan, + _parse_discovery(plan.issuer, url, payload), + clock_tolerance_seconds, + None, + jwks_ttl_seconds=jwks_ttl_seconds, + jwks_min_refresh_interval_seconds=jwks_min_refresh_interval_seconds, + ) + + +# ─── internals ────────────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class _Plan: + issuer: str + resource: str + resource_origin: str + mcp_path: str + metadata_path: str + metadata_url: str + scopes_supported: tuple + resource_name: str | None + resource_documentation: str | None + + +def _plan( + *, + issuer: str, + resource: str, + scopes_supported: Sequence[str], + resource_name: str | None, + resource_documentation: str | None, +) -> _Plan: + checked_issuer = _validate_issuer(issuer) + parts = _validate_resource(resource) + resource_url = urlunsplit(parts) + origin = f"{parts.scheme}://{parts.netloc}" + mcp_path = parts.path or "/" + metadata_path = protected_resource_metadata_path(resource_url) + return _Plan( + issuer=checked_issuer, + resource=resource_url, + resource_origin=origin, + mcp_path=mcp_path, + metadata_path=metadata_path, + metadata_url=f"{origin}{metadata_path}", + scopes_supported=tuple(scopes_supported), + resource_name=resource_name, + resource_documentation=resource_documentation, + ) + + +def _assemble( + plan: _Plan, + document: Mapping[str, Any], + clock_tolerance_seconds: int, + http_client: httpx.AsyncClient | None, + *, + jwks_ttl_seconds: float = _DEFAULT_JWKS_TTL_SECONDS, + jwks_min_refresh_interval_seconds: float = _MIN_JWKS_REFRESH_INTERVAL_SECONDS, +) -> NamoIDMcpAuth: + metadata: dict = { + "resource": plan.resource, + # `document["issuer"]` is the issuer's own spelling of itself. + # Publishing that exact string keeps RFC 8414's exact-match issuer + # comparison working for a client that follows this document back to + # the authorization server. A normalized or slash-appended copy breaks + # strict clients. + "authorization_servers": [document["issuer"]], + "scopes_supported": list(plan.scopes_supported), + "bearer_methods_supported": ["header"], + } + if plan.resource_name is not None: + metadata["resource_name"] = plan.resource_name + if plan.resource_documentation is not None: + metadata["resource_documentation"] = plan.resource_documentation + + verifier = _TokenVerifier( + jwks=_JwksCache( + str(document["jwks_uri"]), + http_client=http_client, + ttl_seconds=jwks_ttl_seconds, + min_refresh_interval_seconds=jwks_min_refresh_interval_seconds, + ), + issuer=plan.issuer, + resource=plan.resource, + clock_tolerance_seconds=clock_tolerance_seconds, + ) + + return NamoIDMcpAuth( + issuer=plan.issuer, + resource=plan.resource, + resource_origin=plan.resource_origin, + mcp_path=plan.mcp_path, + metadata_path=plan.metadata_path, + metadata_url=plan.metadata_url, + protected_resource_metadata=metadata, + authorization_server=document, + scopes_supported=plan.scopes_supported, + _verifier=verifier, + ) + + +def _discovery_url(issuer: str) -> str: + return f"{issuer}/.well-known/oauth-authorization-server" + + +async def _get_json_async(client: httpx.AsyncClient, url: str, timeout: float) -> Any: + try: + response = await client.get( + url, + headers={"accept": "application/json"}, + timeout=timeout, + follow_redirects=False, + ) + response.raise_for_status() + return response.json() + except httpx.HTTPError as exc: + raise NamoIDMcpConfigurationError(f"could not reach {url}: {exc}") from exc + except ValueError as exc: + raise NamoIDMcpConfigurationError(f"{url} did not return JSON") from exc + + +def _parse_discovery(issuer: str, url: str, payload: Any) -> Mapping[str, Any]: + if not isinstance(payload, dict): + raise NamoIDMcpConfigurationError(f"{url} did not return an OAuth metadata document") + + declared = payload.get("issuer") + if not isinstance(declared, str) or not declared: + raise NamoIDMcpConfigurationError(f"{url} does not declare an issuer") + + # RFC 9700 mix-up defence: the document must claim the issuer we asked for. + if declared != issuer: + raise NamoIDMcpConfigurationError( + f"issuer mismatch: expected {issuer} but {url} declares {declared}" + ) + + jwks_uri = payload.get("jwks_uri") + if not isinstance(jwks_uri, str) or not jwks_uri: + raise NamoIDMcpConfigurationError(f"{url} does not advertise a jwks_uri") + + return payload + + +class _JwksCache: + """Caches the environment's JWKS, refetching when a ``kid`` is unknown. + + The authorization server signs with one active key and keeps retired public + keys published during rotation, so a token may legitimately carry a ``kid`` + that is newer than the cached document. + """ + + def __init__( + self, + uri: str, + *, + http_client: httpx.AsyncClient | None = None, + ttl_seconds: float = _DEFAULT_JWKS_TTL_SECONDS, + min_refresh_interval_seconds: float = _MIN_JWKS_REFRESH_INTERVAL_SECONDS, + ) -> None: + self._uri = uri + self._http_client = http_client + self._ttl = ttl_seconds + self._min_refresh_interval = min_refresh_interval_seconds + self._key_set: jwk.KeySet | None = None + self._kids: frozenset = frozenset() + self._fetched_at = 0.0 + self._lock = asyncio.Lock() + + async def key_set(self, *, kid: str | None) -> jwk.KeySet: + async with self._lock: + now = time.monotonic() + fresh = self._key_set is not None and (now - self._fetched_at) < self._ttl + known_kid = kid is None or kid in self._kids + if fresh and known_kid: + return self._key_set # type: ignore[return-value] + + # An unknown kid justifies an early refetch, but only at a bounded + # rate so invented key IDs cannot be turned into traffic. + if self._key_set is not None and (now - self._fetched_at) < self._min_refresh_interval: + return self._key_set + + document = await self._fetch() + try: + key_set = jwk.KeySet.import_key_set(document) + except (JoseError, ValueError, TypeError, KeyError) as exc: + raise NamoIDMcpTokenError("the signing keys could not be read") from exc + + self._key_set = key_set + self._kids = frozenset( + str(entry["kid"]) + for entry in document.get("keys", []) + if isinstance(entry, dict) and entry.get("kid") + ) + self._fetched_at = now + return key_set + + async def _fetch(self) -> dict: + try: + if self._http_client is not None: + response = await self._http_client.get( + self._uri, headers={"accept": "application/json"} + ) + else: + async with httpx.AsyncClient() as client: + response = await client.get( + self._uri, headers={"accept": "application/json"} + ) + response.raise_for_status() + document = response.json() + except httpx.HTTPError as exc: + raise NamoIDMcpTokenError("the signing keys are unavailable") from exc + except ValueError as exc: + raise NamoIDMcpTokenError("the signing keys could not be read") from exc + if not isinstance(document, dict) or not isinstance(document.get("keys"), list): + raise NamoIDMcpTokenError("the signing keys could not be read") + return document + + +class _TokenVerifier: + def __init__( + self, + *, + jwks: _JwksCache, + issuer: str, + resource: str, + clock_tolerance_seconds: int, + ) -> None: + self._jwks = jwks + self._issuer = issuer + self._resource = resource + self._claims = jwt.JWTClaimsRegistry( + leeway=clock_tolerance_seconds, + iss={"essential": True, "value": issuer}, + aud={"essential": True, "value": resource}, + sub={"essential": True}, + exp={"essential": True}, + ) + + async def verify(self, token: str) -> McpCaller: + kid = _unverified_kid(token) + key_set = await self._jwks.key_set(kid=kid) + + try: + decoded = jwt.decode(token, key_set, algorithms=_ALLOWED_ALGORITHMS) + except JoseError as exc: + raise NamoIDMcpTokenError(_describe(exc)) from exc + except (ValueError, TypeError) as exc: + raise NamoIDMcpTokenError("token could not be verified") from exc + + try: + self._claims.validate(decoded.claims) + except JoseError as exc: + raise NamoIDMcpTokenError(_describe(exc)) from exc + + claims = decoded.claims + + # An ID token is not an API token. NamoID stamps `token_use` so a + # resource server can tell them apart even though both are RS256 JWTs + # from the same issuer carrying the same `iss`. + if claims.get("token_use") != "access": + raise NamoIDMcpTokenError("token is not an access token") + + return caller_from_claims(claims, resource=self._resource) + + +def _unverified_kid(token: str) -> str | None: + """Read ``kid`` from the JOSE header without trusting anything in it. + + Used only to decide whether the cached JWKS is stale; verification still + resolves the signing key through the key set. A failure here is not fatal, + it just means the cache cannot be pre-warmed for this token. + """ + try: + header = jws.extract_compact(token.encode()).headers() + except (JoseError, ValueError, TypeError, UnicodeEncodeError): + return None + kid = header.get("kid") + return kid if isinstance(kid, str) and kid else None + + +_MISMATCH = "token issuer or audience does not match this MCP server" + + +def _describe(error: JoseError) -> str: + """Map a verification failure to a short, non-revealing reason. + + Never include the token, a claim value, or a stack trace: this string is + returned to the caller as an OAuth ``error_description``. + """ + name = type(error).__name__ + text = str(error).lower() + + if name == "ExpiredTokenError" or "expired" in text: + return "token has expired" + if name in ("MissingKeyError", "InvalidKeyIdError"): + return "token was signed with an unknown key" + if name in ("UnsupportedAlgorithmError", "MissingAlgorithmError", "ConflictAlgorithmError"): + return "token uses an unsupported signing algorithm" + if name in ("BadSignatureError", "InvalidSignatureError"): + return "token signature could not be verified" + if name == "MissingClaimError": + # Distinct from a mismatch: saying "audience does not match" for an + # absent `sub` would send an integrator down the wrong path. + return "token is missing a required claim" + if name == "InvalidClaimError": + # A wrong `iss` or a wrong `aud`. + return _MISMATCH + if "audience" in text or "issuer" in text: + return _MISMATCH + return "token could not be verified" + + +def _parse_scopes(scope: Any) -> list[str]: + if not isinstance(scope, str): + return [] + return [item for item in scope.split() if item] + + +def _required_str(claims: Mapping[str, Any], claim: str) -> str: + value = claims.get(claim) + if not isinstance(value, str) or not value: + raise NamoIDMcpTokenError(f"token is missing the {claim} claim") + return value + + +def _optional_str(claims: Mapping[str, Any], claim: str) -> str | None: + value = claims.get(claim) + return value if isinstance(value, str) and value else None + + +def _validate_issuer(value: str) -> str: + issuer = value.strip().rstrip("/") + parts = urlsplit(issuer) + if not parts.scheme or not parts.netloc: + raise NamoIDMcpConfigurationError( + f"issuer must be an absolute URL, received {value!r}" + ) + if parts.query or parts.fragment: + raise NamoIDMcpConfigurationError( + "issuer must not contain a query string or fragment" + ) + if parts.scheme != "https" and not _is_local_hostname(parts.hostname): + raise NamoIDMcpConfigurationError("issuer must use https outside local development") + return issuer + + +def _validate_resource(value: str): + parts = urlsplit(value.strip()) + if not parts.scheme or not parts.netloc: + raise NamoIDMcpConfigurationError( + f"resource must be an absolute URL, received {value!r}" + ) + if parts.fragment: + # RFC 8707 resource indicators carry no fragment, and `aud` is compared + # as an exact string. + raise NamoIDMcpConfigurationError("resource must not contain a fragment") + if parts.scheme != "https" and not _is_local_hostname(parts.hostname): + raise NamoIDMcpConfigurationError("resource must use https outside local development") + return parts + + +def _is_local_hostname(hostname: str | None) -> bool: + if hostname is None: + return False + return hostname in _LOCAL_HOSTNAMES or hostname.endswith(".localhost") diff --git a/src/namoid/mcp/fastmcp.py b/src/namoid/mcp/fastmcp.py new file mode 100644 index 0000000..a7ed8e2 --- /dev/null +++ b/src/namoid/mcp/fastmcp.py @@ -0,0 +1,283 @@ +"""FastMCP adapter for NamoID protected-resource authorization. + +Requires the ``fastmcp`` extra:: + + pip install "namoid[fastmcp]" + +Typical use, at module scope where there is no event loop to await on:: + + from fastmcp import FastMCP + from namoid.mcp.fastmcp import create_namoid_auth, require_namoid_scopes + + auth = create_namoid_auth( + issuer="https://acme-test.id.namoid.in", + resource="https://mcp.acme.example/mcp", + scopes_supported=["customers:read", "invoices:read"], + resource_name="Acme Finance MCP", + ) + + mcp = FastMCP(name="acme-finance-mcp", auth=auth.provider) + + @mcp.tool + @require_namoid_scopes(auth, "refunds:create") + def issue_refund(invoice_id: str, amount_minor: int) -> dict: + caller = current_caller(auth) + ... + + mcp.run(transport="http", path=auth.mcp_path) +""" + +from __future__ import annotations + +import functools +from dataclasses import dataclass +from typing import Any, Callable, Sequence, TypeVar + +from fastmcp.server.auth import AccessToken, RemoteAuthProvider, TokenVerifier +from fastmcp.server.dependencies import get_access_token +from fastmcp.tools.tool import ToolResult +from pydantic import AnyHttpUrl +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route + +from namoid.mcp._authorization import ( + McpCaller, + NamoIDMcpAuth, + NamoIDMcpTokenError, + caller_from_claims, + create_namoid_mcp_auth_sync, + insufficient_scope_message, + insufficient_scope_payload, +) + +__all__ = [ + "NamoIDAuthProvider", + "NamoIDFastMCPAuth", + "NamoIDTokenVerifier", + "create_namoid_auth", + "current_caller", + "require_namoid_scopes", +] + +F = TypeVar("F", bound=Callable[..., Any]) + + +class NamoIDTokenVerifier(TokenVerifier): + """Adapts NamoID token verification to FastMCP's ``TokenVerifier``. + + This wraps :class:`~namoid.mcp.NamoIDMcpAuth` rather than subclassing + FastMCP's ``JWTVerifier``, which matters for two reasons. + + ``JWTVerifier`` has no opinion on ``token_use``, so an ID token for the same + audience would be accepted as an MCP API token. It also leaves + ``AccessToken.subject`` unset, which would make every caller look like the + same anonymous principal — per-user authorization in tool handlers would + then all apply to one shared identity. Both are enforced by the core + verifier, so neither can be lost by a change in FastMCP's internals. + """ + + def __init__(self, authorization: NamoIDMcpAuth, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._authorization = authorization + + @property + def scopes_supported(self) -> list[str]: + return list(self._authorization.scopes_supported) + + async def verify_token(self, token: str) -> AccessToken | None: + try: + caller = await self._authorization.verify_access_token(token) + except NamoIDMcpTokenError: + # FastMCP turns None into a 401 plus the WWW-Authenticate challenge + # that starts client discovery. The reason stays server-side. + return None + return AccessToken( + token=token, + client_id=caller.client_id, + scopes=sorted(caller.scopes), + expires_at=caller.expires_at, + subject=caller.subject, + claims=dict(caller.claims), + ) + + +class NamoIDAuthProvider(RemoteAuthProvider): + """Publishes NamoID's protected-resource metadata verbatim. + + ``RemoteAuthProvider`` stores authorization servers as pydantic + ``AnyHttpUrl``, which appends a trailing slash to a bare-authority URL: + ``https://acme-test.id.namoid.in`` serializes as + ``https://acme-test.id.namoid.in/``. NamoID's discovery document declares + the issuer without that slash, and RFC 8414 compares issuer identifiers + exactly, so a strict client that follows the advertised authorization server + and compares the returned ``issuer`` would see a mismatch. A client building + the well-known URL by concatenation would also produce a double slash. + + Token verification and the ``WWW-Authenticate`` challenge are inherited + unchanged; only the served document is replaced. + """ + + def __init__(self, authorization: NamoIDMcpAuth, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._authorization = authorization + + def get_routes(self, mcp_path: str | None = None) -> list[Route]: + # Preserve the lifecycle hook the base class documents, then publish our + # own document instead of a re-serialized copy. + self.set_mcp_path(mcp_path) + + metadata = dict(self._authorization.protected_resource_metadata) + + async def protected_resource_metadata(_request: Request) -> JSONResponse: + return JSONResponse( + metadata, + headers={ + "Cache-Control": "public, max-age=300", + # Browser-based MCP clients fetch discovery cross-origin. + # Only this document is world-readable. + "Access-Control-Allow-Origin": "*", + }, + ) + + return [ + Route( + self._authorization.metadata_path, + protected_resource_metadata, + methods=["GET", "OPTIONS"], + ) + ] + + +@dataclass(frozen=True) +class NamoIDFastMCPAuth: + """A ready-to-mount FastMCP auth provider plus the values needed to serve it.""" + + provider: NamoIDAuthProvider + """Pass to ``FastMCP(auth=...)``.""" + + authorization: NamoIDMcpAuth + """The framework-agnostic core, for direct token verification or tests.""" + + @property + def issuer(self) -> str: + return self.authorization.issuer + + @property + def resource(self) -> str: + return self.authorization.resource + + @property + def mcp_path(self) -> str: + """Pass to ``mcp.run(path=...)`` so the endpoint matches the audience.""" + return self.authorization.mcp_path + + @property + def metadata_path(self) -> str: + return self.authorization.metadata_path + + @property + def metadata_url(self) -> str: + return self.authorization.metadata_url + + @property + def scopes_supported(self) -> tuple: + return self.authorization.scopes_supported + + +def create_namoid_auth( + *, + issuer: str, + resource: str, + scopes_supported: Sequence[str], + resource_name: str | None = None, + resource_documentation: str | None = None, + clock_tolerance_seconds: int = 30, + discovery_timeout_seconds: float = 5.0, +) -> NamoIDFastMCPAuth: + """Run NamoID discovery and build a FastMCP auth provider. + + Blocking, because FastMCP servers are built at module import where there is + no running event loop. Token verification is async. + + Raises: + NamoIDMcpConfigurationError: If the issuer or resource URL is unusable, + the issuer is unreachable, or its metadata declares a different + issuer — at startup, rather than as an opaque 401 later. + """ + authorization = create_namoid_mcp_auth_sync( + issuer=issuer, + resource=resource, + scopes_supported=scopes_supported, + resource_name=resource_name, + resource_documentation=resource_documentation, + clock_tolerance_seconds=clock_tolerance_seconds, + discovery_timeout_seconds=discovery_timeout_seconds, + ) + + verifier = NamoIDTokenVerifier( + authorization, + # Connecting needs a valid token, not every scope. Individual tools + # enforce their own, so a read-only client can still connect. + required_scopes=None, + ) + provider = NamoIDAuthProvider( + authorization, + token_verifier=verifier, + authorization_servers=[AnyHttpUrl(authorization.issuer)], + base_url=authorization.resource_origin, + scopes_supported=list(authorization.scopes_supported), + resource_name=resource_name, + ) + return NamoIDFastMCPAuth(provider=provider, authorization=authorization) + + +def current_caller(auth: NamoIDFastMCPAuth | NamoIDMcpAuth) -> McpCaller: + """The verified caller behind the current tool invocation. + + Returns the NamoID user who consented — never the MCP host or the client + application. + """ + # Both wrapper and core expose `resource`; the audience is the same either way. + token = get_access_token() + return caller_from_claims(dict(token.claims or {}), resource=auth.resource) + + +def require_namoid_scopes( + auth: NamoIDFastMCPAuth | NamoIDMcpAuth, *scopes: str +) -> Callable[[F], F]: + """Run a tool only when the caller's token carries every required scope. + + Prefer this over FastMCP's built-in ``require_scopes`` for MCP + authorization. The built-in *filters* the tool out of ``tools/list`` for + callers who lack the scope, so the host never learns the tool exists and + cannot ask the user to approve it. Incremental authorization needs the + opposite: the tool stays visible, and an attempted call answers with an + ``insufficient_scope`` challenge naming exactly what is missing. + + Use the built-in when hiding a capability is the goal. Use this when the + user should be able to grant it. + + A scope is permission to *attempt* an action. Ownership, organization + boundaries, and transaction limits still belong in the tool body. + """ + core = auth.authorization if isinstance(auth, NamoIDFastMCPAuth) else auth + + def decorate(func: F) -> F: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + granted = set(get_access_token().scopes) + missing = [scope for scope in scopes if scope not in granted] + if missing: + return ToolResult( + content=insufficient_scope_message(missing), + structured_content=insufficient_scope_payload( + core, missing, sorted(granted) + ), + is_error=True, + ) + return func(*args, **kwargs) + + return wrapper # type: ignore[return-value] + + return decorate diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d2fb91d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,135 @@ +"""A stand-in NamoID environment: RFC 8414 discovery plus a JWKS endpoint. + +Served over loopback in a background thread so the code under test exercises its +real httpx paths, including JWKS caching and refetch on an unknown ``kid``. +""" + +from __future__ import annotations + +import json +import threading +import time +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +# The fixture below needs the `mcp` extra. Import it softly so a base-only +# install can still collect and run the Hosted Auth and modularity tests. +try: + from joserfc import jwk, jwt +except ImportError: # pragma: no cover - exercised by the base-only CI job + jwk = jwt = None + +DEFAULT_KID = "test-key-1" + + +@dataclass +class FakeAuthorizationServer: + issuer: str + keys: dict = field(default_factory=dict) + """kid -> RSAKey, all published in the JWKS document.""" + + declared_issuer: str | None = None + """Overrides the ``issuer`` claim in the discovery document.""" + + omit_jwks_uri: bool = False + jwks_request_count: int = 0 + + def add_key(self, kid: str) -> jwk.RSAKey: + key = jwk.RSAKey.generate_key(2048, parameters={"kid": kid, "alg": "RS256"}) + self.keys[kid] = key + return key + + def discovery_document(self) -> dict: + document = { + "issuer": self.declared_issuer or self.issuer, + "authorization_endpoint": f"{self.issuer}/oauth/authorize", + "token_endpoint": f"{self.issuer}/v1/oauth/token", + "jwks_uri": f"{self.issuer}/v1/oauth/jwks.json", + "response_types_supported": ["code"], + "grant_types_supported": ["authorization_code", "refresh_token"], + "code_challenge_methods_supported": ["S256"], + "client_id_metadata_document_supported": True, + } + if self.omit_jwks_uri: + del document["jwks_uri"] + return document + + def jwks_document(self) -> dict: + return {"keys": [key.as_dict(private=False) for key in self.keys.values()]} + + def mint( + self, + *, + audience: str, + scope: str = "", + subject: str | None = "user-uuid-1", + client_id: str | None = "https://client.example/oauth/metadata.json", + token_use: str | None = "access", + issuer: str | None = None, + kid: str = DEFAULT_KID, + expires_in: int = 300, + ) -> str: + """Mint a token shaped like NamoID's access-token contract.""" + now = int(time.time()) + claims: dict = { + "iss": issuer or self.issuer, + "aud": audience, + "iat": now, + "nbf": now, + "exp": now + expires_in, + "tid": "tenant-uuid", + "pid": "project-uuid", + "eid": "environment-uuid", + "jti": "token-uuid", + "scope": scope, + } + if subject is not None: + claims["sub"] = subject + if client_id is not None: + claims["client_id"] = client_id + if token_use is not None: + claims["token_use"] = token_use + return jwt.encode({"alg": "RS256", "kid": kid}, claims, self.keys[kid]) + + +@pytest.fixture +def authorization_server(): + if jwk is None: + pytest.skip("requires the mcp extra (joserfc)") + state = FakeAuthorizationServer(issuer="") + state.add_key(DEFAULT_KID) + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 - BaseHTTPRequestHandler's naming + if self.path == "/.well-known/oauth-authorization-server": + return self._json(state.discovery_document()) + if self.path == "/v1/oauth/jwks.json": + state.jwks_request_count += 1 + return self._json(state.jwks_document()) + self.send_response(404) + self.end_headers() + self.wfile.write(b"{}") + + def _json(self, payload: dict) -> None: + body = json.dumps(payload).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_args): # silence per-request logging + return + + server = HTTPServer(("127.0.0.1", 0), Handler) + state.issuer = f"http://127.0.0.1:{server.server_address[1]}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield state + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) diff --git a/tests/test_hosted_auth.py b/tests/test_hosted_auth.py new file mode 100644 index 0000000..0683f6c --- /dev/null +++ b/tests/test_hosted_auth.py @@ -0,0 +1,425 @@ +"""Hosted Auth: PKCE, URL construction, and both client flavours.""" + +from __future__ import annotations + +import base64 +import hashlib +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest + +from namoid import ( + AsyncNamoIDClient, + AuthConfig, + NamoIDClient, + NamoIDError, + create_hosted_auth_transaction, +) +from namoid.hosted_auth import ( + build_configured_hosted_auth_url, + build_hosted_auth_url, + pkce_challenge, + random_base64url, +) + +CLIENT_ID = "namoid_client_test_abcdefghijklmnop" +CLIENT_SECRET = "namoid_secret_test_abcdefghijklmnop" # noqa: S105 - test fixture + +CONFIG_PAYLOAD = { + "client_id": CLIENT_ID, + "issuer": "https://acme-test.id.namoid.in", + "hosted_auth_base_url": "https://acme-test.id.namoid.in", + "hosted_auth_pages": { + "sign_in": f"https://acme-test.id.namoid.in/sign-in?client_id={CLIENT_ID}", + "sign_up": f"https://acme-test.id.namoid.in/sign-up?client_id={CLIENT_ID}", + }, + "access_mode": "open", + "waitlist_enabled": False, + "signin_methods": ["email_otp", "passkey"], + "mfa_mode": "optional", + "brand_logo_url": None, + "brand_primary_color": "#0b6650", + "brand_accent_color": None, + "brand_dark_mode": False, + "brand_locale_default": "en", + "support_email": "help@acme.example", + "signup_tos_required": True, + "signup_tos_url": "https://acme.example/terms", + "signup_privacy_url": "https://acme.example/privacy", +} + +TOKEN_PAYLOAD = { + "access_token": "access-token-value", + "refresh_token": "refresh-token-value", + "token_type": "Bearer", + "expires_in": 900, + "user_id": "11111111-1111-1111-1111-111111111111", +} + + +def recorder(handler): + """Collect the requests a client makes, alongside a response handler.""" + seen: list[httpx.Request] = [] + + def transport_handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return handler(request) + + return seen, httpx.MockTransport(transport_handler) + + +def sync_client(handler, **kwargs): + seen, transport = recorder(handler) + return seen, NamoIDClient( + client_id=CLIENT_ID, http_client=httpx.Client(transport=transport), **kwargs + ) + + +def async_client(handler, **kwargs): + seen, transport = recorder(handler) + return seen, AsyncNamoIDClient( + client_id=CLIENT_ID, http_client=httpx.AsyncClient(transport=transport), **kwargs + ) + + +# ─── pure helpers ─────────────────────────────────────────────────────────── + + +def test_transaction_carries_rfc_7636_material(): + transaction = create_hosted_auth_transaction() + + assert transaction.code_challenge_method == "S256" + assert 43 <= len(transaction.code_verifier) <= 128 + assert transaction.state != transaction.code_verifier + # The challenge must be the SHA-256 of the verifier, base64url without padding. + expected = ( + base64.urlsafe_b64encode(hashlib.sha256(transaction.code_verifier.encode()).digest()) + .decode() + .rstrip("=") + ) + assert transaction.code_challenge == expected + assert "=" not in transaction.code_challenge + + +def test_transactions_are_unique(): + states = {create_hosted_auth_transaction().state for _ in range(50)} + assert len(states) == 50 + + +def test_random_and_challenge_are_url_safe(): + for _ in range(20): + value = random_base64url(48) + assert "=" not in value and "+" not in value and "/" not in value + assert pkce_challenge("a" * 43) == pkce_challenge("a" * 43) + assert pkce_challenge("a" * 43) != pkce_challenge("b" * 43) + + +def test_builds_a_hosted_auth_url_from_a_domain(): + url = build_hosted_auth_url( + "https://acme-test.id.namoid.in", + return_to="https://app.example/callback", + state="state-value", + completion_mode="confidential", + code_challenge="challenge-value", + ) + parts = urlsplit(url) + query = parse_qs(parts.query) + + assert parts.path == "/sign-in" + assert query["return_to"] == ["https://app.example/callback"] + assert query["state"] == ["state-value"] + assert query["completion_mode"] == ["confidential"] + assert query["code_challenge"] == ["challenge-value"] + assert query["code_challenge_method"] == ["S256"] + + +@pytest.mark.parametrize( + ("mode", "path"), + [("sign_in", "/sign-in"), ("sign_up", "/sign-up"), ("waitlist", "/waitlist")], +) +def test_hosted_auth_url_modes(mode, path): + url = build_hosted_auth_url( + "https://acme-test.id.namoid.in", + mode=mode, + return_to="https://app.example/callback", + state="s", + completion_mode="public", + ) + assert urlsplit(url).path == path + + +def test_rejects_an_unknown_mode_and_a_bad_completion_mode(): + with pytest.raises(NamoIDError, match="unknown Hosted Auth mode"): + build_hosted_auth_url( + "https://acme-test.id.namoid.in", + mode="nope", + return_to="https://app.example/cb", + state="s", + completion_mode="public", + ) + with pytest.raises(NamoIDError, match="completion_mode"): + build_hosted_auth_url( + "https://acme-test.id.namoid.in", + return_to="https://app.example/cb", + state="s", + completion_mode="sometimes", + ) + + +def test_configured_url_preserves_the_application_parameters(): + config = AuthConfig.from_payload(CONFIG_PAYLOAD) + url = build_configured_hosted_auth_url( + config, + return_to="https://app.example/callback", + state="state-value", + completion_mode="public", + code_challenge="challenge-value", + extra_params={"login_hint": "user@example.com", "client_id": "attacker"}, + ) + query = parse_qs(urlsplit(url).query) + + # The client_id already on the configured page must survive untouched. + assert query["client_id"] == [CLIENT_ID] + assert query["login_hint"] == ["user@example.com"] + assert query["state"] == ["state-value"] + + +def test_configured_url_refuses_a_disabled_page(): + config = AuthConfig.from_payload(CONFIG_PAYLOAD) + with pytest.raises(NamoIDError, match="not enabled: waitlist"): + build_configured_hosted_auth_url( + config, + mode="waitlist", + return_to="https://app.example/cb", + state="s", + completion_mode="public", + ) + + +def test_auth_config_keeps_unknown_fields_and_reports_missing_ones(): + config = AuthConfig.from_payload({**CONFIG_PAYLOAD, "future_flag": True}) + assert config.support_email == "help@acme.example" + assert config.raw["future_flag"] is True + + with pytest.raises(NamoIDError, match="missing 'issuer'"): + AuthConfig.from_payload({"client_id": CLIENT_ID}) + + +# ─── client behaviour ─────────────────────────────────────────────────────── + + +def test_requires_a_client_id(): + with pytest.raises(NamoIDError, match="client_id is required"): + NamoIDClient(client_id="") + + +def test_fetches_and_caches_the_auth_config(): + seen, client = sync_client(lambda _r: httpx.Response(200, json=CONFIG_PAYLOAD)) + + first = client.get_auth_config() + second = client.get_auth_config() + + assert first.client_id == CLIENT_ID + assert first.signin_methods == ["email_otp", "passkey"] + assert second is first, "the config should be cached" + assert len(seen) == 1 + assert seen[0].url.path == "/v1/auth/config" + assert seen[0].url.params["client_id"] == CLIENT_ID + + client.get_auth_config(refresh=True) + assert len(seen) == 2 + + +def test_builds_the_hosted_url_from_the_fetched_config(): + _seen, client = sync_client(lambda _r: httpx.Response(200, json=CONFIG_PAYLOAD)) + transaction = client.create_transaction() + + url = client.hosted_auth_url( + return_to="https://app.example/callback", + state=transaction.state, + completion_mode="confidential", + code_challenge=transaction.code_challenge, + ) + query = parse_qs(urlsplit(url).query) + + assert query["client_id"] == [CLIENT_ID] + assert query["code_challenge"] == [transaction.code_challenge] + assert query["code_challenge_method"] == ["S256"] + # The verifier must never appear in the redirect. + assert transaction.code_verifier not in url + + +def test_exchanges_a_code_with_a_client_secret(): + import json + + seen, client = sync_client( + lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD), client_secret=CLIENT_SECRET + ) + + tokens = client.exchange_code(code="c" * 40, code_verifier="v" * 50) + + assert tokens.access_token == "access-token-value" + assert tokens.refresh_token == "refresh-token-value" + assert tokens.expires_in == 900 + assert tokens.user_id == "11111111-1111-1111-1111-111111111111" + + body = json.loads(seen[0].content) + assert seen[0].url.path == "/v1/auth/hosted/exchange" + assert body["client_secret"] == CLIENT_SECRET + assert body["code_verifier"] == "v" * 50 + # Unset optional fields are omitted rather than sent as null. + assert "device_id" not in body + + +def test_public_exchange_sends_no_secret(): + import json + + seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + client.exchange_code(code="c" * 40, code_verifier="v" * 50, confidential=False) + assert "client_secret" not in json.loads(seen[0].content) + + +def test_confidential_calls_refuse_to_run_without_a_secret(): + _seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + with pytest.raises(NamoIDError, match="client_secret is required"): + client.exchange_code(code="c" * 40) + with pytest.raises(NamoIDError, match="client_secret is required"): + client.validate_access_token("token") + + +def test_validates_an_access_token(): + seen, client = sync_client( + lambda _r: httpx.Response( + 200, + json={ + "valid": True, + "user_id": "22222222-2222-2222-2222-222222222222", + "session_id": "33333333-3333-3333-3333-333333333333", + "client_id": CLIENT_ID, + "scopes": ["openid", "email"], + "error": None, + }, + ), + client_secret=CLIENT_SECRET, + ) + + result = client.validate_access_token("access-token-value") + + assert result.valid is True + assert result.user_id == "22222222-2222-2222-2222-222222222222" + assert result.scopes == ["openid", "email"] + assert seen[0].url.path == "/v1/auth/tokens/validate" + + +def test_refreshes_a_session(): + seen, client = sync_client(lambda _r: httpx.Response(200, json=TOKEN_PAYLOAD)) + tokens = client.refresh("refresh-token-value") + assert tokens.access_token == "access-token-value" + assert seen[0].url.path == "/v1/auth/refresh" + + +def test_revokes_a_session_and_tolerates_an_empty_204(): + seen, client = sync_client(lambda _r: httpx.Response(204)) + client.revoke_session(access_token="access-token-value", refresh_token="refresh-token-value") + + assert seen[0].url.path == "/v1/auth/logout" + assert seen[0].headers["authorization"] == "Bearer access-token-value" + + +def test_surfaces_the_api_error_message_and_code(): + _seen, client = sync_client( + lambda _r: httpx.Response( + 400, json={"error": "invalid_grant", "message": "authorization code expired"} + ), + client_secret=CLIENT_SECRET, + ) + + with pytest.raises(NamoIDError) as excinfo: + client.exchange_code(code="c" * 40) + + error = excinfo.value + assert "authorization code expired" in str(error) + assert error.code == "invalid_grant" + assert error.status == 400 + assert error.detail["error"] == "invalid_grant" + + +def test_falls_back_to_a_generic_message_for_an_opaque_failure(): + _seen, client = sync_client(lambda _r: httpx.Response(502, text="upstream boom")) + with pytest.raises(NamoIDError) as excinfo: + client.refresh("refresh-token-value") + assert excinfo.value.status == 502 + assert excinfo.value.code == "token_refresh_failed" + + +def test_reports_a_transport_failure_without_a_status(): + def explode(_request): + raise httpx.ConnectError("connection refused") + + _seen, client = sync_client(explode) + with pytest.raises(NamoIDError) as excinfo: + client.get_auth_config() + assert excinfo.value.status is None + assert excinfo.value.code == "auth_config_failed" + + +def test_rejects_a_token_response_without_an_access_token(): + _seen, client = sync_client(lambda _r: httpx.Response(200, json={"expires_in": 900})) + with pytest.raises(NamoIDError, match="did not include an access_token"): + client.refresh("refresh-token-value") + + +def test_sync_client_works_as_a_context_manager(): + _seen, transport = recorder(lambda _r: httpx.Response(200, json=CONFIG_PAYLOAD)) + with NamoIDClient(client_id=CLIENT_ID, http_client=httpx.Client(transport=transport)) as c: + assert c.get_auth_config().client_id == CLIENT_ID + + +# ─── the async client must behave identically ─────────────────────────────── + + +async def test_async_client_mirrors_the_sync_one(): + def handler(request: httpx.Response) -> httpx.Response: + if request.url.path == "/v1/auth/config": + return httpx.Response(200, json=CONFIG_PAYLOAD) + if request.url.path == "/v1/auth/hosted/exchange": + return httpx.Response(200, json=TOKEN_PAYLOAD) + if request.url.path == "/v1/auth/logout": + return httpx.Response(204) + return httpx.Response(404, json={"error": "not_found"}) + + seen, client = async_client(handler, client_secret=CLIENT_SECRET) + async with client: + config = await client.get_auth_config() + assert config.client_id == CLIENT_ID + + url = await client.hosted_auth_url( + return_to="https://app.example/callback", + state="state-value", + completion_mode="confidential", + ) + assert parse_qs(urlsplit(url).query)["client_id"] == [CLIENT_ID] + + tokens = await client.exchange_code(code="c" * 40, code_verifier="v" * 50) + assert tokens.access_token == "access-token-value" + + await client.revoke_session(access_token=tokens.access_token) + + assert [r.url.path for r in seen] == [ + "/v1/auth/config", + "/v1/auth/hosted/exchange", + "/v1/auth/logout", + ] + + +async def test_async_client_surfaces_errors_the_same_way(): + seen, client = async_client( + lambda _r: httpx.Response(401, json={"error": "invalid_token", "message": "nope"}) + ) + async with client: + with pytest.raises(NamoIDError) as excinfo: + await client.refresh("refresh-token-value") + assert excinfo.value.code == "invalid_token" + assert excinfo.value.status == 401 + assert len(seen) == 1 diff --git a/tests/test_mcp_authorization.py b/tests/test_mcp_authorization.py new file mode 100644 index 0000000..b9b8b32 --- /dev/null +++ b/tests/test_mcp_authorization.py @@ -0,0 +1,253 @@ +"""Core protected-resource authorization: discovery, verification, metadata.""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("joserfc", reason="requires the mcp extra") + +from namoid.mcp import ( # noqa: E402 - guarded by importorskip above + NamoIDMcpConfigurationError, + NamoIDMcpTokenError, + create_namoid_mcp_auth, + create_namoid_mcp_auth_sync, + insufficient_scope_payload, + protected_resource_metadata_path, +) + +RESOURCE = "https://mcp.acme.example/mcp" + + +def build(server, **overrides): + options = { + "issuer": server.issuer, + "resource": RESOURCE, + "scopes_supported": ["customers:read", "invoices:read"], + "resource_name": "Acme Finance MCP", + } + options.update(overrides) + return create_namoid_mcp_auth_sync(**options) + + +def test_derives_the_rfc_9728_metadata_path(): + assert ( + protected_resource_metadata_path("https://mcp.acme.example/mcp") + == "/.well-known/oauth-protected-resource/mcp" + ) + assert ( + protected_resource_metadata_path("https://mcp.acme.example/") + == "/.well-known/oauth-protected-resource" + ) + assert ( + protected_resource_metadata_path("https://mcp.acme.example/a/b/") + == "/.well-known/oauth-protected-resource/a/b" + ) + + +def test_publishes_the_issuer_exactly_as_the_issuer_spells_it(authorization_server): + auth = build(authorization_server) + metadata = auth.protected_resource_metadata + + # A trailing slash here would break RFC 8414's exact issuer comparison for + # any client that follows this document back to the authorization server. + assert metadata["authorization_servers"] == [authorization_server.issuer] + assert metadata["resource"] == RESOURCE + assert metadata["resource_name"] == "Acme Finance MCP" + assert metadata["bearer_methods_supported"] == ["header"] + assert auth.metadata_path == "/.well-known/oauth-protected-resource/mcp" + assert auth.metadata_url == f"https://mcp.acme.example{auth.metadata_path}" + assert auth.mcp_path == "/mcp" + + +def test_rejects_discovery_that_declares_a_different_issuer(authorization_server): + authorization_server.declared_issuer = "https://someone-else.example" + with pytest.raises(NamoIDMcpConfigurationError, match="issuer mismatch"): + build(authorization_server) + + +def test_rejects_discovery_without_a_jwks_uri(authorization_server): + authorization_server.omit_jwks_uri = True + with pytest.raises(NamoIDMcpConfigurationError, match="jwks_uri"): + build(authorization_server) + + +def test_rejects_an_unreachable_issuer(): + # Port 1 on loopback refuses connections, so this never leaves the machine. + with pytest.raises(NamoIDMcpConfigurationError, match="could not reach"): + create_namoid_mcp_auth_sync( + issuer="http://127.0.0.1:1", + resource=RESOURCE, + scopes_supported=[], + ) + + +@pytest.mark.parametrize( + ("options", "message"), + [ + ({"issuer": "not-a-url", "resource": RESOURCE}, "absolute URL"), + ({"issuer": "http://acme.example", "resource": RESOURCE}, "https"), + ({"issuer": "https://a.example?x=1", "resource": RESOURCE}, "query"), + ({"issuer": "https://a.example", "resource": f"{RESOURCE}#frag"}, "fragment"), + ({"issuer": "https://a.example", "resource": "http://mcp.acme.example/mcp"}, "https"), + ({"issuer": "https://a.example", "resource": "/relative"}, "absolute URL"), + ], +) +def test_rejects_unusable_configuration_before_any_network_call(options, message): + with pytest.raises(NamoIDMcpConfigurationError, match=message): + create_namoid_mcp_auth_sync(scopes_supported=[], **options) + + +async def test_accepts_an_audience_bound_access_token(authorization_server): + auth = build(authorization_server) + token = authorization_server.mint( + audience=RESOURCE, scope="customers:read invoices:read" + ) + + caller = await auth.verify_access_token(token) + + assert caller.subject == "user-uuid-1" + assert caller.client_id == "https://client.example/oauth/metadata.json" + assert caller.scopes == frozenset({"customers:read", "invoices:read"}) + assert caller.resource == RESOURCE + assert caller.tenant_id == "tenant-uuid" + assert caller.project_id == "project-uuid" + assert caller.environment_id == "environment-uuid" + assert caller.token_id == "token-uuid" + assert caller.has_scopes("customers:read") + assert caller.missing_scopes("refunds:create") == ["refunds:create"] + + +@pytest.mark.parametrize( + ("label", "mint_kwargs", "message"), + [ + ( + "a token minted for another MCP server", + {"audience": "https://other.example/mcp"}, + "issuer or audience", + ), + ( + "an ID token presented as an API token", + {"audience": RESOURCE, "token_use": "id"}, + "not an access token", + ), + ( + "a token with no token_use at all", + {"audience": RESOURCE, "token_use": None}, + "not an access token", + ), + ( + "a token from a different issuer", + {"audience": RESOURCE, "issuer": "https://evil.example"}, + "issuer or audience", + ), + ( + "a token with no subject", + {"audience": RESOURCE, "subject": None}, + "missing a required claim", + ), + ( + "a token with no client_id", + {"audience": RESOURCE, "client_id": None}, + "missing the client_id claim", + ), + ( + "an expired token", + {"audience": RESOURCE, "expires_in": -60}, + "expired", + ), + ], +) +async def test_rejects_tokens_that_must_never_reach_a_tool( + authorization_server, label, mint_kwargs, message +): + auth = build(authorization_server) + token = authorization_server.mint(**mint_kwargs) + with pytest.raises(NamoIDMcpTokenError, match=message): + await auth.verify_access_token(token) + + +async def test_rejects_a_malformed_token(authorization_server): + auth = build(authorization_server) + with pytest.raises(NamoIDMcpTokenError): + await auth.verify_access_token("not-a-jwt-at-all") + + +async def test_rejects_a_signature_from_a_key_outside_the_jwks(authorization_server): + """Every claim is right; only the signing key is wrong.""" + auth = build(authorization_server) + + # Mint with a key the server knows, then drop it from the published JWKS so + # verification has to fail on the signature rather than on a claim. + authorization_server.add_key("rogue-key") + token = authorization_server.mint(audience=RESOURCE, kid="rogue-key") + del authorization_server.keys["rogue-key"] + + with pytest.raises(NamoIDMcpTokenError): + await auth.verify_access_token(token) + + +async def test_caches_jwks_and_refetches_when_a_kid_is_unknown(authorization_server): + auth = build(authorization_server, jwks_min_refresh_interval_seconds=0) + + first = authorization_server.mint(audience=RESOURCE, scope="invoices:read") + await auth.verify_access_token(first) + after_first = authorization_server.jwks_request_count + assert after_first == 1, "the first verification should fetch the key set" + + # A second token signed with the same key must reuse the cache. + await auth.verify_access_token(authorization_server.mint(audience=RESOURCE)) + assert authorization_server.jwks_request_count == after_first, "JWKS should be cached" + + # Key rotation: a token signed with a newly published kid forces a refetch + # and then verifies, because the issuer keeps retired keys during rotation. + authorization_server.add_key("test-key-2") + rotated = authorization_server.mint(audience=RESOURCE, kid="test-key-2") + caller = await auth.verify_access_token(rotated) + assert caller.subject == "user-uuid-1" + assert authorization_server.jwks_request_count == after_first + 1 + + +async def test_unknown_kid_refetches_are_rate_limited(authorization_server): + """Invented key IDs must not be usable to hammer the issuer's JWKS.""" + auth = build(authorization_server, jwks_min_refresh_interval_seconds=3600) + + await auth.verify_access_token(authorization_server.mint(audience=RESOURCE)) + baseline = authorization_server.jwks_request_count + + # A brand-new key is published, but the cooldown has not elapsed, so the + # cached key set is used and the token fails instead of triggering a fetch. + authorization_server.add_key("test-key-3") + with pytest.raises(NamoIDMcpTokenError): + await auth.verify_access_token( + authorization_server.mint(audience=RESOURCE, kid="test-key-3") + ) + assert authorization_server.jwks_request_count == baseline + + +async def test_async_factory_matches_the_sync_one(authorization_server): + auth = await create_namoid_mcp_auth( + issuer=authorization_server.issuer, + resource=RESOURCE, + scopes_supported=["invoices:read"], + ) + assert auth.protected_resource_metadata["authorization_servers"] == [ + authorization_server.issuer + ] + caller = await auth.verify_access_token( + authorization_server.mint(audience=RESOURCE, scope="invoices:read") + ) + assert caller.scopes == frozenset({"invoices:read"}) + + +def test_insufficient_scope_payload_carries_what_a_host_needs(authorization_server): + auth = build(authorization_server) + payload = insufficient_scope_payload(auth, ["refunds:create"], ["invoices:read"]) + + assert payload["error"] == "insufficient_scope" + assert payload["required_scopes"] == ["refunds:create"] + assert payload["granted_scopes"] == ["invoices:read"] + assert payload["resource"] == RESOURCE + assert payload["resource_metadata"] == auth.metadata_url + assert 'error="insufficient_scope"' in payload["www_authenticate"] + assert 'scope="refunds:create"' in payload["www_authenticate"] + assert auth.metadata_url in payload["www_authenticate"] diff --git a/tests/test_mcp_fastmcp.py b/tests/test_mcp_fastmcp.py new file mode 100644 index 0000000..787ce64 --- /dev/null +++ b/tests/test_mcp_fastmcp.py @@ -0,0 +1,204 @@ +"""The FastMCP adapter: token verifier, metadata route, and scope guard.""" + +from __future__ import annotations + +import json + +import pytest + +fastmcp_adapter = pytest.importorskip( + "namoid.mcp.fastmcp", reason="requires the fastmcp extra" +) + +create_namoid_auth = fastmcp_adapter.create_namoid_auth +require_namoid_scopes = fastmcp_adapter.require_namoid_scopes + +RESOURCE = "https://mcp.acme.example/mcp" + + +def build(server, **overrides): + options = { + "issuer": server.issuer, + "resource": RESOURCE, + "scopes_supported": ["customers:read", "invoices:read"], + "resource_name": "Acme Finance MCP", + } + options.update(overrides) + return create_namoid_auth(**options) + + +class FakeAccessToken: + """Stands in for what FastMCP injects into a tool invocation.""" + + def __init__(self, scopes, claims=None): + self.scopes = scopes + self.claims = claims or {} + self.token = "opaque" + self.client_id = "client" + + +def test_exposes_the_values_needed_to_mount_the_server(authorization_server): + auth = build(authorization_server) + + assert auth.issuer == authorization_server.issuer + assert auth.resource == RESOURCE + assert auth.mcp_path == "/mcp" + assert auth.metadata_path == "/.well-known/oauth-protected-resource/mcp" + assert auth.metadata_url == f"https://mcp.acme.example{auth.metadata_path}" + assert auth.scopes_supported == ("customers:read", "invoices:read") + assert auth.provider.token_verifier.scopes_supported == [ + "customers:read", + "invoices:read", + ] + + +async def test_verifier_populates_subject(authorization_server): + """FastMCP's own JWTVerifier leaves `subject` unset, collapsing all callers + into one anonymous identity. This adapter must not.""" + auth = build(authorization_server) + token = authorization_server.mint( + audience=RESOURCE, scope="customers:read invoices:read", subject="user-42" + ) + + access = await auth.provider.verify_token(token) + + assert access is not None + assert access.subject == "user-42" + assert access.client_id == "https://client.example/oauth/metadata.json" + assert access.scopes == ["customers:read", "invoices:read"] + assert access.claims["tid"] == "tenant-uuid" + + +@pytest.mark.parametrize( + ("label", "mint_kwargs"), + [ + ("another MCP server's audience", {"audience": "https://other.example/mcp"}), + ("an ID token", {"audience": RESOURCE, "token_use": "id"}), + ("a different issuer", {"audience": RESOURCE, "issuer": "https://evil.example"}), + ("no subject", {"audience": RESOURCE, "subject": None}), + ("an expired token", {"audience": RESOURCE, "expires_in": -60}), + ], +) +async def test_verifier_returns_none_for_rejected_tokens( + authorization_server, label, mint_kwargs +): + """FastMCP turns None into a 401 plus the discovery challenge.""" + auth = build(authorization_server) + token = authorization_server.mint(**mint_kwargs) + assert await auth.provider.verify_token(token) is None, label + + +async def test_verifier_returns_none_for_a_malformed_token(authorization_server): + auth = build(authorization_server) + assert await auth.provider.verify_token("not-a-jwt") is None + + +async def test_publishes_metadata_with_the_exact_issuer(authorization_server): + auth = build(authorization_server) + + routes = auth.provider.get_routes(mcp_path="/mcp") + metadata_routes = [r for r in routes if r.path == auth.metadata_path] + assert len(metadata_routes) == 1, f"expected one metadata route, got {routes}" + + response = await metadata_routes[0].endpoint(None) + document = json.loads(bytes(response.body)) + + # pydantic's AnyHttpUrl would render this as "/", breaking RFC 8414 + # exact issuer comparison for a client that follows it back to the AS. + assert document["authorization_servers"] == [authorization_server.issuer] + assert not document["authorization_servers"][0].endswith("/") + assert document["resource"] == RESOURCE + assert document["scopes_supported"] == ["customers:read", "invoices:read"] + assert document["bearer_methods_supported"] == ["header"] + assert response.headers["cache-control"] == "public, max-age=300" + + +def test_require_scopes_runs_the_handler_when_granted(authorization_server, monkeypatch): + auth = build(authorization_server) + monkeypatch.setattr( + fastmcp_adapter, + "get_access_token", + lambda: FakeAccessToken(["invoices:read", "refunds:create"]), + ) + + @require_namoid_scopes(auth, "refunds:create") + def issue_refund(invoice_id: str) -> dict: + return {"refunded": invoice_id} + + assert issue_refund(invoice_id="inv_1") == {"refunded": "inv_1"} + + +def test_require_scopes_challenges_a_missing_scope(authorization_server, monkeypatch): + auth = build(authorization_server) + monkeypatch.setattr( + fastmcp_adapter, + "get_access_token", + lambda: FakeAccessToken(["invoices:read"]), + ) + + ran = False + + @require_namoid_scopes(auth, "refunds:create") + def issue_refund(invoice_id: str) -> dict: + nonlocal ran + ran = True + return {} + + result = issue_refund(invoice_id="inv_1") + + assert ran is False, "the handler must not run without its scope" + assert result.is_error is True + payload = result.structured_content + assert payload["error"] == "insufficient_scope" + assert payload["required_scopes"] == ["refunds:create"] + assert payload["granted_scopes"] == ["invoices:read"] + assert payload["resource_metadata"] == auth.metadata_url + assert 'scope="refunds:create"' in payload["www_authenticate"] + + +def test_require_scopes_preserves_the_tool_signature(authorization_server, monkeypatch): + """FastMCP builds a tool schema from the wrapped function's signature.""" + import inspect + + auth = build(authorization_server) + monkeypatch.setattr( + fastmcp_adapter, "get_access_token", lambda: FakeAccessToken([]) + ) + + @require_namoid_scopes(auth, "invoices:read") + def list_invoices(status: str = "open") -> dict: + """List invoices.""" + return {} + + assert list_invoices.__name__ == "list_invoices" + assert list_invoices.__doc__ == "List invoices." + assert list(inspect.signature(list_invoices).parameters) == ["status"] + + +def test_current_caller_reconstructs_the_verified_identity( + authorization_server, monkeypatch +): + auth = build(authorization_server) + monkeypatch.setattr( + fastmcp_adapter, + "get_access_token", + lambda: FakeAccessToken( + ["invoices:read"], + claims={ + "sub": "user-77", + "client_id": "https://client.example/metadata.json", + "scope": "invoices:read", + "exp": 2_000_000_000, + "tid": "tenant-uuid", + "eid": "environment-uuid", + }, + ), + ) + + caller = fastmcp_adapter.current_caller(auth) + + assert caller.subject == "user-77" + assert caller.resource == RESOURCE + assert caller.tenant_id == "tenant-uuid" + assert caller.environment_id == "environment-uuid" + assert caller.has_scopes("invoices:read") diff --git a/tests/test_modularity.py b/tests/test_modularity.py new file mode 100644 index 0000000..62ee450 --- /dev/null +++ b/tests/test_modularity.py @@ -0,0 +1,215 @@ +"""The two surfaces stay independent. + +Each test runs in a fresh interpreter, because `sys.modules` is process-global +and any earlier import in this test session would mask the thing being checked. +""" + +from __future__ import annotations + +import subprocess +import sys + +import pytest + + +def run(source: str) -> str: + """Execute ``source`` in a clean interpreter and return its stdout.""" + result = subprocess.run( + [sys.executable, "-c", source], + capture_output=True, + text=True, + timeout=120, + ) + assert result.returncode == 0, f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" + return result.stdout.strip() + + +def test_importing_the_package_loads_neither_surface(): + assert ( + run( + """ +import sys +import namoid + +loaded = [m for m in sys.modules if m.startswith("namoid")] +assert loaded == ["namoid"], f"importing namoid pulled in {loaded}" +assert "httpx" not in sys.modules, "importing namoid should not load httpx" +assert "joserfc" not in sys.modules, "importing namoid should not load joserfc" +print("ok") +""" + ) + == "ok" + ) + + +def test_lazy_exports_resolve_and_are_discoverable(): + assert ( + run( + """ +import sys +import namoid + +# Advertised before anything is imported, so editors and dir() see the surface. +assert "NamoIDClient" in dir(namoid) +assert "NamoIDClient" in namoid.__all__ +assert "namoid._client" not in sys.modules + +client_cls = namoid.NamoIDClient +assert client_cls.__name__ == "NamoIDClient" +assert "namoid._client" in sys.modules, "naming the export should load its module" + +# Second access is served from the module globals, not __getattr__ again. +assert namoid.NamoIDClient is client_cls + +try: + namoid.NotAThing +except AttributeError as exc: + assert "NotAThing" in str(exc) +else: + raise AssertionError("unknown attributes must still raise AttributeError") +print("ok") +""" + ) + == "ok" + ) + + +def test_hosted_auth_never_loads_the_mcp_surface(): + assert ( + run( + """ +import sys +from namoid import NamoIDClient, create_hosted_auth_transaction + +create_hosted_auth_transaction() +NamoIDClient(client_id="namoid_client_test_aaaaaaaaaaaaaaaa") + +assert "namoid.mcp" not in sys.modules, "Hosted Auth must not load the MCP surface" +assert "joserfc" not in sys.modules, "Hosted Auth must not need the mcp extra" +assert "fastmcp" not in sys.modules +print("ok") +""" + ) + == "ok" + ) + + +def test_mcp_core_never_loads_hosted_auth_or_a_framework(): + pytest.importorskip("joserfc", reason="requires the mcp extra") + assert ( + run( + """ +import sys +import namoid.mcp + +assert "namoid._client" not in sys.modules, "the MCP core must not load the Hosted Auth client" +assert "namoid.hosted_auth" not in sys.modules, "the MCP core must not load Hosted Auth" +assert "fastmcp" not in sys.modules, "the MCP core must not require an MCP framework" +assert "mcp" not in sys.modules +assert "starlette" not in sys.modules +print("ok") +""" + ) + == "ok" + ) + + +def test_mcp_core_carries_no_framework_imports_even_transitively(): + """A stricter form of the above: block the frameworks at import time.""" + pytest.importorskip("joserfc", reason="requires the mcp extra") + assert ( + run( + """ +import builtins +import sys + +blocked = {"fastmcp", "starlette", "mcp", "namoid.hosted_auth", "namoid._client"} +real_import = builtins.__import__ + +def guarded(name, *args, **kwargs): + if name in blocked or any(name.startswith(b + ".") for b in blocked): + raise AssertionError(f"the MCP core must not import {name}") + return real_import(name, *args, **kwargs) + +builtins.__import__ = guarded +try: + import namoid.mcp + namoid.mcp.protected_resource_metadata_path("https://mcp.acme.example/mcp") +finally: + builtins.__import__ = real_import +print("ok") +""" + ) + == "ok" + ) + + +def test_the_fastmcp_adapter_is_opt_in_and_builds_on_the_core(): + pytest.importorskip("fastmcp", reason="requires the fastmcp extra") + assert ( + run( + """ +import sys +import namoid.mcp.fastmcp + +# The adapter is the only place a framework appears. +assert "fastmcp" in sys.modules +assert "namoid.mcp" in sys.modules, "the adapter must build on the shared core" +assert "namoid._client" not in sys.modules, "the adapter must not load Hosted Auth" +assert "namoid.hosted_auth" not in sys.modules +print("ok") +""" + ) + == "ok" + ) + + +def test_one_error_type_catches_both_surfaces(): + pytest.importorskip("joserfc", reason="requires the mcp extra") + assert ( + run( + """ +from namoid import NamoIDError +from namoid.mcp import NamoIDMcpConfigurationError, NamoIDMcpTokenError + +assert issubclass(NamoIDMcpConfigurationError, NamoIDError) +assert issubclass(NamoIDMcpTokenError, NamoIDError) + +# Sharing a base must not change the message a resource server surfaces. +assert str(NamoIDMcpTokenError("token has expired")) == "token has expired" +print("ok") +""" + ) + == "ok" + ) + + +def test_the_mcp_extra_reports_a_clear_error_when_missing(): + """`namoid.mcp` without its extra must name the dependency, not fail obscurely.""" + assert ( + run( + """ +import builtins +import sys + +real_import = builtins.__import__ + +def without_joserfc(name, *args, **kwargs): + if name == "joserfc" or name.startswith("joserfc."): + raise ModuleNotFoundError("No module named 'joserfc'", name="joserfc") + return real_import(name, *args, **kwargs) + +builtins.__import__ = without_joserfc +try: + import namoid.mcp +except ModuleNotFoundError as exc: + assert exc.name == "joserfc", exc.name +else: + raise AssertionError("expected the missing extra to surface") +finally: + builtins.__import__ = real_import +print("ok") +""" + ) + == "ok" + )