|
| 1 | +"""Optional JWT crypto packaging must preserve asymmetric OAuth semantics.""" |
| 2 | + |
| 3 | +import subprocess |
| 4 | +import sys |
| 5 | +from pathlib import Path |
| 6 | + |
| 7 | +import jwt |
| 8 | +import pytest |
| 9 | +import tomllib |
| 10 | + |
| 11 | +from mcp.client.auth.extensions.client_credentials import JWTParameters, SignedJWTParameters |
| 12 | + |
| 13 | + |
| 14 | +def test_crypto_is_an_explicit_extra() -> None: |
| 15 | + project = tomllib.loads((Path(__file__).resolve().parents[4] / "pyproject.toml").read_text())["project"] |
| 16 | + assert "pyjwt>=2.10.1" in project["dependencies"] |
| 17 | + assert not any("[crypto]" in requirement for requirement in project["dependencies"]) |
| 18 | + assert project["optional-dependencies"]["oauth"] == ["pyjwt[crypto]>=2.10.1"] |
| 19 | + |
| 20 | + |
| 21 | +def test_stdio_and_asymmetric_failure_without_crypto() -> None: |
| 22 | + # Exercise PyJWT's real no-backend behavior in a clean subprocess, without |
| 23 | + # altering its algorithms or replacing signing/verification implementations. |
| 24 | + script = r""" |
| 25 | +import anyio |
| 26 | +import importlib.abc |
| 27 | +import sys |
| 28 | +class NoCrypto(importlib.abc.MetaPathFinder): |
| 29 | + def find_spec(self, fullname, path=None, target=None): |
| 30 | + if fullname == 'cryptography' or fullname.startswith('cryptography.'): |
| 31 | + raise ModuleNotFoundError('crypto unavailable in this fixture', name=fullname) |
| 32 | +sys.meta_path.insert(0, NoCrypto()) |
| 33 | +from mcp import ClientSession, StdioServerParameters, stdio_client |
| 34 | +from mcp.server.fastmcp import FastMCP |
| 35 | +from mcp.client.auth.extensions.client_credentials import JWTParameters, SignedJWTParameters |
| 36 | +import jwt |
| 37 | +assert not jwt.algorithms.has_crypto |
| 38 | +params = SignedJWTParameters(issuer='client', subject='client', signing_key='fixture-only') |
| 39 | +try: |
| 40 | + anyio.run(params.create_assertion_provider(), 'https://auth.example/') |
| 41 | +except NotImplementedError as error: |
| 42 | + assert 'RS256' in str(error) |
| 43 | +else: |
| 44 | + raise AssertionError('missing backend did not reject default RS256') |
| 45 | +params = JWTParameters(issuer='client', subject='client', audience='https://auth.example/', |
| 46 | + jwt_signing_key='fixture-only') |
| 47 | +try: |
| 48 | + params.to_assertion() |
| 49 | +except NotImplementedError as error: |
| 50 | + assert 'RS256' in str(error) |
| 51 | +else: |
| 52 | + raise AssertionError('legacy default RS256 silently fell back') |
| 53 | +assert JWTParameters(assertion='opaque-prebuilt-fixture').to_assertion() == 'opaque-prebuilt-fixture' |
| 54 | +""" |
| 55 | + subprocess.run([sys.executable, "-c", script], check=True, timeout=30) |
| 56 | + |
| 57 | + |
| 58 | +@pytest.mark.anyio |
| 59 | +@pytest.mark.parametrize("legacy", [False, True]) |
| 60 | +async def test_rs256_assertions_verify_and_reject_tampering(legacy: bool) -> None: |
| 61 | + from cryptography.hazmat.primitives import serialization |
| 62 | + from cryptography.hazmat.primitives.asymmetric import rsa |
| 63 | + from jwt.utils import base64url_decode, base64url_encode |
| 64 | + |
| 65 | + # Ephemeral fixture key; never written to disk or included in logs. |
| 66 | + private = rsa.generate_private_key(public_exponent=65537, key_size=2048) |
| 67 | + pem = private.private_bytes( |
| 68 | + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() |
| 69 | + ).decode() |
| 70 | + audience = "https://auth.example/" |
| 71 | + if legacy: |
| 72 | + token = JWTParameters(issuer="client", subject="client", audience=audience, jwt_signing_key=pem).to_assertion() |
| 73 | + else: |
| 74 | + params = SignedJWTParameters(issuer="client", subject="client", signing_key=pem) |
| 75 | + token = await params.create_assertion_provider()(audience) |
| 76 | + assert jwt.get_unverified_header(token)["alg"] == "RS256" |
| 77 | + options = {"require": ["iss", "sub", "aud", "exp", "iat", "jti"]} |
| 78 | + claims = jwt.decode( |
| 79 | + token, private.public_key(), algorithms=["RS256"], issuer="client", audience=audience, options=options |
| 80 | + ) |
| 81 | + assert claims["sub"] == "client" |
| 82 | + assert claims["exp"] - claims["iat"] == 300 |
| 83 | + assert claims["jti"] |
| 84 | + with pytest.raises(jwt.InvalidAudienceError): |
| 85 | + jwt.decode( |
| 86 | + token, |
| 87 | + private.public_key(), |
| 88 | + algorithms=["RS256"], |
| 89 | + issuer="client", |
| 90 | + audience="https://other.example/", |
| 91 | + options=options, |
| 92 | + ) |
| 93 | + header, payload, signature = token.split(".") |
| 94 | + corrupted = bytearray(base64url_decode(signature)) |
| 95 | + corrupted[0] ^= 1 |
| 96 | + forged = ".".join((header, payload, base64url_encode(bytes(corrupted)).decode())) |
| 97 | + with pytest.raises(jwt.InvalidSignatureError): |
| 98 | + jwt.decode( |
| 99 | + forged, private.public_key(), algorithms=["RS256"], issuer="client", audience=audience, options=options |
| 100 | + ) |
| 101 | + with pytest.raises(jwt.InvalidAlgorithmError): |
| 102 | + jwt.decode(token, "fixture-only", algorithms=["HS256"]) |
0 commit comments