diff --git a/.github/workflows/oauth-verify.yml b/.github/workflows/oauth-verify.yml new file mode 100644 index 00000000..39cb31ea --- /dev/null +++ b/.github/workflows/oauth-verify.yml @@ -0,0 +1,36 @@ +name: OAuth Verify + +on: + push: + branches: [main, feature/sdk-oauth-verify] + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +jobs: + custom-idp-oauth: + name: Custom IdP OAuth (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-14] + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - name: Install package + run: python -m pip install -e ".[dev]" + + - name: Run custom IdP OAuth verification + run: scripts/test-oauth-custom-idp.sh diff --git a/.github/workflows/verify-examples.yml b/.github/workflows/verify-examples.yml index 4f815295..09de1e77 100644 --- a/.github/workflows/verify-examples.yml +++ b/.github/workflows/verify-examples.yml @@ -1,10 +1,10 @@ name: Verify SDK Examples on: - pull_request: - branches: [main] push: branches: [iml_verify_auto] + schedule: + - cron: '17 10 * * 1' workflow_dispatch: inputs: mode: @@ -66,9 +66,7 @@ jobs: python -m venv native-preflight source native-preflight/bin/activate python -m pip install --upgrade pip - if [ "${{ github.event_name }}" = "pull_request" ]; then - python -m pip install -e . "gopher-mcp-python-native-${{ matrix.platform }}" - elif [ "$VERIFY_PYPI_VERSION" = "latest" ]; then + if [ "$VERIFY_PYPI_VERSION" = "latest" ]; then python -m pip install gopher-mcp-python gopher-mcp-python-native-${{ matrix.platform }} else python -m pip install "gopher-mcp-python==${VERIFY_PYPI_VERSION}" "gopher-mcp-python-native-${{ matrix.platform }}==${VERIFY_PYPI_VERSION}" @@ -181,7 +179,6 @@ jobs: ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} GOPHER_API_KEY: ${{ secrets.GOPHER_API_KEY }} GOPHER_MCP_URL: ${{ secrets.GOPHER_MCP_URL }} - SDK_INSTALL_SPEC: ${{ github.event_name == 'pull_request' && github.workspace || '' }} run: | unset GOPHER_SDK_TEST VERIFY_LIVE_PROMPT="Get my mail profile" \ diff --git a/README.md b/README.md index e392010f..0e343abe 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,10 @@ Result class with status and metadata. pytest ``` +For deterministic OAuth auto verification with a local custom IdP and local MCP +server/gateway endpoints, see +[`docs/oauth-auto-custom-idp.md`](docs/oauth-auto-custom-idp.md). + ### Code Formatting This project uses Black for code formatting and Ruff for linting. diff --git a/docs/oauth-auto-custom-idp.md b/docs/oauth-auto-custom-idp.md new file mode 100644 index 00000000..2da60e94 --- /dev/null +++ b/docs/oauth-auto-custom-idp.md @@ -0,0 +1,70 @@ +# OAuth Auto Verification With Custom IdP + +The stable OAuth auto verification path uses local test fixtures instead of +Gmail, hosted Gopher services, or real OAuth provider credentials. It verifies +the Python SDK behavior that matters for automatic OAuth: + +- discovering OAuth protection from an MCP endpoint +- reading protected resource metadata +- using OAuth authorization server metadata +- refreshing a cached token through the token endpoint +- injecting the refreshed bearer token into `GopherAgent.create_with_url` + runtime options before the native FFI call + +The tests cover both endpoint shapes used by deployments: + +- direct MCP server endpoint +- MCP gateway endpoint + +Fixture credentials such as `test-client`, `test-secret`, and +`test-refresh-token` are local test data. They are not GitHub Secrets, and the +suite asserts that fixture secrets do not appear in captured output or errors. + +## Local Command + +Run the deterministic custom IdP suite with: + +```bash +scripts/test-oauth-custom-idp.sh +``` + +The script runs: + +```bash +python -m pytest \ + tests/test_oauth_auto_custom_idp.py \ + tests/test_oauth_auto_custom_idp_failures.py \ + tests/test_custom_oauth_test_idp.py \ + tests/test_custom_protected_mcp_endpoints.py \ + tests/test_oauth_test_token_helper.py +``` + +Extra pytest arguments can be passed through: + +```bash +scripts/test-oauth-custom-idp.sh -q +``` + +## CI Coverage + +`.github/workflows/oauth-verify.yml` runs the same suite on pull requests and +manual dispatch. It installs the package with development dependencies and does +not require hosted endpoints, Gmail accounts, OAuth client secrets, refresh +tokens, LLM provider keys, or other real credentials. + +## Live Smoke Tests + +Gmail or real Gopher endpoint verification remains useful as an optional smoke +test because it proves compatibility with external provider policy, hosted +gateway configuration, and real account consent. Those checks are operationally +different from SDK correctness tests: they depend on provider availability, +account security rules, valid refresh tokens, and live service configuration. + +Keep live smoke tests manual, scheduled, or otherwise separate from the stable +pull-request gate. In this repository, `.github/workflows/oauth-verify.yml` is +the pull-request OAuth gate, while `.github/workflows/verify-examples.yml` is +reserved for scheduled, branch-triggered, or manually dispatched example smoke +verification. The API example workflow and docs live under `examples/api/`. + +For a possible full native end-to-end extension, see +[`oauth-auto-native-follow-up.md`](oauth-auto-native-follow-up.md). diff --git a/docs/oauth-auto-native-follow-up.md b/docs/oauth-auto-native-follow-up.md new file mode 100644 index 00000000..f9cb6396 --- /dev/null +++ b/docs/oauth-auto-native-follow-up.md @@ -0,0 +1,45 @@ +# Optional Native OAuth E2E Follow-Up + +The custom IdP OAuth verification suite is the stable pull-request signal. It +starts local OAuth and protected MCP endpoint harnesses, lets +`GopherAgent.create_with_url` resolve OAuth automatically, and asserts that the +native FFI boundary receives runtime options containing the refreshed bearer +token. + +A full native end-to-end test can be added later, but it should stay separate +from the stable PR gate unless it is fully deterministic. + +## Proposed Shape + +Reuse the same model as `docs/oauth-auto-custom-idp.md`: + +- local custom OAuth/OIDC IdP +- local protected MCP server endpoint +- local protected MCP gateway endpoint +- refresh-token-backed OAuth setup + +Add one deterministic protected MCP tool: + +```text +tool: whoami +input: {} +output: { "subject": "test-user@example.test" } +``` + +Then create a real `GopherAgent` through `GopherAgent.create_with_url` and run a +query that must call the protected `whoami` tool. The test should assert the +authenticated result, not just successful agent creation. + +## When To Enable + +Keep this path manual, scheduled, or non-blocking until these dependencies are +controlled: + +- native package availability for each target platform +- deterministic LLM/provider behavior or a reliable test provider +- stable local MCP tool execution behavior +- clear runtime bounds suitable for CI + +Until those are in place, the focused custom IdP tests remain the correct PR +gate because they verify the SDK OAuth flow without external service or native +runtime flake. diff --git a/scripts/test-oauth-custom-idp.sh b/scripts/test-oauth-custom-idp.sh new file mode 100755 index 00000000..b816b239 --- /dev/null +++ b/scripts/test-oauth-custom-idp.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" + +cd "${REPO_ROOT}" + +PYTHON_BIN="${PYTHON:-}" +if [ -z "${PYTHON_BIN}" ]; then + if command -v python >/dev/null 2>&1; then + PYTHON_BIN="python" + else + PYTHON_BIN="python3" + fi +fi + +"${PYTHON_BIN}" -m pytest \ + tests/test_oauth_auto_custom_idp.py \ + tests/test_oauth_auto_custom_idp_failures.py \ + tests/test_custom_oauth_test_idp.py \ + tests/test_custom_protected_mcp_endpoints.py \ + tests/test_oauth_test_token_helper.py \ + "$@" diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py new file mode 100644 index 00000000..5a6392cd --- /dev/null +++ b/tests/helpers/__init__.py @@ -0,0 +1 @@ +"""Shared test helpers.""" diff --git a/tests/helpers/custom_oauth_test_idp.py b/tests/helpers/custom_oauth_test_idp.py new file mode 100644 index 00000000..b583cb97 --- /dev/null +++ b/tests/helpers/custom_oauth_test_idp.py @@ -0,0 +1,245 @@ +"""Deterministic OAuth/OIDC IdP harness for tests.""" + +from __future__ import annotations + +import json +import threading +import urllib.parse +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import TracebackType +from typing import Dict, List, Optional, Type + + +OAUTH_TEST_CLIENT_ID = "test-client" +OAUTH_TEST_CLIENT_SECRET = "test-secret" +OAUTH_TEST_REFRESH_TOKEN = "test-refresh-token" +OAUTH_TEST_ACCESS_TOKEN = "test-access-token" + + +@dataclass(frozen=True) +class CustomOAuthTestIdpOptions: + client_id: str = OAUTH_TEST_CLIENT_ID + client_secret: str = OAUTH_TEST_CLIENT_SECRET + refresh_token: str = OAUTH_TEST_REFRESH_TOKEN + access_token: str = OAUTH_TEST_ACCESS_TOKEN + scope: str = "openid profile email" + expires_in: int = 3600 + + +@dataclass +class _CustomOAuthTestIdpState: + issuer: str + client_id: str + client_secret: str + refresh_token: str + access_token: str + scope: str + expires_in: int + + +class CustomOAuthTestIdp: + def __init__( + self, + server: ThreadingHTTPServer, + thread: threading.Thread, + state: _CustomOAuthTestIdpState, + ) -> None: + self._server = server + self._thread = thread + self._state = state + self.issuer = state.issuer + self.open_id_configuration_url = ( + f"{self.issuer}/.well-known/openid-configuration" + ) + self.authorization_server_metadata_url = ( + f"{self.issuer}/.well-known/oauth-authorization-server" + ) + self.authorization_endpoint = f"{self.issuer}/authorize" + self.token_endpoint = f"{self.issuer}/token" + self.jwks_url = f"{self.issuer}/jwks" + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=1) + + def __enter__(self) -> "CustomOAuthTestIdp": + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + self.close() + + +def start_custom_oauth_test_idp( + options: Optional[CustomOAuthTestIdpOptions] = None, +) -> CustomOAuthTestIdp: + resolved = options or CustomOAuthTestIdpOptions() + state = _CustomOAuthTestIdpState( + issuer="", + client_id=resolved.client_id, + client_secret=resolved.client_secret, + refresh_token=resolved.refresh_token, + access_token=resolved.access_token, + scope=resolved.scope, + expires_in=resolved.expires_in, + ) + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt: str, *args: object) -> None: + return + + def do_GET(self) -> None: + _handle_get(self, state) + + def do_POST(self) -> None: + _handle_post(self, state) + + server = ThreadingHTTPServer(("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() + return CustomOAuthTestIdp(server, thread, state) + + +def _handle_get( + handler: BaseHTTPRequestHandler, + state: _CustomOAuthTestIdpState, +) -> None: + parsed = urllib.parse.urlparse(handler.path) + if parsed.path in ( + "/.well-known/openid-configuration", + "/.well-known/oauth-authorization-server", + ): + _json(handler, 200, _authorization_server_metadata(state)) + return + + if parsed.path == "/jwks": + _json(handler, 200, {"keys": []}) + return + + if parsed.path == "/authorize": + _json( + handler, + 200, + { + "issuer": state.issuer, + "message": "custom OAuth test IdP authorization endpoint", + }, + ) + return + + handler.send_response(404) + handler.end_headers() + + +def _handle_post( + handler: BaseHTTPRequestHandler, + state: _CustomOAuthTestIdpState, +) -> None: + parsed = urllib.parse.urlparse(handler.path) + if parsed.path == "/register": + _json( + handler, + 201, + { + "client_id": state.client_id, + "client_secret": state.client_secret, + }, + ) + return + + if parsed.path == "/token": + _handle_token_request(handler, state) + return + + handler.send_response(404) + handler.end_headers() + + +def _handle_token_request( + handler: BaseHTTPRequestHandler, + state: _CustomOAuthTestIdpState, +) -> None: + length = int(handler.headers.get("Content-Length", "0")) + body = handler.rfile.read(length).decode("utf-8") + params = urllib.parse.parse_qs(body) + + grant_type = _single(params, "grant_type") + client_id = _single(params, "client_id") + client_secret = _single(params, "client_secret") + refresh_token = _single(params, "refresh_token") + + if grant_type is None: + _oauth_error(handler, 400, "invalid_request") + return + if grant_type != "refresh_token": + _oauth_error(handler, 400, "unsupported_grant_type") + return + if client_id is None or client_secret is None or refresh_token is None: + _oauth_error(handler, 400, "invalid_request") + return + if client_id != state.client_id or client_secret != state.client_secret: + _oauth_error(handler, 401, "invalid_client") + return + if refresh_token != state.refresh_token: + _oauth_error(handler, 400, "invalid_grant") + return + + _json( + handler, + 200, + { + "access_token": state.access_token, + "token_type": "Bearer", + "expires_in": state.expires_in, + "scope": state.scope, + }, + ) + + +def _authorization_server_metadata( + state: _CustomOAuthTestIdpState, +) -> Dict[str, object]: + return { + "issuer": state.issuer, + "authorization_endpoint": f"{state.issuer}/authorize", + "token_endpoint": f"{state.issuer}/token", + "jwks_uri": f"{state.issuer}/jwks", + "registration_endpoint": f"{state.issuer}/register", + "scopes_supported": ["openid", "profile", "email"], + "response_types_supported": ["code"], + "grant_types_supported": ["refresh_token"], + "token_endpoint_auth_methods_supported": ["client_secret_post"], + } + + +def _single(params: Dict[str, List[str]], name: str) -> Optional[str]: + values = params.get(name) + return values[0] if values else None + + +def _oauth_error( + handler: BaseHTTPRequestHandler, + status: int, + error: str, +) -> None: + _json(handler, status, {"error": error}) + + +def _json( + handler: BaseHTTPRequestHandler, + status: int, + body: object, +) -> None: + data = json.dumps(body).encode("utf-8") + handler.send_response(status) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(data))) + handler.end_headers() + handler.wfile.write(data) diff --git a/tests/helpers/custom_protected_mcp_endpoints.py b/tests/helpers/custom_protected_mcp_endpoints.py new file mode 100644 index 00000000..f8f6e322 --- /dev/null +++ b/tests/helpers/custom_protected_mcp_endpoints.py @@ -0,0 +1,203 @@ +"""Protected MCP endpoint harness for OAuth auto tests.""" + +from __future__ import annotations + +import json +import threading +import urllib.parse +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import TracebackType +from typing import Dict, List, Optional, Type + + +@dataclass(frozen=True) +class CustomProtectedMcpEndpointsOptions: + authorization_server: str + access_token: str + scopes_supported: Optional[List[str]] = None + protected_resource_metadata: Optional[Dict[str, object]] = None + + +@dataclass(frozen=True) +class CustomProtectedMcpEndpoint: + kind: str + base_url: str + mcp_url: str + resource_metadata_url: str + + +@dataclass +class _EndpointState: + kind: str + base_url: str + authorization_server: str + access_token: str + scopes_supported: List[str] + protected_resource_metadata: Optional[Dict[str, object]] + + +class _RunningEndpoint: + def __init__( + self, + endpoint: CustomProtectedMcpEndpoint, + server: ThreadingHTTPServer, + thread: threading.Thread, + ) -> None: + self.endpoint = endpoint + self._server = server + self._thread = thread + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=1) + + +class CustomProtectedMcpEndpoints: + def __init__( + self, + server: _RunningEndpoint, + gateway: _RunningEndpoint, + ) -> None: + self._server_endpoint = server + self._gateway_endpoint = gateway + self.server = server.endpoint + self.gateway = gateway.endpoint + + def close(self) -> None: + self._server_endpoint.close() + self._gateway_endpoint.close() + + def __enter__(self) -> "CustomProtectedMcpEndpoints": + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + self.close() + + +def start_custom_protected_mcp_endpoints( + options: CustomProtectedMcpEndpointsOptions, +) -> CustomProtectedMcpEndpoints: + server = _start_custom_protected_mcp_endpoint("server", options) + try: + gateway = _start_custom_protected_mcp_endpoint("gateway", options) + except Exception: + server.close() + raise + return CustomProtectedMcpEndpoints(server, gateway) + + +def _start_custom_protected_mcp_endpoint( + kind: str, + options: CustomProtectedMcpEndpointsOptions, +) -> _RunningEndpoint: + state = _EndpointState( + kind=kind, + base_url="", + authorization_server=options.authorization_server, + access_token=options.access_token, + scopes_supported=options.scopes_supported or ["openid", "profile", "email"], + protected_resource_metadata=options.protected_resource_metadata, + ) + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt: str, *args: object) -> None: + return + + def do_GET(self) -> None: + _handle_endpoint_request(self, state) + + def do_POST(self) -> None: + _handle_endpoint_request(self, state) + + http_server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + state.base_url = f"http://127.0.0.1:{http_server.server_address[1]}" + thread = threading.Thread(target=http_server.serve_forever, daemon=True) + thread.start() + endpoint = CustomProtectedMcpEndpoint( + kind=kind, + base_url=state.base_url, + mcp_url=f"{state.base_url}/mcp", + resource_metadata_url=( + f"{state.base_url}/.well-known/oauth-protected-resource/mcp" + ), + ) + return _RunningEndpoint(endpoint, http_server, thread) + + +def _handle_endpoint_request( + handler: BaseHTTPRequestHandler, + state: _EndpointState, +) -> None: + parsed = urllib.parse.urlparse(handler.path) + + if ( + handler.command == "GET" + and parsed.path == "/.well-known/oauth-protected-resource/mcp" + ): + _json(handler, 200, _protected_resource_metadata(state)) + return + + if handler.command == "POST" and parsed.path == "/mcp": + if not _has_expected_bearer_token(handler, state.access_token): + handler.send_response(401) + handler.send_header( + "WWW-Authenticate", + 'Bearer realm="mcp", resource_metadata="' + f'{state.base_url}/.well-known/oauth-protected-resource/mcp"', + ) + handler.end_headers() + return + + _json( + handler, + 200, + { + "jsonrpc": "2.0", + "id": "custom-protected-mcp-response", + "result": { + "endpoint": state.kind, + "authenticated": True, + }, + }, + ) + return + + handler.send_response(404) + handler.end_headers() + + +def _protected_resource_metadata(state: _EndpointState) -> Dict[str, object]: + if state.protected_resource_metadata is not None: + return state.protected_resource_metadata + return { + "resource": f"{state.base_url}/mcp", + "authorization_servers": [state.authorization_server], + "scopes_supported": state.scopes_supported, + } + + +def _has_expected_bearer_token( + handler: BaseHTTPRequestHandler, + access_token: str, +) -> bool: + return handler.headers.get("Authorization") == f"Bearer {access_token}" + + +def _json( + handler: BaseHTTPRequestHandler, + status: int, + body: object, +) -> None: + data = json.dumps(body).encode("utf-8") + handler.send_response(status) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(data))) + handler.end_headers() + handler.wfile.write(data) diff --git a/tests/helpers/oauth_test_token.py b/tests/helpers/oauth_test_token.py new file mode 100644 index 00000000..662e7bfe --- /dev/null +++ b/tests/helpers/oauth_test_token.py @@ -0,0 +1,121 @@ +"""Generic OAuth token helpers for tests.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from typing import Dict, Optional + + +@dataclass(frozen=True) +class TestOAuthToken: + access_token: str + token_type: str + expires_in: Optional[int] = None + scope: Optional[str] = None + + +def refresh_test_oauth_token( + *, + token_endpoint: str, + client_id: str, + client_secret: str, + refresh_token: str, + timeout: float = 5.0, +) -> TestOAuthToken: + """Refresh a test OAuth token without leaking fixture secrets.""" + + body = urllib.parse.urlencode( + { + "grant_type": "refresh_token", + "client_id": client_id, + "client_secret": client_secret, + "refresh_token": refresh_token, + } + ).encode("utf-8") + request = urllib.request.Request( + token_endpoint, + data=body, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + status = response.status + payload = response.read() + except urllib.error.HTTPError as exc: + detail = _oauth_error_from_body(exc.read()) + suffix = f" ({detail})" if detail else "" + raise RuntimeError( + f"oauth_test_token_refresh_failed: token endpoint returned HTTP " + f"{exc.code}{suffix}" + ) from exc + except urllib.error.URLError as exc: + raise RuntimeError( + f"oauth_test_token_refresh_failed: {exc.reason}" + ) from exc + + if status < 200 or status >= 300: + raise RuntimeError( + f"oauth_test_token_refresh_failed: token endpoint returned HTTP {status}" + ) + + token_response = _parse_json_object(payload) + access_token = token_response.get("access_token") + token_type = token_response.get("token_type") + + if not isinstance(access_token, str) or not access_token: + raise RuntimeError( + "oauth_test_token_refresh_failed: missing access_token in token response" + ) + if not isinstance(token_type, str) or not token_type: + raise RuntimeError( + "oauth_test_token_refresh_failed: missing token_type in token response" + ) + + expires_in = token_response.get("expires_in") + if not isinstance(expires_in, int): + expires_in = None + + scope = token_response.get("scope") + if not isinstance(scope, str): + scope = None + + return TestOAuthToken( + access_token=access_token, + token_type=token_type, + expires_in=expires_in, + scope=scope, + ) + + +def _parse_json_object(payload: bytes) -> Dict[str, object]: + try: + decoded = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RuntimeError( + "oauth_test_token_refresh_failed: invalid JSON token response" + ) from exc + + if not isinstance(decoded, dict): + raise RuntimeError( + "oauth_test_token_refresh_failed: invalid JSON token response" + ) + return decoded + + +def _oauth_error_from_body(payload: bytes) -> Optional[str]: + try: + decoded = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return None + + if not isinstance(decoded, dict): + return None + + error = decoded.get("error") + return error if isinstance(error, str) and error else None diff --git a/tests/test_custom_oauth_test_idp.py b/tests/test_custom_oauth_test_idp.py new file mode 100644 index 00000000..6f54e05c --- /dev/null +++ b/tests/test_custom_oauth_test_idp.py @@ -0,0 +1,186 @@ +"""Tests for custom OAuth test IdP harness.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request +from typing import Dict + +import pytest + +from tests.helpers.custom_oauth_test_idp import ( + OAUTH_TEST_ACCESS_TOKEN, + OAUTH_TEST_CLIENT_ID, + OAUTH_TEST_CLIENT_SECRET, + OAUTH_TEST_REFRESH_TOKEN, + start_custom_oauth_test_idp, +) +from tests.helpers.oauth_test_token import refresh_test_oauth_token + + +def test_custom_oauth_test_idp_serves_metadata() -> None: + idp = start_custom_oauth_test_idp() + try: + openid_metadata = _fetch_json(idp.open_id_configuration_url) + oauth_metadata = _fetch_json(idp.authorization_server_metadata_url) + jwks = _fetch_json(idp.jwks_url) + + assert openid_metadata["issuer"] == idp.issuer + assert openid_metadata["authorization_endpoint"] == idp.authorization_endpoint + assert openid_metadata["token_endpoint"] == idp.token_endpoint + assert openid_metadata["jwks_uri"] == idp.jwks_url + assert openid_metadata["grant_types_supported"] == ["refresh_token"] + assert oauth_metadata["issuer"] == idp.issuer + assert oauth_metadata["token_endpoint"] == idp.token_endpoint + assert jwks == {"keys": []} + finally: + idp.close() + + +def test_custom_oauth_test_idp_exchanges_fixed_refresh_token() -> None: + idp = start_custom_oauth_test_idp() + try: + token = refresh_test_oauth_token( + token_endpoint=idp.token_endpoint, + client_id=OAUTH_TEST_CLIENT_ID, + client_secret=OAUTH_TEST_CLIENT_SECRET, + refresh_token=OAUTH_TEST_REFRESH_TOKEN, + ) + finally: + idp.close() + + assert token.access_token == OAUTH_TEST_ACCESS_TOKEN + assert token.token_type == "Bearer" + assert token.expires_in == 3600 + assert token.scope == "openid profile email" + + +@pytest.mark.parametrize( + ("client_id", "client_secret", "refresh_token", "expected_error"), + [ + ( + "wrong-client", + OAUTH_TEST_CLIENT_SECRET, + OAUTH_TEST_REFRESH_TOKEN, + "invalid_client", + ), + ( + OAUTH_TEST_CLIENT_ID, + "wrong-secret", + OAUTH_TEST_REFRESH_TOKEN, + "invalid_client", + ), + ( + OAUTH_TEST_CLIENT_ID, + OAUTH_TEST_CLIENT_SECRET, + "wrong-refresh-token", + "invalid_grant", + ), + ], +) +def test_custom_oauth_test_idp_returns_oauth_errors_deterministically( + client_id: str, + client_secret: str, + refresh_token: str, + expected_error: str, +) -> None: + idp = start_custom_oauth_test_idp() + try: + with pytest.raises(RuntimeError, match=expected_error): + refresh_test_oauth_token( + token_endpoint=idp.token_endpoint, + client_id=client_id, + client_secret=client_secret, + refresh_token=refresh_token, + ) + finally: + idp.close() + + +def test_custom_oauth_test_idp_rejects_unsupported_grant() -> None: + idp = start_custom_oauth_test_idp() + try: + response = _post_form( + idp.token_endpoint, + { + "grant_type": "client_credentials", + "client_id": OAUTH_TEST_CLIENT_ID, + "client_secret": OAUTH_TEST_CLIENT_SECRET, + }, + ) + finally: + idp.close() + + assert response.status == 400 + assert response.body == {"error": "unsupported_grant_type"} + + +def test_custom_oauth_test_idp_rejects_missing_fields() -> None: + idp = start_custom_oauth_test_idp() + try: + response = _post_form( + idp.token_endpoint, + { + "grant_type": "refresh_token", + "client_id": OAUTH_TEST_CLIENT_ID, + }, + ) + finally: + idp.close() + + assert response.status == 400 + assert response.body == {"error": "invalid_request"} + + +def test_custom_oauth_test_idp_authorize_endpoint_is_deterministic() -> None: + idp = start_custom_oauth_test_idp() + try: + body = _fetch_json(idp.authorization_endpoint) + finally: + idp.close() + + assert body == { + "issuer": idp.issuer, + "message": "custom OAuth test IdP authorization endpoint", + } + + +def test_custom_oauth_test_idp_close_stops_server() -> None: + idp = start_custom_oauth_test_idp() + idp.close() + + with pytest.raises(urllib.error.URLError): + _fetch_json(idp.open_id_configuration_url) + + +class _FormResponse: + def __init__(self, status: int, body: object) -> None: + self.status = status + self.body = body + + +def _fetch_json(url: str) -> object: + with urllib.request.urlopen(url, timeout=5) as response: + return json.loads(response.read().decode("utf-8")) + + +def _post_form(url: str, params: Dict[str, str]) -> _FormResponse: + data = urllib.parse.urlencode(params).encode("utf-8") + request = urllib.request.Request( + url, + data=data, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + try: + with urllib.request.urlopen(request, timeout=5) as response: + status = response.status + body = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + status = exc.code + body = json.loads(exc.read().decode("utf-8")) + + return _FormResponse(status, body) diff --git a/tests/test_custom_protected_mcp_endpoints.py b/tests/test_custom_protected_mcp_endpoints.py new file mode 100644 index 00000000..ad286be5 --- /dev/null +++ b/tests/test_custom_protected_mcp_endpoints.py @@ -0,0 +1,173 @@ +"""Tests for custom protected MCP endpoint harness.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request + +import pytest + +from tests.helpers.custom_oauth_test_idp import ( + OAUTH_TEST_ACCESS_TOKEN, + start_custom_oauth_test_idp, +) +from tests.helpers.custom_protected_mcp_endpoints import ( + CustomProtectedMcpEndpoint, + CustomProtectedMcpEndpointsOptions, + start_custom_protected_mcp_endpoints, +) + + +def test_server_and_gateway_endpoints_advertise_oauth_protection() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + for endpoint in (endpoints.server, endpoints.gateway): + response = _post_json(endpoint.mcp_url, {"jsonrpc": "2.0", "id": "probe"}) + assert response.status == 401 + assert response.headers["WWW-Authenticate"] == ( + 'Bearer realm="mcp", resource_metadata="' + f'{endpoint.resource_metadata_url}"' + ) + finally: + endpoints.close() + idp.close() + + +def test_server_and_gateway_metadata_points_to_custom_idp_issuer() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + for endpoint in (endpoints.server, endpoints.gateway): + metadata = _fetch_json(endpoint.resource_metadata_url) + assert metadata == { + "resource": endpoint.mcp_url, + "authorization_servers": [idp.issuer], + "scopes_supported": ["openid", "profile", "email"], + } + finally: + endpoints.close() + idp.close() + + +def test_server_and_gateway_endpoints_reject_wrong_bearer_token() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + for endpoint in (endpoints.server, endpoints.gateway): + response = _post_json( + endpoint.mcp_url, + {"jsonrpc": "2.0", "id": "wrong-token"}, + access_token="wrong-token", + ) + assert response.status == 401 + finally: + endpoints.close() + idp.close() + + +def test_server_and_gateway_endpoints_accept_deterministic_bearer_token() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + for endpoint in (endpoints.server, endpoints.gateway): + body = _post_authenticated(endpoint) + assert body["result"] == { + "endpoint": endpoint.kind, + "authenticated": True, + } + finally: + endpoints.close() + idp.close() + + +def test_close_stops_both_local_endpoints() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + server_metadata_url = endpoints.server.resource_metadata_url + gateway_metadata_url = endpoints.gateway.resource_metadata_url + endpoints.close() + idp.close() + + with pytest.raises(urllib.error.URLError): + _fetch_json(server_metadata_url) + with pytest.raises(urllib.error.URLError): + _fetch_json(gateway_metadata_url) + + +def _post_authenticated(endpoint: CustomProtectedMcpEndpoint) -> object: + response = _post_json( + endpoint.mcp_url, + {"jsonrpc": "2.0", "id": endpoint.kind}, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + assert response.status == 200 + return response.body + + +class _Response: + def __init__( + self, + status: int, + headers: dict, + body: object = None, + ) -> None: + self.status = status + self.headers = headers + self.body = body + + +def _fetch_json(url: str) -> object: + with urllib.request.urlopen(url, timeout=5) as response: + return json.loads(response.read().decode("utf-8")) + + +def _post_json( + url: str, + body: object, + access_token: str = "", +) -> _Response: + data = json.dumps(body).encode("utf-8") + headers = {"Content-Type": "application/json"} + if access_token: + headers["Authorization"] = f"Bearer {access_token}" + request = urllib.request.Request( + url, + data=data, + method="POST", + headers=headers, + ) + + try: + with urllib.request.urlopen(request, timeout=5) as response: + response_body = response.read() + parsed_body = json.loads(response_body.decode("utf-8")) + return _Response(response.status, dict(response.headers), parsed_body) + except urllib.error.HTTPError as exc: + return _Response(exc.code, dict(exc.headers)) diff --git a/tests/test_linux_native_packaging.py b/tests/test_linux_native_packaging.py index cb429464..235f63bc 100644 --- a/tests/test_linux_native_packaging.py +++ b/tests/test_linux_native_packaging.py @@ -23,6 +23,10 @@ def _verify_examples_workflow() -> str: return (ROOT / ".github" / "workflows" / "verify-examples.yml").read_text() +def _oauth_verify_workflow() -> str: + return (ROOT / ".github" / "workflows" / "oauth-verify.yml").read_text() + + def _verify_examples_script() -> str: return (ROOT / "scripts" / "verify-examples.sh").read_text() @@ -46,22 +50,28 @@ def test_linux_x64_uses_digest_pinned_ubuntu_builder_image() -> None: assert re.search(r"\subuntu:20\.04\s", build_script) is None -def test_verify_examples_prs_install_checked_out_sdk() -> None: +def test_oauth_verify_prs_install_checked_out_sdk() -> None: + workflow = _oauth_verify_workflow() + + assert "pull_request:" in workflow + assert "branches: [main]" in workflow + assert 'python -m pip install -e ".[dev]"' in workflow + assert "scripts/test-oauth-custom-idp.sh" in workflow + + +def test_verify_examples_workflow_stays_optional_for_live_smoke() -> None: workflow = _verify_examples_workflow() assert ( "VERIFY_EXAMPLES_MODE: ${{ github.event_name == 'workflow_dispatch' " "&& inputs.mode || 'auto' }}" ) in workflow - assert 'if [ "${{ github.event_name }}" = "pull_request" ]; then' in workflow - assert ( - 'python -m pip install -e . "gopher-mcp-python-native-${{ matrix.platform }}"' - in workflow - ) - assert ( - "SDK_INSTALL_SPEC: ${{ github.event_name == 'pull_request' && " - "github.workspace || '' }}" - ) in workflow + assert "pull_request:" not in workflow + assert "schedule:" in workflow + assert "workflow_dispatch:" in workflow + assert "branches: [iml_verify_auto]" in workflow + assert 'if [ "${{ github.event_name }}" = "pull_request" ]; then' not in workflow + assert "SDK_INSTALL_SPEC:" not in workflow def test_verify_examples_workflow_bounds_pr_cost_and_runtime() -> None: diff --git a/tests/test_oauth_auto_custom_idp.py b/tests/test_oauth_auto_custom_idp.py new file mode 100644 index 00000000..d135cec5 --- /dev/null +++ b/tests/test_oauth_auto_custom_idp.py @@ -0,0 +1,155 @@ +"""OAuth auto verification with custom IdP.""" + +from __future__ import annotations + +from typing import Dict, List, Optional + +import gopher_mcp_python.agent as agent_module +import gopher_mcp_python.oauth_resolver as oauth_resolver +from gopher_mcp_python import GopherAgent +from gopher_mcp_python.runtime_options import GopherAgentTokenRecord +from tests.helpers.custom_oauth_test_idp import ( + OAUTH_TEST_ACCESS_TOKEN, + OAUTH_TEST_CLIENT_SECRET, + OAUTH_TEST_REFRESH_TOKEN, + start_custom_oauth_test_idp, +) +from tests.helpers.custom_protected_mcp_endpoints import ( + CustomProtectedMcpEndpoint, + CustomProtectedMcpEndpointsOptions, + start_custom_protected_mcp_endpoints, +) + + +PROVIDER = "AnthropicProvider" +MODEL = "test-model" + + +class _FakeLibrary: + def __init__(self) -> None: + self.calls: List[tuple] = [] + + def agent_create_by_url(self, provider, model, url, runtime_options=None): + self.calls.append(("url", provider, model, url, runtime_options)) + return 4001 + + def agent_release(self, handle): + self.calls.append(("release", handle)) + + def get_last_error_message(self): + return None + + def clear_error(self): + self.calls.append(("clear_error",)) + + +class _RefreshTokenStore: + def __init__(self) -> None: + self.tokens: Dict[str, GopherAgentTokenRecord] = {} + self.set_calls: List[tuple] = [] + self.delete_calls: List[str] = [] + + async def get(self, key: str) -> Optional[GopherAgentTokenRecord]: + if key not in self.tokens: + return GopherAgentTokenRecord( + access_token="expired-access-token", + refresh_token=OAUTH_TEST_REFRESH_TOKEN, + token_type="Bearer", + expires_at=0, + ) + return self.tokens[key] + + async def set(self, key: str, token: GopherAgentTokenRecord) -> None: + self.set_calls.append((key, token)) + self.tokens[key] = token + + async def delete(self, key: str) -> None: + self.delete_calls.append(key) + self.tokens.pop(key, None) + + +def test_injects_refreshed_token_for_direct_mcp_server_endpoint( + monkeypatch, + capsys, +) -> None: + _expect_refreshed_token_injected_for_endpoint_name( + monkeypatch, + capsys, + endpoint_name="server", + ) + + +def test_injects_refreshed_token_for_mcp_gateway_endpoint( + monkeypatch, + capsys, +) -> None: + _expect_refreshed_token_injected_for_endpoint_name( + monkeypatch, + capsys, + endpoint_name="gateway", + ) + + +def _expect_refreshed_token_injected_for_endpoint_name( + monkeypatch, + capsys, + endpoint_name: str, +) -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + _expect_refreshed_token_injected_for_endpoint( + monkeypatch, + endpoint=getattr(endpoints, endpoint_name), + ) + finally: + endpoints.close() + idp.close() + oauth_resolver.set_oauth_resolver_hooks_for_test() + oauth_resolver.set_oauth_url_runtime_options_resolver_for_test() + + captured = capsys.readouterr() + assert OAUTH_TEST_CLIENT_SECRET not in captured.out + assert OAUTH_TEST_CLIENT_SECRET not in captured.err + assert OAUTH_TEST_REFRESH_TOKEN not in captured.out + assert OAUTH_TEST_REFRESH_TOKEN not in captured.err + assert OAUTH_TEST_ACCESS_TOKEN not in captured.out + assert OAUTH_TEST_ACCESS_TOKEN not in captured.err + + +def _expect_refreshed_token_injected_for_endpoint( + monkeypatch, + endpoint: CustomProtectedMcpEndpoint, +) -> None: + fake = _FakeLibrary() + token_store = _RefreshTokenStore() + monkeypatch.setattr(agent_module, "_initialized", True) + monkeypatch.setattr( + agent_module.GopherOrchLibrary, + "get_instance", + staticmethod(lambda: fake), + ) + + agent = GopherAgent.create_with_url( + PROVIDER, + MODEL, + endpoint.mcp_url, + { + "oauth": { + "token_store": token_store, + }, + }, + ) + + call = fake.calls[0] + assert call[:4] == ("url", PROVIDER, MODEL, endpoint.mcp_url) + assert call[4].access_token == OAUTH_TEST_ACCESS_TOKEN + assert token_store.set_calls + assert token_store.set_calls[0][1].access_token == OAUTH_TEST_ACCESS_TOKEN + assert token_store.set_calls[0][1].token_type == "Bearer" + agent.dispose() diff --git a/tests/test_oauth_auto_custom_idp_failures.py b/tests/test_oauth_auto_custom_idp_failures.py new file mode 100644 index 00000000..fe3afdb1 --- /dev/null +++ b/tests/test_oauth_auto_custom_idp_failures.py @@ -0,0 +1,285 @@ +"""Failure-mode tests for custom IdP OAuth auto verification.""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request +from typing import Dict, List, Optional + +import pytest + +import gopher_mcp_python.agent as agent_module +import gopher_mcp_python.oauth_resolver as oauth_resolver +from gopher_mcp_python import GopherAgent +from gopher_mcp_python.runtime_options import GopherAgentTokenRecord +from tests.helpers.custom_oauth_test_idp import ( + OAUTH_TEST_ACCESS_TOKEN, + OAUTH_TEST_CLIENT_ID, + OAUTH_TEST_CLIENT_SECRET, + OAUTH_TEST_REFRESH_TOKEN, + start_custom_oauth_test_idp, +) +from tests.helpers.custom_protected_mcp_endpoints import ( + CustomProtectedMcpEndpointsOptions, + start_custom_protected_mcp_endpoints, +) +from tests.helpers.oauth_test_token import refresh_test_oauth_token + + +PROVIDER = "AnthropicProvider" +MODEL = "test-model" +FIXTURE_SECRETS = [ + OAUTH_TEST_CLIENT_SECRET, + OAUTH_TEST_REFRESH_TOKEN, + OAUTH_TEST_ACCESS_TOKEN, +] + + +class _FakeLibrary: + def __init__(self) -> None: + self.calls: List[tuple] = [] + + def agent_create_by_url(self, provider, model, url, runtime_options=None): + self.calls.append(("url", provider, model, url, runtime_options)) + return 5001 + + def agent_release(self, handle): + self.calls.append(("release", handle)) + + def get_last_error_message(self): + return None + + def clear_error(self): + self.calls.append(("clear_error",)) + + +class _RefreshTokenStore: + def __init__(self, refresh_token: str) -> None: + self.refresh_token = refresh_token + self.deleted_keys: List[str] = [] + + async def get(self, key: str) -> Optional[GopherAgentTokenRecord]: + return GopherAgentTokenRecord( + access_token="expired-access-token", + refresh_token=self.refresh_token, + token_type="Bearer", + expires_at=0, + ) + + async def set(self, key: str, token: GopherAgentTokenRecord) -> None: + return None + + async def delete(self, key: str) -> None: + self.deleted_keys.append(key) + + +def test_wrong_refresh_token_returns_secret_safe_invalid_grant_failure() -> None: + idp = start_custom_oauth_test_idp() + try: + _expect_failure_without_fixture_secrets( + lambda: refresh_test_oauth_token( + token_endpoint=idp.token_endpoint, + client_id=OAUTH_TEST_CLIENT_ID, + client_secret=OAUTH_TEST_CLIENT_SECRET, + refresh_token="wrong-refresh-token", + ), + "invalid_grant", + ) + finally: + idp.close() + + +def test_wrong_client_credentials_return_secret_safe_invalid_client_failure() -> None: + idp = start_custom_oauth_test_idp() + try: + _expect_failure_without_fixture_secrets( + lambda: refresh_test_oauth_token( + token_endpoint=idp.token_endpoint, + client_id="wrong-client", + client_secret=OAUTH_TEST_CLIENT_SECRET, + refresh_token=OAUTH_TEST_REFRESH_TOKEN, + ), + "invalid_client", + ) + finally: + idp.close() + + +def test_unsupported_grant_type_returns_clear_oauth_failure() -> None: + idp = start_custom_oauth_test_idp() + try: + response = _post_form( + idp.token_endpoint, + { + "grant_type": "client_credentials", + "client_id": OAUTH_TEST_CLIENT_ID, + "client_secret": OAUTH_TEST_CLIENT_SECRET, + }, + ) + finally: + idp.close() + + assert response.status == 400 + assert response.body == {"error": "unsupported_grant_type"} + + +def test_missing_protected_resource_metadata_fields_fail_clearly(monkeypatch) -> None: + fake = _install_fake_library(monkeypatch) + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + protected_resource_metadata={"resource": "missing-authorization-servers"}, + ) + ) + try: + _expect_failure_without_fixture_secrets( + lambda: GopherAgent.create_with_url( + PROVIDER, + MODEL, + endpoints.server.mcp_url, + ), + "authorization_servers", + ) + finally: + endpoints.close() + idp.close() + oauth_resolver.set_oauth_resolver_hooks_for_test() + oauth_resolver.set_oauth_url_runtime_options_resolver_for_test() + + assert fake.calls == [] + + +def test_wrong_bearer_token_is_rejected_by_protected_endpoint() -> None: + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + try: + response = _post_json( + endpoints.server.mcp_url, + {"jsonrpc": "2.0", "id": "wrong-token"}, + access_token="wrong-token", + ) + finally: + endpoints.close() + idp.close() + + assert response.status == 401 + assert response.headers["WWW-Authenticate"].endswith( + f'{endpoints.server.resource_metadata_url}"' + ) + + +def test_sdk_refresh_failure_stays_secret_safe_before_fallback_failure( + monkeypatch, +) -> None: + _install_fake_library(monkeypatch) + idp = start_custom_oauth_test_idp() + endpoints = start_custom_protected_mcp_endpoints( + CustomProtectedMcpEndpointsOptions( + authorization_server=idp.issuer, + access_token=OAUTH_TEST_ACCESS_TOKEN, + ) + ) + token_store = _RefreshTokenStore("wrong-refresh-token") + + def open_authorization_url(url, open_browser=None): + raise RuntimeError("authorization fallback disabled for failure test") + + monkeypatch.setattr(oauth_resolver, "open_authorization_url", open_authorization_url) + + try: + _expect_failure_without_fixture_secrets( + lambda: GopherAgent.create_with_url( + PROVIDER, + MODEL, + endpoints.server.mcp_url, + { + "oauth": { + "token_store": token_store, + }, + }, + ), + "authorization fallback disabled for failure test", + ) + finally: + endpoints.close() + idp.close() + oauth_resolver.set_oauth_resolver_hooks_for_test() + oauth_resolver.set_oauth_url_runtime_options_resolver_for_test() + + assert token_store.deleted_keys + + +def _install_fake_library(monkeypatch) -> _FakeLibrary: + fake = _FakeLibrary() + monkeypatch.setattr(agent_module, "_initialized", True) + monkeypatch.setattr( + agent_module.GopherOrchLibrary, + "get_instance", + staticmethod(lambda: fake), + ) + return fake + + +def _expect_failure_without_fixture_secrets(call, expected_message: str) -> None: + with pytest.raises(Exception) as exc_info: + call() + + message = str(exc_info.value) + assert expected_message in message + for secret in FIXTURE_SECRETS: + assert secret not in message + + +class _Response: + def __init__(self, status: int, headers: dict, body: object = None) -> None: + self.status = status + self.headers = headers + self.body = body + + +def _post_form(url: str, params: Dict[str, str]) -> _Response: + data = urllib.parse.urlencode(params).encode("utf-8") + request = urllib.request.Request( + url, + data=data, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + return _send_request(request) + + +def _post_json(url: str, body: object, access_token: str = "") -> _Response: + headers = {"Content-Type": "application/json"} + if access_token: + headers["Authorization"] = f"Bearer {access_token}" + request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8"), + method="POST", + headers=headers, + ) + return _send_request(request) + + +def _send_request(request: urllib.request.Request) -> _Response: + try: + with urllib.request.urlopen(request, timeout=5) as response: + body = response.read() + parsed_body = json.loads(body.decode("utf-8")) if body else None + return _Response(response.status, dict(response.headers), parsed_body) + except urllib.error.HTTPError as exc: + try: + body = exc.read() + except ConnectionError: + body = b"" + parsed_body = json.loads(body.decode("utf-8")) if body else None + return _Response(exc.code, dict(exc.headers), parsed_body) diff --git a/tests/test_oauth_test_token_helper.py b/tests/test_oauth_test_token_helper.py new file mode 100644 index 00000000..aa87650c --- /dev/null +++ b/tests/test_oauth_test_token_helper.py @@ -0,0 +1,184 @@ +"""Tests for generic test OAuth token helper.""" + +import json +import urllib.parse +from http.server import BaseHTTPRequestHandler + +import pytest + +from tests.helpers.oauth_test_token import refresh_test_oauth_token +from tests.test_oauth_discovery import _start_server + + +CLIENT_ID = "test-client" +CLIENT_SECRET = "test-secret" +REFRESH_TOKEN = "test-refresh-token" +ACCESS_TOKEN = "test-access-token" + + +def test_refresh_test_oauth_token_posts_refresh_grant() -> None: + captured = {} + + def handle(handler: BaseHTTPRequestHandler) -> None: + length = int(handler.headers.get("Content-Length", "0")) + body = handler.rfile.read(length).decode("utf-8") + captured["content_type"] = handler.headers.get("Content-Type") + captured["params"] = urllib.parse.parse_qs(body) + _json( + handler, + 200, + { + "access_token": ACCESS_TOKEN, + "token_type": "Bearer", + "expires_in": 3600, + "scope": "openid profile email", + }, + ) + + server = _start_server(handle) + try: + token = refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + assert token.access_token == ACCESS_TOKEN + assert token.token_type == "Bearer" + assert token.expires_in == 3600 + assert token.scope == "openid profile email" + assert captured["content_type"] == "application/x-www-form-urlencoded" + assert captured["params"]["grant_type"] == ["refresh_token"] + assert captured["params"]["client_id"] == [CLIENT_ID] + assert captured["params"]["client_secret"] == [CLIENT_SECRET] + assert captured["params"]["refresh_token"] == [REFRESH_TOKEN] + + +@pytest.mark.parametrize( + ("error", "status"), + [ + ("invalid_grant", 400), + ("invalid_client", 401), + ("unsupported_grant_type", 400), + ], +) +def test_refresh_test_oauth_token_surfaces_oauth_errors_without_secrets( + error: str, status: int +) -> None: + server = _start_server( + lambda handler: _json( + handler, + status, + { + "error": error, + "error_description": ( + f"{CLIENT_SECRET} {REFRESH_TOKEN} {ACCESS_TOKEN}" + ), + }, + ) + ) + try: + with pytest.raises(RuntimeError) as exc_info: + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + message = str(exc_info.value) + assert error in message + assert CLIENT_SECRET not in message + assert REFRESH_TOKEN not in message + assert ACCESS_TOKEN not in message + + +def test_refresh_test_oauth_token_surfaces_non_oauth_http_error() -> None: + server = _start_server(lambda handler: _json(handler, 500, {})) + try: + with pytest.raises(RuntimeError, match="HTTP 500"): + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + +def test_refresh_test_oauth_token_requires_access_token() -> None: + server = _start_server( + lambda handler: _json(handler, 200, {"token_type": "Bearer"}) + ) + try: + with pytest.raises(RuntimeError, match="missing access_token"): + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + +def test_refresh_test_oauth_token_requires_token_type() -> None: + server = _start_server( + lambda handler: _json(handler, 200, {"access_token": ACCESS_TOKEN}) + ) + try: + with pytest.raises(RuntimeError, match="missing token_type"): + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + +def test_refresh_test_oauth_token_rejects_invalid_json() -> None: + server = _start_server(lambda handler: _raw(handler, 200, b"not json")) + try: + with pytest.raises(RuntimeError, match="invalid JSON"): + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + +def test_refresh_test_oauth_token_rejects_non_object_json() -> None: + server = _start_server(lambda handler: _raw(handler, 200, b"[]")) + try: + with pytest.raises(RuntimeError, match="invalid JSON"): + refresh_test_oauth_token( + token_endpoint=f"{server.url}/token", + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + refresh_token=REFRESH_TOKEN, + ) + finally: + server.close() + + +def _json(handler: BaseHTTPRequestHandler, status: int, body: object) -> None: + _raw(handler, status, json.dumps(body).encode("utf-8")) + + +def _raw(handler: BaseHTTPRequestHandler, status: int, body: bytes) -> None: + handler.send_response(status) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body)