"""Deterministic offline OIDC provider used by integration and E2E tests.""" from __future__ import annotations import hashlib import json from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any, cast from urllib.parse import parse_qs, urlparse from uuid import uuid4 import httpx import jwt from cryptography.hazmat.primitives.asymmetric import rsa @dataclass class _FakeKey: kid: str private_key: Any class FakeOIDCProvider: """A local MockTransport provider; it never uses real credentials or network I/O.""" def __init__( self, *, issuer: str = "https://fake.oidc.test", audience: str = "itworx-mcp-hub", user_subject: str = "user-management-1", user_groups: tuple[str, ...] = ("hub-viewer",), persona_groups: dict[str, tuple[str, ...]] | None = None, token_ttl_seconds: int = 300, ) -> None: self.issuer = issuer self._base_url = issuer.rstrip("/") self.audience = audience self.user_subject = user_subject self.user_groups = user_groups self._token_ttl_seconds = token_ttl_seconds # Test-only: lets a real browser flow (GET /authorize, unlike # authorize() which is called directly from Python) select which # group claim to issue via a "persona" query param, so E2E tests can # exercise multiple roles against one running fake-provider process # without touching the real BFF login route to add a hint parameter. self._persona_groups = persona_groups or {} self._key = self._new_key("fake-key-1") self._codes: dict[str, tuple[str, str, tuple[str, ...]]] = {} @staticmethod def _new_key(kid: str) -> _FakeKey: return _FakeKey( kid=kid, private_key=rsa.generate_private_key(public_exponent=65537, key_size=2048), ) def rotate_keys(self, kid: str = "fake-key-rotated") -> None: self._key = self._new_key(kid) def transport(self) -> httpx.MockTransport: return httpx.MockTransport(self._handle) def asgi_app(self) -> Any: """A real ASGI app serving the same endpoints as transport(), for E2E harnesses where an actual browser must navigate to /authorize (a MockTransport only intercepts server-to-server httpx calls).""" from starlette.applications import Starlette from starlette.requests import Request as StarletteRequest from starlette.responses import Response as StarletteResponse from starlette.routing import Route async def endpoint(request: StarletteRequest) -> StarletteResponse: body = await request.body() httpx_request = httpx.Request( request.method, str(request.url), headers=list(request.headers.items()), content=body, ) httpx_response = self._handle(httpx_request) headers = { key: value for key, value in httpx_response.headers.items() if key.lower() not in {"content-length", "transfer-encoding"} } return StarletteResponse( content=httpx_response.content, status_code=httpx_response.status_code, headers=headers, ) return Starlette( routes=[Route("/{path:path}", endpoint, methods=["GET", "POST"])], ) def _issue_code(self, challenge: str, nonce: str, groups: tuple[str, ...]) -> str: code = f"fake-code-{uuid4().hex}" self._codes[code] = (challenge, nonce, groups) return code def authorize(self, authorization_url: str, *, now: datetime | None = None) -> str: query = parse_qs(urlparse(authorization_url).query) state = query.get("state", [""])[0] challenge = query.get("code_challenge", [""])[0] if not state or not challenge: raise ValueError("invalid fake authorization request") return self._issue_code(challenge, query.get("nonce", [""])[0], self.user_groups) def issue_machine_token( self, subject: str, *, now: datetime | None = None, issuer: str | None = None, audience: str | None = None, ) -> str: reference = now or datetime.now(UTC) return self._encode( { "sub": subject, "client_id": subject, "iss": issuer or self.issuer, "aud": audience or self.audience, "iat": int(reference.timestamp()), "exp": int((reference + timedelta(minutes=5)).timestamp()), } ) def _encode(self, claims: dict[str, Any]) -> str: return jwt.encode( claims, self._key.private_key, algorithm="RS256", headers={"kid": self._key.kid} ) def _public_jwk(self) -> dict[str, Any]: raw = cast( dict[str, Any], json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(self._key.private_key.public_key())), ) return raw | { "kid": self._key.kid, "use": "sig", "alg": "RS256", } def _handle(self, request: httpx.Request) -> httpx.Response: path = urlparse(str(request.url)).path base_path = urlparse(self._base_url).path.rstrip("/") if request.method == "GET" and path == f"{base_path}/.well-known/openid-configuration": return httpx.Response( 200, json={ "issuer": self.issuer, "authorization_endpoint": f"{self._base_url}/authorize", "token_endpoint": f"{self._base_url}/token", "jwks_uri": f"{self._base_url}/jwks", "id_token_signing_alg_values_supported": ["RS256"], }, request=request, ) if request.method == "GET" and path == f"{base_path}/jwks": return httpx.Response(200, json={"keys": [self._public_jwk()]}, request=request) if request.method == "GET" and path == f"{base_path}/authorize": query = parse_qs(urlparse(str(request.url)).query) redirect_uri = query.get("redirect_uri", [""])[0] state = query.get("state", [""])[0] challenge = query.get("code_challenge", [""])[0] nonce = query.get("nonce", [""])[0] persona = query.get("persona", [""])[0] if not redirect_uri or not state or not challenge: return httpx.Response( 400, text="invalid fake authorization request", request=request ) groups = self._persona_groups.get(persona, self.user_groups) code = self._issue_code(challenge, nonce, groups) separator = "&" if "?" in redirect_uri else "?" location = f"{redirect_uri}{separator}code={code}&state={state}" return httpx.Response(302, headers={"location": location}, request=request) if request.method == "POST" and path == f"{base_path}/token": form = parse_qs(request.content.decode("utf-8")) code = form.get("code", [""])[0] verifier = form.get("code_verifier", [""])[0] expected = self._codes.pop(code, None) if expected is None: return httpx.Response(400, json={"error": "invalid_grant"}, request=request) challenge = jwt.utils.base64url_encode( hashlib.sha256(verifier.encode()).digest() ).decode() if challenge != expected[0]: return httpx.Response(400, json={"error": "invalid_grant"}, request=request) now = datetime.now(UTC) id_token = self._encode( { "sub": self.user_subject, "iss": self.issuer, "aud": self.audience, "iat": int(now.timestamp()), "exp": int((now + timedelta(seconds=self._token_ttl_seconds)).timestamp()), "nonce": expected[1], "groups": list(expected[2]), } ) return httpx.Response( 200, json={ "access_token": id_token, "id_token": id_token, "token_type": "Bearer", "expires_in": self._token_ttl_seconds, }, request=request, ) return httpx.Response(404, request=request)