Adding OAuth to a remote MCP server, and five checks that catch the usual bugs

HTTP 401 WWW-Authenticate: Bearer resource_metadata="http://127.0.0.1:8080/.well-known/oauth-protected-resource/mcp", scope="mcp:tools" 

{ "resource": "http://127.0.0.1:8080/mcp", "authorization_servers": [ "http://127.0.0.1:8081" ], "bearer_methods_supported": [ "header" ], "scopes_supported": [ "mcp:tools", "mcp:write" ] } 

{ "issuer": "http://127.0.0.1:8081", "authorization_endpoint": "http://127.0.0.1:8081/authorize", "token_endpoint": "http://127.0.0.1:8081/token", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code", "refresh_token"], "code_challenge_methods_supported": ["S256"], "token_endpoint_auth_methods_supported": ["none"], "authorization_response_iss_parameter_supported": true, "scopes_supported": ["mcp:tools", "mcp:write"], "resource_indicators_supported": true } 

HTTP 403 WWW-Authenticate: Bearer resource_metadata="http://127.0.0.1:8080/.well-known/oauth-protected-resource/mcp", scope="mcp:write", error="insufficient_scope" 

alg = header.get("alg") if alg != "HS256": raise ValueError("invalid_alg") # ... compare the HMAC, then: if payload.get("iss") != issuer: raise ValueError("invalid_issuer") if payload.get("aud") != audience: raise ValueError("invalid_audience") if not isinstance(payload.get("exp"), int) or payload["exp"] <= now: raise ValueError("expired") 

#!/usr/bin/env python3 """Check a remote MCP server's OAuth 2.1 surface. This does not replace a security audit. It fetches the 401 challenge, Protected Resource Metadata, authorization server metadata, and optional token behaviour, then prints what a client would trip over. Exit 0 if no errors (warnings are allowed). Exit 1 if any error is flagged. Standard library only. Usage: python3 check_mcp_oauth.py http://127.0.0.1:8080/mcp python3 check_mcp_oauth.py http://127.0.0.1:8080/mcp --token "$ACCESS_TOKEN" python3 check_mcp_oauth.py http://127.0.0.1:8080/mcp --json """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.parse import urllib.request from typing import Any USER_AGENT = "check_mcp_oauth/0.1 (stdlib tutorial checker)" class Report: def __init__(self) -> None: self.errors: list[str] = [] self.warnings: list[str] = [] self.notes: list[str] = [] def error(self, msg: str) -> None: self.errors.append(msg) def warn(self, msg: str) -> None: self.warnings.append(msg) def note(self, msg: str) -> None: self.notes.append(msg) def as_dict(self) -> dict[str, Any]: return {"errors": self.errors, "warnings": self.warnings, "notes": self.notes} def http( url: str, *, method: str = "GET", headers: dict[str, str] | None = None, data: bytes | None = None, timeout: float = 8.0, ) -> tuple[int, dict[str, str], bytes]: hdrs = {"User-Agent": USER_AGENT, "Accept": "application/json"} if headers: hdrs.update(headers) req = urllib.request.Request(url, data=data, headers=hdrs, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, {k.lower(): v for k, v in resp.headers.items()}, resp.read() except urllib.error.HTTPError as exc: return exc.code, {k.lower(): v for k, v in exc.headers.items()}, exc.read() def parse_www_authenticate(value: str) -> dict[str, str]: """Parse a single Bearer WWW-Authenticate challenge into params.""" out: dict[str, str] = {} text = value.strip() if text.lower().startswith("bearer"): text = text[6:].strip() out["scheme"] = "Bearer" # Split on commas that are not inside quotes. buf = "" in_quote = False parts: list[str] = [] for ch in text: if ch == '"': in_quote = not in_quote buf += ch elif ch == "," and not in_quote: parts.append(buf.strip()) buf = "" else: buf += ch if buf.strip(): parts.append(buf.strip()) for part in parts: if "=" not in part: continue key, val = part.split("=", 1) out[key.strip()] = val.strip().strip('"') return out def json_body(raw: bytes) -> Any | None: try: return json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return None def path_inserted_well_known(resource: str, name: str) -> list[str]: parsed = urllib.parse.urlparse(resource) path = parsed.path or "" root = f"{parsed.scheme}://{parsed.netloc}/.well-known/{name}" urls = [root] if path not in ("", "/"): urls.insert(0, f"{parsed.scheme}://{parsed.netloc}/.well-known/{name}{path}") urls.append(f"{parsed.scheme}://{parsed.netloc}{path}/.well-known/{name}") return urls def fetch_json_first(urls: list[str], report: Report, label: str) -> tuple[str | None, Any | None]: for url in urls: status, _, raw = http(url) if status == 200: body = json_body(raw) if isinstance(body, dict): report.note(f"{label} at {url}") return url, body report.error(f"{label} at {url} returned 200 but not JSON object") return url, None report.error(f"{label} not found at: {', '.join(urls)}") return None, None def check_server(resource: str, token: str | None, report: Report) -> None: parsed = urllib.parse.urlparse(resource) if parsed.scheme not in {"http", "https"} or not parsed.netloc: report.error("resource URL must be an absolute http(s) URI") return if parsed.scheme == "http" and parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: report.warn("MCP authorization is specified over HTTPS; HTTP is only reasonable on loopback") init = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2026-07-28", "capabilities": {}, "clientInfo": {"name": "checker", "version": "0.1"}}, }).encode() status, headers, raw = http( resource, method="POST", headers={"Content-Type": "application/json"}, data=init, ) if status != 401: report.error(f"unauthenticated initialize returned {status}, expected 401") if status == 200: report.error("a 200 here usually means the server is open, or it redirected to a login page the client will not follow") return www = headers.get("www-authenticate", "") if not www: report.error("401 is missing WWW-Authenticate") return challenge = parse_www_authenticate(www) report.note(f"401 WWW-Authenticate: {www}") if challenge.get("scheme") != "Bearer": report.error("WWW-Authenticate is not a Bearer challenge") meta_url = challenge.get("resource_metadata") if not meta_url: report.error('WWW-Authenticate is missing resource_metadata="..."') if "scope" not in challenge: report.warn("WWW-Authenticate has no scope parameter; clients will fall back to scopes_supported") else: report.note(f"challenge scope: {challenge['scope']}") prm_urls = [] if meta_url: prm_urls.append(meta_url) for url in path_inserted_well_known(resource, "oauth-protected-resource"): if url not in prm_urls: prm_urls.append(url) found_prm, prm = fetch_json_first(prm_urls, report, "protected resource metadata") if not isinstance(prm, dict): return if prm.get("resource") != resource: report.error( f"PRM resource {prm.get('resource')!r} does not equal the server URL {resource!r}. " "MCP clients reject a mismatch; this is a common bug behind a reverse proxy." ) else: report.note("PRM resource matches the server URL exactly") auth_servers = prm.get("authorization_servers") if not isinstance(auth_servers, list) or not auth_servers: report.error("PRM is missing authorization_servers") return issuer = auth_servers[0] if not isinstance(issuer, str): report.error("authorization_servers[0] is not a string") return as_urls = path_inserted_well_known(issuer, "oauth-authorization-server") # RFC 8414 also allows OpenID Discovery. as_urls.extend(path_inserted_well_known(issuer, "openid-configuration")) _, meta = fetch_json_first(as_urls, report, "authorization server metadata") if not isinstance(meta, dict): return if meta.get("issuer") != issuer: report.error(f"AS metadata issuer {meta.get('issuer')!r} does not match {issuer!r}") else: report.note("authorization server issuer matches") for field in ("authorization_endpoint", "token_endpoint"): if not meta.get(field): report.error(f"AS metadata is missing {field}") methods = meta.get("code_challenge_methods_supported") or [] if "S256" not in methods: report.error("code_challenge_methods_supported does not include S256; MCP clients stop here") else: report.note("PKCE S256 is advertised") if meta.get("authorization_response_iss_parameter_supported") is True: report.note("authorization_response_iss_parameter_supported is true") else: report.warn(Source：https://dev.to/cryptomoneymakr/adding-oauth-to-a-remote-mcp-server-and-five-checks-that-catch-the-usual-bugs-34ol
My favorites