From 3c2e91163b386ce81c2ea83dde3d1f40e4321c87 Mon Sep 17 00:00:00 2001 From: ShivSankalp <121006829+ShivSankalp@users.noreply.github.com> Date: Tue, 4 Aug 2026 13:20:56 +0200 Subject: [PATCH 1/3] feat: protect an MCP server with NamoID NamoID is the authorization server, the customer's MCP server is the protected resource, and an MCP host is the OAuth client. This covers the resource-server half: RFC 9728 metadata, audience-bound token verification, and per-tool scope enforcement. No NamoID credentials are needed, because a resource server only consumes public discovery metadata and JWKS. namoid.mcp imports no MCP framework, so it also backs the official MCP Python SDK, a bare Starlette app, or a test. The framework appears only in namoid.mcp.fastmcp. Both arrive as extras so the base install is unaffected. The FastMCP adapter wraps the shared core rather than subclassing FastMCP's JWTVerifier, which avoids two failure modes by construction: - JWTVerifier has no opinion on token_use, so an ID token for the same audience would be accepted as an MCP API token. - JWTVerifier leaves AccessToken.subject unset, which would make every caller look like one anonymous principal and silently apply per-user checks to a shared identity. require_namoid_scopes is preferred over FastMCP's require_scopes, which filters a tool out of tools/list rather than challenging, leaving the host unable to ask the user to approve it. Incremental authorization needs the tool to stay visible and answer with insufficient_scope. The provider also publishes the issuer verbatim. RemoteAuthProvider stores authorization servers as pydantic AnyHttpUrl, which appends a trailing slash to a bare authority; RFC 8414 compares issuer identifiers exactly, so a strict client would see a mismatch. Also caches JWKS with refetch on an unrecognised kid, bounded by a cooldown so invented key IDs cannot be turned into traffic against the issuer, and adds a CI workflow, which this repository had none of. --- .github/workflows/ci.yml | 69 ++++ README.md | 129 +++++- pyproject.toml | 33 +- src/namoid/__init__.py | 2 +- src/namoid/mcp/__init__.py | 42 ++ src/namoid/mcp/_authorization.py | 688 +++++++++++++++++++++++++++++++ src/namoid/mcp/fastmcp.py | 283 +++++++++++++ tests/conftest.py | 127 ++++++ tests/test_mcp_authorization.py | 251 +++++++++++ tests/test_mcp_fastmcp.py | 204 +++++++++ 10 files changed, 1820 insertions(+), 8 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 src/namoid/mcp/__init__.py create mode 100644 src/namoid/mcp/_authorization.py create mode 100644 src/namoid/mcp/fastmcp.py create mode 100644 tests/conftest.py create mode 100644 tests/test_mcp_authorization.py create mode 100644 tests/test_mcp_fastmcp.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..96a2087 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,69 @@ +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 tests only + run: python -m pytest tests/test_mcp_authorization.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..1a2363e 100644 --- a/README.md +++ b/README.md @@ -12,12 +12,131 @@ Python SDK for [NamoID](https://namoid.in) — enterprise identity for India (OA pip install namoid ``` -## Usage +The base package has no dependencies. 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. + +## 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..639e181 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,29 @@ 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", ] +[project.optional-dependencies] +# Protect an MCP server. Framework-agnostic core: discovery, audience-bound +# token verification, and RFC 9728 metadata. +mcp = [ + "httpx>=0.28", + "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 = [ + "httpx>=0.28", + "joserfc>=1.0", + "fastmcp>=3.4.5,<4", +] + [project.urls] Homepage = "https://namoid.in" Documentation = "https://namoid.in" @@ -44,3 +65,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..720d961 100644 --- a/src/namoid/__init__.py +++ b/src/namoid/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations -__version__ = "0.0.2" +__version__ = "0.1.0" __homepage__ = "https://namoid.in" __all__ = ["__homepage__", "__version__"] diff --git a/src/namoid/mcp/__init__.py b/src/namoid/mcp/__init__.py new file mode 100644 index 0000000..ea5bdda --- /dev/null +++ b/src/namoid/mcp/__init__.py @@ -0,0 +1,42 @@ +"""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.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", + "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..ea02d0a --- /dev/null +++ b/src/namoid/mcp/_authorization.py @@ -0,0 +1,688 @@ +"""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 + +__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(Exception): + """Configuration or discovery is wrong. Raised at startup, never per request.""" + + +class NamoIDMcpTokenError(Exception): + """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..097bc9f --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,127 @@ +"""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 +from joserfc import jwk, jwt + +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(): + 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_mcp_authorization.py b/tests/test_mcp_authorization.py new file mode 100644 index 0000000..9703939 --- /dev/null +++ b/tests/test_mcp_authorization.py @@ -0,0 +1,251 @@ +"""Core protected-resource authorization: discovery, verification, metadata.""" + +from __future__ import annotations + +import pytest + +from namoid.mcp import ( + 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") From 0bddf10d78b1d613e472004cee9535d9b7b844d9 Mon Sep 17 00:00:00 2001 From: ShivSankalp <121006829+ShivSankalp@users.noreply.github.com> Date: Tue, 4 Aug 2026 13:22:07 +0200 Subject: [PATCH 2/3] feat: add the Hosted Auth client 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. NamoIDClient and AsyncNamoIDClient cover the whole flow: auth config, PKCE transactions, hosted URL construction, code exchange, refresh, server-side token validation, and session revocation. Both share one definition of every request and one response parser, so the sync and async surfaces cannot drift apart. Pure pieces live in namoid.hosted_auth, so an application that wants to drive the redirect itself does not have to construct a client. Every endpoint and payload shape was taken from the running service rather than assumed, which surfaced three fields the JavaScript SDK does not model: user_id on the token response, support_email on the auth config, and the /v1/auth/refresh endpoint, which has no method in @namoidhq/js even though an example relies on it. httpx moves from the extras to a package dependency, since making HTTP requests is now the package's core job rather than something only the MCP extra needed. Failures raise NamoIDError carrying status, the API's own error code where one is returned, and the parsed detail. --- .github/workflows/ci.yml | 4 +- README.md | 78 ++++++- pyproject.toml | 6 +- src/namoid/__init__.py | 34 ++- src/namoid/_client.py | 401 +++++++++++++++++++++++++++++++++++ src/namoid/_errors.py | 41 ++++ src/namoid/hosted_auth.py | 321 ++++++++++++++++++++++++++++ tests/test_hosted_auth.py | 425 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 1303 insertions(+), 7 deletions(-) create mode 100644 src/namoid/_client.py create mode 100644 src/namoid/_errors.py create mode 100644 src/namoid/hosted_auth.py create mode 100644 tests/test_hosted_auth.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 96a2087..c59000d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,8 +55,8 @@ jobs: python -m pip install --upgrade pip python -m pip install -e ".[mcp]" pytest pytest-asyncio - - name: Core tests only - run: python -m pytest tests/test_mcp_authorization.py -q + - name: Core and Hosted Auth tests + run: python -m pytest tests/test_mcp_authorization.py tests/test_hosted_auth.py -q - name: Assert fastmcp is absent run: | diff --git a/README.md b/README.md index 1a2363e..f2a42f5 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Python SDK for [NamoID](https://namoid.in) — enterprise identity for India (OA pip install namoid ``` -The base package has no dependencies. Protecting an MCP server adds an extra: +Hosted Auth needs only the base install. Protecting an MCP server adds an extra: ```bash pip install "namoid[fastmcp]" # protect a FastMCP server (Python 3.11+) @@ -21,6 +21,82 @@ pip install "namoid[mcp]" # the same core, without FastMCP Python 3.10 or newer. The FastMCP extra requires 3.11. +## 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. diff --git a/pyproject.toml b/pyproject.toml index 639e181..fff4208 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,17 +42,19 @@ classifiers = [ "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 = [ - "httpx>=0.28", "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 = [ - "httpx>=0.28", "joserfc>=1.0", "fastmcp>=3.4.5,<4", ] diff --git a/src/namoid/__init__.py b/src/namoid/__init__.py index 720d961..cf59687 100644 --- a/src/namoid/__init__.py +++ b/src/namoid/__init__.py @@ -1,8 +1,38 @@ -"""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). + +Hosted Auth lives here: :class:`NamoIDClient` and :class:`AsyncNamoIDClient`, +with pure helpers in :mod:`namoid.hosted_auth`. MCP authorization is in +:mod:`namoid.mcp`, behind the ``mcp`` or ``fastmcp`` extra. +""" from __future__ import annotations +from namoid._client import AsyncNamoIDClient, NamoIDClient +from namoid._errors import NamoIDError +from namoid.hosted_auth import ( + AuthConfig, + HostedAuthTransaction, + TokenResponse, + TokenValidation, + build_configured_hosted_auth_url, + build_hosted_auth_url, + create_hosted_auth_transaction, +) + __version__ = "0.1.0" __homepage__ = "https://namoid.in" -__all__ = ["__homepage__", "__version__"] +__all__ = [ + "AsyncNamoIDClient", + "AuthConfig", + "HostedAuthTransaction", + "NamoIDClient", + "NamoIDError", + "TokenResponse", + "TokenValidation", + "__homepage__", + "__version__", + "build_configured_hosted_auth_url", + "build_hosted_auth_url", + "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/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 From 65da343adfb6ac4a5b274fcf7cdfcc02b245c0dd Mon Sep 17 00:00:00 2001 From: ShivSankalp <121006829+ShivSankalp@users.noreply.github.com> Date: Tue, 4 Aug 2026 13:22:39 +0200 Subject: [PATCH 3/3] refactor: make the two surfaces strictly opt-in Importing namoid loaded the Hosted Auth client eagerly, so an MCP-only server paid for code it never asked for. The top-level names now resolve on first use (PEP 562), which gives three guarantees: - Importing namoid loads neither surface, and neither httpx nor joserfc. - A Hosted Auth application never imports the MCP code or needs its extra. - An MCP server never imports the Hosted Auth client. dir() and __all__ still list the full surface and a TYPE_CHECKING block keeps every export resolvable, so laziness costs nothing in editor or type-checker support. The MCP errors now subclass NamoIDError, so one except catches both surfaces. The error module is stdlib-only, so sharing it couples no dependency, and the message a resource server surfaces is unchanged. tests/test_modularity.py enforces the boundaries rather than leaving them as an intention. Each test runs in a fresh interpreter, because sys.modules is process-global and an earlier import in the session would mask exactly what is being checked. The strictest one replaces builtins.__import__ and fails if the MCP core so much as touches fastmcp, starlette, mcp, or the Hosted Auth modules. This also fixed the same coupling in the test suite itself: conftest imported joserfc at module scope, so a base-only install could not even collect the Hosted Auth tests. The suite now degrades by install: 31 tests on the base, 59 with the mcp extra, 73 with fastmcp. --- .github/workflows/ci.yml | 6 +- README.md | 28 ++++ src/namoid/__init__.py | 80 ++++++++++-- src/namoid/mcp/__init__.py | 2 + src/namoid/mcp/_authorization.py | 7 +- tests/conftest.py | 10 +- tests/test_mcp_authorization.py | 4 +- tests/test_modularity.py | 215 +++++++++++++++++++++++++++++++ 8 files changed, 332 insertions(+), 20 deletions(-) create mode 100644 tests/test_modularity.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c59000d..67f7b3f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,8 +55,10 @@ jobs: python -m pip install --upgrade pip python -m pip install -e ".[mcp]" pytest pytest-asyncio - - name: Core and Hosted Auth tests - run: python -m pytest tests/test_mcp_authorization.py tests/test_hosted_auth.py -q + - 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: | diff --git a/README.md b/README.md index f2a42f5..5db9e0f 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,34 @@ 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 diff --git a/src/namoid/__init__.py b/src/namoid/__init__.py index cf59687..f3fb7d4 100644 --- a/src/namoid/__init__.py +++ b/src/namoid/__init__.py @@ -1,27 +1,47 @@ """namoid: Python SDK for NamoID, enterprise identity for India (OAuth 2.1 / OIDC). -Hosted Auth lives here: :class:`NamoIDClient` and :class:`AsyncNamoIDClient`, -with pure helpers in :mod:`namoid.hosted_auth`. MCP authorization is in -:mod:`namoid.mcp`, behind the ``mcp`` or ``fastmcp`` extra. +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 -from namoid._client import AsyncNamoIDClient, NamoIDClient -from namoid._errors import NamoIDError -from namoid.hosted_auth import ( - AuthConfig, - HostedAuthTransaction, - TokenResponse, - TokenValidation, - build_configured_hosted_auth_url, - build_hosted_auth_url, - create_hosted_auth_transaction, -) +from importlib import import_module +from typing import TYPE_CHECKING, Any __version__ = "0.1.0" __homepage__ = "https://namoid.in" +# 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", @@ -36,3 +56,35 @@ "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/mcp/__init__.py b/src/namoid/mcp/__init__.py index ea5bdda..4a10db0 100644 --- a/src/namoid/mcp/__init__.py +++ b/src/namoid/mcp/__init__.py @@ -15,6 +15,7 @@ from __future__ import annotations +from namoid._errors import NamoIDError from namoid.mcp._authorization import ( McpCaller, NamoIDMcpAuth, @@ -30,6 +31,7 @@ __all__ = [ "McpCaller", + "NamoIDError", "NamoIDMcpAuth", "NamoIDMcpConfigurationError", "NamoIDMcpTokenError", diff --git a/src/namoid/mcp/_authorization.py b/src/namoid/mcp/_authorization.py index ea02d0a..84439d2 100644 --- a/src/namoid/mcp/_authorization.py +++ b/src/namoid/mcp/_authorization.py @@ -17,6 +17,9 @@ 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", @@ -45,11 +48,11 @@ _LOCAL_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "::1"}) -class NamoIDMcpConfigurationError(Exception): +class NamoIDMcpConfigurationError(NamoIDError): """Configuration or discovery is wrong. Raised at startup, never per request.""" -class NamoIDMcpTokenError(Exception): +class NamoIDMcpTokenError(NamoIDError): """A bearer token was rejected. The message is deliberately short and free of token contents, so it is safe diff --git a/tests/conftest.py b/tests/conftest.py index 097bc9f..d2fb91d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,7 +13,13 @@ from http.server import BaseHTTPRequestHandler, HTTPServer import pytest -from joserfc import jwk, jwt + +# 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" @@ -90,6 +96,8 @@ def mint( @pytest.fixture def authorization_server(): + if jwk is None: + pytest.skip("requires the mcp extra (joserfc)") state = FakeAuthorizationServer(issuer="") state.add_key(DEFAULT_KID) diff --git a/tests/test_mcp_authorization.py b/tests/test_mcp_authorization.py index 9703939..b9b8b32 100644 --- a/tests/test_mcp_authorization.py +++ b/tests/test_mcp_authorization.py @@ -4,7 +4,9 @@ import pytest -from namoid.mcp import ( +pytest.importorskip("joserfc", reason="requires the mcp extra") + +from namoid.mcp import ( # noqa: E402 - guarded by importorskip above NamoIDMcpConfigurationError, NamoIDMcpTokenError, create_namoid_mcp_auth, 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" + )