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("authorization_response_iss_parameter_supported is not true; clients still should emit and check iss")

    if token:
        check_token(resource, token, init, report)
    else:
        report.note("no --token; skipped token handling checks")


def check_token(resource: str, token: str, init: bytes, report: Report) -> None:
    def call(headers: dict[str, str], url: str | None = None) -> tuple[int, dict[str, str], Any | None]:
        status, hdrs, raw = http(
            url or resource,
            method="POST",
            headers={"Content-Type": "application/json", **headers},
            data=init,
        )
        return status, hdrs, json_body(raw)

    status, _, body = call({"Authorization": f"Bearer {token}"})
    if status != 200:
        report.error(f"valid --token was rejected with {status}: {body}")
    else:
        report.note("valid --token was accepted on initialize")

    status, hdrs, body = call({"Authorization": "Bearer not-a-real-token"})
    if status != 401:
        report.error(f"garbage token returned {status}, expected 401")
    else:
        challenge = parse_www_authenticate(hdrs.get("www-authenticate", ""))
        if challenge.get("error") and challenge.get("error") != "invalid_token":
            report.warn(f"garbage token 401 error={challenge.get('error')!r} (expected invalid_token)")
        else:
            report.note("garbage token returned 401")

    none_token = "eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJzdWIiOiJhdHRhY2tlciJ9."
    status, _, _ = call({"Authorization": f"Bearer {none_token}"})
    if status == 200:
        report.error("an alg=none JWT was accepted; the server must reject it")
    else:
        report.note(f"alg=none JWT was rejected with {status}")

    # Token in the query string must not work.
    joiner = "&" if "?" in resource else "?"
    status, _, _ = call({}, url=f"{resource}{joiner}access_token={urllib.parse.quote(token)}")
    if status == 200:
        report.error("the access token was accepted from the query string; MCP forbids this")
    else:
        report.note(f"query-string token was rejected with {status}")


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("resource", help="Canonical MCP server URL, e.g. https://mcp.example.com/mcp")
    p.add_argument("--token", help="Optional access token to test token handling")
    p.add_argument("--json", action="store_true", help="Print machine-readable output")
    args = p.parse_args(argv)
    report = Report()
    # Do not strip a trailing slash: that is a common resource mismatch.
    check_server(args.resource, args.token, report)
    if args.json:
        print(json.dumps(report.as_dict(), indent=2))
    else:
        for msg in report.notes:
            print(f"OK: {msg}")
        for msg in report.warnings:
            print(f"WARN: {msg}")
        for msg in report.errors:
            print(f"ERROR: {msg}")
        if report.errors:
            print(f"\n{len(report.errors)} error(s), {len(report.warnings)} warning(s)")
        else:
            print("\nOK: no errors flagged" + (f" ({len(report.warnings)} warning(s))" if report.warnings else ""))
    return 1 if report.errors else 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 check_mcp_oauth.py http://127.0.0.1:8080/mcp

OK: 401 WWW-Authenticate: Bearer resource_metadata="http://127.0.0.1:8080/.well-known/oauth-protected-resource/mcp", scope="mcp:tools"
OK: challenge scope: mcp:tools
OK: protected resource metadata at http://127.0.0.1:8080/.well-known/oauth-protected-resource/mcp
OK: PRM resource matches the server URL exactly
OK: authorization server metadata at http://127.0.0.1:8081/.well-known/oauth-authorization-server
OK: authorization server issuer matches
OK: PKCE S256 is advertised
OK: authorization_response_iss_parameter_supported is true
OK: no --token; skipped token handling checks

OK: no errors flagged

python3 check_mcp_oauth.py http://127.0.0.1:8080/mcp --token "$ACCESS_TOKEN"

OK: valid --token was accepted on initialize
OK: garbage token returned 401
OK: alg=none JWT was rejected with 401
OK: query-string token was rejected with 401

OK: no errors flagged

ERROR: PRM resource 'http://127.0.0.1:8082/mcp/' does not equal the server URL 'http://127.0.0.1:8082/mcp'. MCP clients reject a mismatch; this is a common bug behind a reverse proxy.

1 error(s), 0 warning(s)