From 28c1863ee30463754e1b04b6e0db3b4d41c386d3 Mon Sep 17 00:00:00 2001 From: Eric Gustin <34000337+EricGustin@users.noreply.github.com> Date: Fri, 16 Jan 2026 15:55:05 -0800 Subject: [PATCH] Support Ed25519 Algorithm (#742) Ed25519 is needed for Arcade AS. This required migrating from `python-jose` to `joserfc`, because `python-jose` didn't seem to support Ed25519 --- .../resource_server/validators/jwks.py | 168 +++-- libs/arcade-mcp-server/pyproject.toml | 4 +- .../test_resource_server_auth.py | 577 ++++++++++++------ 3 files changed, 518 insertions(+), 231 deletions(-) diff --git a/libs/arcade-mcp-server/arcade_mcp_server/resource_server/validators/jwks.py b/libs/arcade-mcp-server/arcade_mcp_server/resource_server/validators/jwks.py index de018dd4..7311ff00 100644 --- a/libs/arcade-mcp-server/arcade_mcp_server/resource_server/validators/jwks.py +++ b/libs/arcade-mcp-server/arcade_mcp_server/resource_server/validators/jwks.py @@ -4,11 +4,16 @@ JWKS-based token validator for MCP Resource Servers. Implements OAuth 2.1 Resource Server token validation using JWT with JWKS. """ +import binascii import time from typing import Any, cast import httpx -from jose import jwk, jwt +from joserfc import jws, jwt +from joserfc.errors import JoseError +from joserfc.jwk import KeySet, KeySetSerialization +from joserfc.jws import JWSRegistry +from joserfc.registry import HeaderParameter from arcade_mcp_server.resource_server.base import ( AccessTokenValidationOptions, @@ -19,19 +24,36 @@ from arcade_mcp_server.resource_server.base import ( TokenExpiredError, ) -# Note: Only asymmetric algorithms supported SUPPORTED_ALGORITHMS = { + # RSA "RS256", "RS384", "RS512", + # ECDSA "ES256", "ES384", "ES512", + # RSA-PSS "PS256", "PS384", "PS512", + # EdDSA + "Ed25519", + "EdDSA", } +# EdDSA algorithm aliases +EDDSA_ALGORITHMS = {"Ed25519", "EdDSA"} + +# Custom JWS registry that allows additional header params beyond joserfc's default +_ACCESS_TOKEN_REGISTRY = JWSRegistry( + header_registry={ + "iss": HeaderParameter("Issuer", "str"), + "aud": HeaderParameter("Audience", "str"), + }, + algorithms=list(SUPPORTED_ALGORITHMS), +) + class JWKSTokenValidator(ResourceServerValidator): """JWKS-based JWT token validator for simple, explicit token validation. @@ -114,6 +136,31 @@ class JWKSTokenValidator(ResourceServerValidator): self._jwks_cache: dict[str, Any] | None = None self._cache_timestamp: float = 0 + def _normalize_algorithm(self, alg: str) -> str: + """Normalize algorithm name for comparison. + + EdDSA has multiple names (EdDSA, Ed25519) that should be treated as equivalent. + + Args: + alg: Algorithm name + + Returns: + Normalized algorithm name + """ + return "Ed25519" if alg in EDDSA_ALGORITHMS else alg + + def _algorithms_match(self, alg1: str, alg2: str) -> bool: + """Check if two algorithm names match (considering EdDSA aliases). + + Args: + alg1: First algorithm name + alg2: Second algorithm name + + Returns: + True if algorithms match + """ + return self._normalize_algorithm(alg1) == self._normalize_algorithm(alg2) + async def _fetch_jwks(self) -> dict[str, Any]: """Fetch JWKS with caching. @@ -139,6 +186,25 @@ class JWKSTokenValidator(ResourceServerValidator): else: return self._jwks_cache + def _get_headers_without_verification(self, token: str) -> dict[str, Any]: + """Extract header from JWT without verification. + + Args: + token: JWT token string + + Returns: + Header dictionary containing 'alg', 'kid', etc. + + Raises: + InvalidTokenError: If token format is invalid + """ + try: + # Use joserfc's extract_compact to parse JWT without verification + obj = jws.extract_compact(token.encode()) + return obj.headers() + except (JoseError, ValueError, binascii.Error) as e: + raise InvalidTokenError(f"Invalid JWT format: {e}") from e + def _find_signing_key(self, jwks: dict[str, Any], token: str) -> Any: """Find the signing key from JWKS that matches the token's kid. @@ -147,74 +213,92 @@ class JWKSTokenValidator(ResourceServerValidator): token: JWT token Returns: - Signing key in PEM format + Key object from joserfc KeySet Raises: InvalidTokenError: If no matching key found or algorithm mismatch """ - unverified_header = jwt.get_unverified_header(token) - kid = unverified_header.get("kid") - token_alg = unverified_header.get("alg") + header = self._get_headers_without_verification(token) + kid = header.get("kid") + token_alg = header.get("alg") # Validate token algorithm matches configuration (prevent algorithm confusion) - if token_alg and token_alg != self.algorithm: + if token_alg and not self._algorithms_match(token_alg, self.algorithm): raise InvalidTokenError( f"Token algorithm '{token_alg}' doesn't match " f"configured algorithm '{self.algorithm}'" ) - for key_data in jwks.get("keys", []): - if key_data.get("kid") == kid: - key_alg = key_data.get("alg") + try: + key_set = KeySet.import_key_set(cast(KeySetSerialization, jwks)) + except Exception as e: + raise InvalidTokenError(f"Failed to import JWKS: {e}") from e - if key_alg and key_alg != self.algorithm: + # Find key by kid + for key in key_set.keys: + if key.kid == kid: + key_alg = key.alg + if key_alg and not self._algorithms_match(key_alg, self.algorithm): raise InvalidTokenError( f"Key algorithm '{key_alg}' doesn't match " f"configured algorithm '{self.algorithm}'" ) - - key_obj = jwk.construct(key_data, algorithm=self.algorithm) - return key_obj.to_pem().decode("utf-8") + return key raise InvalidTokenError("No matching key found in JWKS") - def _decode_token(self, token: str, signing_key: str) -> dict[str, Any]: + def _decode_token(self, token: str, signing_key: Any) -> dict[str, Any]: """Decode and verify the provided JWT token. + Uses jwt.decode for signature verification and payload parsing, then + performs time-based claims validation manually for leeway handling + Args: token: JWT token - signing_key: Public key in PEM format + signing_key: Key object from joserfc Returns: Decoded token claims Raises: - jwt.ExpiredSignatureError: Token has expired - jwt.JWTClaimsError: Token claims validation failed (audience/issuer mismatch) - jwt.JWTError: Token is invalid + TokenExpiredError: Token has expired + InvalidTokenError: Token validation failed """ - decode_options = { - "verify_signature": True, # Always verify signature. Cannot be disabled. - "verify_exp": self.validation_options.verify_exp, - "verify_iat": self.validation_options.verify_iat, - "verify_nbf": self.validation_options.verify_nbf, - "verify_aud": False, # Manual validation for multi-audience support - "verify_iss": False, # Manual validation for multi-issuer support - "leeway": self.validation_options.leeway, - } + if self.algorithm in EDDSA_ALGORITHMS: + algorithms = list(EDDSA_ALGORITHMS) + else: + algorithms = [self.algorithm] - # Decode token once without aud/iss validation - decoded = cast( - dict[str, Any], - jwt.decode( - token, - signing_key, - algorithms=[self.algorithm], - options=decode_options, - ), - ) + # First, verify signature & decode payload + try: + result = jwt.decode( + token, signing_key, algorithms=algorithms, registry=_ACCESS_TOKEN_REGISTRY + ) + except JoseError as e: + raise InvalidTokenError(f"Token signature verification failed: {e}") from e - # Manually validate issuer (if flag is enabled) + decoded = result.claims + + # Validate time based claims with leeway support + current_time = int(time.time()) + leeway = self.validation_options.leeway + + if self.validation_options.verify_exp: + exp = decoded.get("exp") + if exp is not None and exp + leeway < current_time: + raise TokenExpiredError("Token has expired") + + if self.validation_options.verify_iat: + iat = decoded.get("iat") + if iat is not None and iat - leeway > current_time: + raise InvalidTokenError("Token issued in the future") + + if self.validation_options.verify_nbf: + nbf = decoded.get("nbf") + if nbf is not None and nbf - leeway > current_time: + raise InvalidTokenError("Token not yet valid") + + # Manually validate issuer (if enabled) if self.validation_options.verify_iss: token_iss = decoded.get("iss") if isinstance(self.issuer, list): @@ -313,12 +397,6 @@ class JWKSTokenValidator(ResourceServerValidator): claims=decoded, ) - except jwt.ExpiredSignatureError as e: - raise TokenExpiredError("Token has expired") from e - except jwt.JWTClaimsError as e: - raise InvalidTokenError(f"Token claims validation failed: {e}") from e - except jwt.JWTError as e: - raise InvalidTokenError(f"Invalid token: {e}") from e except (InvalidTokenError, TokenExpiredError): raise except Exception as e: diff --git a/libs/arcade-mcp-server/pyproject.toml b/libs/arcade-mcp-server/pyproject.toml index ca8a2bb5..67be5f96 100644 --- a/libs/arcade-mcp-server/pyproject.toml +++ b/libs/arcade-mcp-server/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "arcade-mcp-server" -version = "1.14.2" +version = "1.15.0" description = "Model Context Protocol (MCP) server framework for Arcade.dev" readme = "README.md" authors = [{ name = "Arcade.dev" }] @@ -34,7 +34,7 @@ dependencies = [ "anyio>=4.0.0", "python-dotenv>=1.0.0", "pydantic-settings>=2.10.1", - "python-jose[cryptography]>=3.3.0,<4.0.0", + "joserfc>=1.5.0", "httpx>=0.27.0,<1.0.0", ] diff --git a/libs/tests/arcade_mcp_server/test_resource_server_auth.py b/libs/tests/arcade_mcp_server/test_resource_server_auth.py index 14445e1b..f94fb0a6 100644 --- a/libs/tests/arcade_mcp_server/test_resource_server_auth.py +++ b/libs/tests/arcade_mcp_server/test_resource_server_auth.py @@ -19,7 +19,9 @@ from arcade_mcp_server.resource_server.middleware import ResourceServerMiddlewar from arcade_mcp_server.worker import create_arcade_mcp from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa -from jose import jwt +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from joserfc import jwt +from joserfc.jwk import OKPKey, RSAKey # Test fixtures @@ -49,11 +51,23 @@ def rsa_keypair(): return private_key, public_key +@pytest.fixture +def rsa_joserfc_key(rsa_keypair): + """Generate joserfc RSAKey from keypair.""" + private_key, _ = rsa_keypair + pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + return RSAKey.import_key(pem) + + @pytest.fixture def serialized_private_key(rsa_keypair): """Generate private key as PEM format for testing.""" private_key, _ = rsa_keypair - # Serialize private key to PEM format for python-jose + # Serialize private key to PEM format pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, @@ -94,11 +108,9 @@ def jwks_data(rsa_keypair): @pytest.fixture -def valid_jwt_token(rsa_keypair): +def valid_jwt_token(rsa_joserfc_key): """Generate valid JWT token for testing.""" - private_key, _ = rsa_keypair - - payload = { + claims = { "sub": "user123", "email": "user@example.com", "iss": "https://auth.example.com", @@ -107,22 +119,16 @@ def valid_jwt_token(rsa_keypair): "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) return token @pytest.fixture -def expired_jwt_token(rsa_keypair): +def expired_jwt_token(rsa_joserfc_key): """Generate expired JWT token for testing.""" - private_key, _ = rsa_keypair - - payload = { + claims = { "sub": "user123", "email": "user@example.com", "iss": "https://auth.example.com", @@ -131,12 +137,92 @@ def expired_jwt_token(rsa_keypair): "iat": int(time.time()) - 7200, } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) + + return token + + +# Ed25519 fixtures +@pytest.fixture +def ed25519_keypair(): + """Generate Ed25519 key pair for testing.""" + private_key = Ed25519PrivateKey.generate() + public_key = private_key.public_key() + return private_key, public_key + + +@pytest.fixture +def ed25519_joserfc_key(ed25519_keypair): + """Generate joserfc OKPKey from Ed25519 keypair.""" + private_key, _ = ed25519_keypair + pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), ) + return OKPKey.import_key(pem) + + +@pytest.fixture +def ed25519_jwks_data(ed25519_keypair): + """Generate Ed25519 JWKS data for testing.""" + _, public_key = ed25519_keypair + + # Get the raw public key bytes + public_bytes = public_key.public_bytes_raw() + + # Base64url encode the public key + x_b64 = base64.urlsafe_b64encode(public_bytes).decode("utf-8").rstrip("=") + + return { + "keys": [ + { + "kty": "OKP", + "kid": "ed25519-key-1", + "use": "sig", + "alg": "Ed25519", + "crv": "Ed25519", + "x": x_b64, + } + ] + } + + +@pytest.fixture +def valid_ed25519_token(ed25519_joserfc_key): + """Generate valid Ed25519 JWT token for testing.""" + claims = { + "sub": "user456", + "email": "ed25519user@example.com", + "iss": "https://cloud.arcade.dev/oauth2", + "aud": "urn:arcade:mcp", + "exp": int(time.time()) + 3600, + "iat": int(time.time()), + } + + header = {"alg": "Ed25519", "kid": "ed25519-key-1"} + # Ed25519 is not in joserfc's recommended algorithms, so we must explicitly allow it + token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"]) + + return token + + +@pytest.fixture +def expired_ed25519_token(ed25519_joserfc_key): + """Generate expired Ed25519 JWT token for testing.""" + claims = { + "sub": "user456", + "email": "ed25519user@example.com", + "iss": "https://cloud.arcade.dev/oauth2", + "aud": "urn:arcade:mcp", + "exp": int(time.time()) - 3600, + "iat": int(time.time()) - 7200, + } + + header = {"alg": "Ed25519", "kid": "ed25519-key-1"} + # Ed25519 is not in joserfc's recommended algorithms, so we must explicitly allow it + token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"]) return token @@ -185,9 +271,9 @@ class TestJWKSTokenValidator: await validator.validate_token(expired_jwt_token) @pytest.mark.asyncio - async def test_validate_wrong_audience(self, serialized_private_key, jwks_data): + async def test_validate_wrong_audience(self, rsa_joserfc_key, jwks_data): """Test validating token with wrong audience.""" - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://wrong-server.com", # Wrong audience @@ -195,12 +281,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -218,9 +300,9 @@ class TestJWKSTokenValidator: await validator.validate_token(token) @pytest.mark.asyncio - async def test_validate_wrong_issuer(self, serialized_private_key, jwks_data): + async def test_validate_wrong_issuer(self, rsa_joserfc_key, jwks_data): """Test validating token with wrong issuer.""" - payload = { + claims = { "sub": "user123", "iss": "https://wrong-issuer.com", # Wrong issuer "aud": "https://mcp.example.com/mcp", @@ -228,12 +310,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -251,21 +329,17 @@ class TestJWKSTokenValidator: await validator.validate_token(token) @pytest.mark.asyncio - async def test_validate_missing_sub_claim(self, serialized_private_key, jwks_data): + async def test_validate_missing_sub_claim(self, rsa_joserfc_key, jwks_data): """Test validating token without sub claim.""" - payload = { + claims = { "iss": "https://auth.example.com", "aud": "https://mcp.example.com/mcp", "exp": int(time.time()) + 3600, "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -307,11 +381,9 @@ class TestJWKSTokenValidator: assert mock_get.call_count == 1 @pytest.mark.asyncio - async def test_validate_multiple_audiences_single_token_aud( - self, serialized_private_key, jwks_data - ): + async def test_validate_multiple_audiences_single_token_aud(self, rsa_joserfc_key, jwks_data): """Test validator with multiple audiences accepts token with matching single aud.""" - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://old-mcp.example.com", # Matches first audience @@ -319,12 +391,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -342,12 +410,10 @@ class TestJWKSTokenValidator: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_validate_multiple_audiences_list_token_aud( - self, serialized_private_key, jwks_data - ): + async def test_validate_multiple_audiences_list_token_aud(self, rsa_joserfc_key, jwks_data): """Test validator with multiple audiences accepts token with list aud.""" # Token with list of audiences where one matches the validator's accepted audiences - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": ["https://api1.com", "https://new-mcp.example.com"], # Second matches @@ -355,12 +421,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -378,10 +440,10 @@ class TestJWKSTokenValidator: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_validate_multiple_audiences_no_match(self, serialized_private_key, jwks_data): + async def test_validate_multiple_audiences_no_match(self, rsa_joserfc_key, jwks_data): """Test validator with multiple audiences rejects token with non-matching aud.""" # Token with audience that doesn't match any of validator's accepted audiences - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://different-server.com", # Doesn't match @@ -389,12 +451,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -412,12 +470,10 @@ class TestJWKSTokenValidator: await validator.validate_token(token) @pytest.mark.asyncio - async def test_validate_single_audience_with_list_token_aud( - self, serialized_private_key, jwks_data - ): + async def test_validate_single_audience_with_list_token_aud(self, rsa_joserfc_key, jwks_data): """Test validator with single audience accepts token with list aud containing match.""" # Token with list of audiences where one matches validator's single audience - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": ["https://api1.com", "https://mcp.example.com/mcp", "https://api2.com"], @@ -425,12 +481,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -448,10 +500,10 @@ class TestJWKSTokenValidator: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_validate_multiple_issuers_efficient(self, serialized_private_key, jwks_data): + async def test_validate_multiple_issuers_efficient(self, rsa_joserfc_key, jwks_data): """Test that multi-issuer validation is efficient (single decode).""" # Token from second issuer in list - payload = { + claims = { "sub": "user123", "iss": "https://auth2.example.com", # Second in list "aud": "https://mcp.example.com/mcp", @@ -459,12 +511,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()), } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -472,9 +520,9 @@ class TestJWKSTokenValidator: mock_response.raise_for_status = Mock() mock_get.return_value = mock_response - # Patch jwt.decode to count calls with patch( - "arcade_mcp_server.resource_server.validators.jwks.jwt.decode", wraps=jwt.decode + "arcade_mcp_server.resource_server.validators.jwks.jwt.decode", + wraps=jwt.decode, ) as mock_decode: validator = JWKSTokenValidator( jwks_uri="https://auth.example.com/.well-known/jwks.json", @@ -493,9 +541,9 @@ class TestJWKSTokenValidator: assert mock_decode.call_count == 1 @pytest.mark.asyncio - async def test_validate_nbf_claim_before_time(self, serialized_private_key, jwks_data): + async def test_validate_nbf_claim_before_time(self, rsa_joserfc_key, jwks_data): """Test that token with nbf claim in the future is rejected.""" - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://mcp.example.com/mcp", @@ -504,12 +552,8 @@ class TestJWKSTokenValidator: "nbf": int(time.time()) + 3600, # Not valid for 1 hour } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -528,9 +572,9 @@ class TestJWKSTokenValidator: await validator.validate_token(token) @pytest.mark.asyncio - async def test_validate_nbf_claim_disabled(self, serialized_private_key, jwks_data): + async def test_validate_nbf_claim_disabled(self, rsa_joserfc_key, jwks_data): """Test that token with nbf in future is accepted when verify_nbf=False.""" - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://mcp.example.com/mcp", @@ -539,12 +583,8 @@ class TestJWKSTokenValidator: "nbf": int(time.time()) + 3600, # Not valid for 1 hour } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -564,10 +604,10 @@ class TestJWKSTokenValidator: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_validate_with_leeway(self, serialized_private_key, jwks_data): + async def test_validate_with_leeway(self, rsa_joserfc_key, jwks_data): """Test that leeway allows slightly expired tokens.""" # Token expired 30 seconds ago - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://mcp.example.com/mcp", @@ -575,12 +615,8 @@ class TestJWKSTokenValidator: "iat": int(time.time()) - 3600, } - token = jwt.encode( - payload, - serialized_private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -600,6 +636,209 @@ class TestJWKSTokenValidator: assert user.user_id == "user123" +class TestJWKSTokenValidatorEd25519: + """Tests for JWKSTokenValidator with Ed25519 algorithm.""" + + @pytest.mark.asyncio + async def test_validate_valid_ed25519_token(self, valid_ed25519_token, ed25519_jwks_data): + """Test validating a valid Ed25519 JWT token.""" + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + validator = JWKSTokenValidator( + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + audience="urn:arcade:mcp", + algorithm="Ed25519", + ) + + user = await validator.validate_token(valid_ed25519_token) + + assert isinstance(user, ResourceOwner) + assert user.user_id == "user456" + assert user.email == "ed25519user@example.com" + assert user.claims["iss"] == "https://cloud.arcade.dev/oauth2" + + @pytest.mark.asyncio + async def test_validate_expired_ed25519_token(self, expired_ed25519_token, ed25519_jwks_data): + """Test validating an expired Ed25519 JWT token.""" + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + validator = JWKSTokenValidator( + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + audience="urn:arcade:mcp", + algorithm="Ed25519", + ) + + with pytest.raises(TokenExpiredError): + await validator.validate_token(expired_ed25519_token) + + @pytest.mark.asyncio + async def test_ed25519_algorithm_mismatch(self, valid_jwt_token, ed25519_jwks_data): + """Test that RS256 token is rejected when validator expects Ed25519.""" + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + validator = JWKSTokenValidator( + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + issuer="https://auth.example.com", + audience="https://mcp.example.com/mcp", + algorithm="Ed25519", + ) + + # RS256 token should be rejected when Ed25519 is expected + with pytest.raises(InvalidTokenError, match="algorithm"): + await validator.validate_token(valid_jwt_token) + + @pytest.mark.asyncio + async def test_ed25519_wrong_audience(self, ed25519_joserfc_key, ed25519_jwks_data): + """Test Ed25519 token with wrong audience is rejected.""" + claims = { + "sub": "user456", + "iss": "https://cloud.arcade.dev/oauth2", + "aud": "wrong:audience", + "exp": int(time.time()) + 3600, + "iat": int(time.time()), + } + + header = {"alg": "Ed25519", "kid": "ed25519-key-1"} + token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"]) + + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + validator = JWKSTokenValidator( + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + audience="urn:arcade:mcp", + algorithm="Ed25519", + ) + + with pytest.raises(InvalidTokenError, match="audience"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_eddsa_algorithm_alias(self, ed25519_joserfc_key, ed25519_jwks_data): + """Test that EdDSA algorithm alias works for Ed25519.""" + claims = { + "sub": "user456", + "email": "ed25519user@example.com", + "iss": "https://cloud.arcade.dev/oauth2", + "aud": "urn:arcade:mcp", + "exp": int(time.time()) + 3600, + "iat": int(time.time()), + } + + header = {"alg": "Ed25519", "kid": "ed25519-key-1"} + token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"]) + + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + # Use EdDSA alias + validator = JWKSTokenValidator( + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + audience="urn:arcade:mcp", + algorithm="EdDSA", # Using EdDSA alias + ) + + user = await validator.validate_token(token) + assert user.user_id == "user456" + + def test_ed25519_supported_algorithm(self): + """Test that Ed25519 is in supported algorithms.""" + # Should not raise + validator = JWKSTokenValidator( + jwks_uri="https://example.com/jwks", + issuer="https://example.com", + audience="https://example.com", + algorithm="Ed25519", + ) + assert validator.algorithm == "Ed25519" + + def test_eddsa_supported_algorithm(self): + """Test that EdDSA is in supported algorithms.""" + # Should not raise + validator = JWKSTokenValidator( + jwks_uri="https://example.com/jwks", + issuer="https://example.com", + audience="https://example.com", + algorithm="EdDSA", + ) + assert validator.algorithm == "EdDSA" + + +class TestArcadeASConfiguration: + """Tests for Arcade AS configuration.""" + + @pytest.mark.asyncio + async def test_arcade_as_config(self, valid_ed25519_token, ed25519_jwks_data): + """Test configuration matching Arcade AS.""" + with patch("httpx.AsyncClient.get") as mock_get: + mock_response = Mock() + mock_response.json.return_value = ed25519_jwks_data + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + # Configuration matching Arcade AS + resource_server_auth = ResourceServerAuth( + canonical_url="https://gateway-manager.arcade.dev/mcp", + authorization_servers=[ + AuthorizationServerEntry( + authorization_server_url="https://cloud.arcade.dev/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + algorithm="Ed25519", + expected_audiences=[ + "urn:arcade:mcp", + "https://gateway-manager.arcade.dev/mcp", + ], + ) + ], + ) + + user = await resource_server_auth.validate_token(valid_ed25519_token) + assert user.user_id == "user456" + assert user.email == "ed25519user@example.com" + + def test_arcade_as_metadata(self): + """Test OAuth metadata for Arcade AS configuration.""" + resource_server_auth = ResourceServerAuth( + canonical_url="https://gateway-manager.arcade.dev/mcp", + authorization_servers=[ + AuthorizationServerEntry( + authorization_server_url="https://cloud.arcade.dev/oauth2", + issuer="https://cloud.arcade.dev/oauth2", + jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2", + algorithm="Ed25519", + expected_audiences=["urn:arcade:mcp"], + ) + ], + ) + + metadata = resource_server_auth.get_resource_metadata() + assert metadata["resource"] == "https://gateway-manager.arcade.dev/mcp" + assert metadata["authorization_servers"] == ["https://cloud.arcade.dev/oauth2"] + + # ResourceServerAuth Tests class TestResourceServerAuth: """Tests for ResourceServerAuth class.""" @@ -639,12 +878,10 @@ class TestResourceServerAuth: assert metadata["bearer_methods_supported"] == ["header"] @pytest.mark.asyncio - async def test_expected_audiences_override(self, rsa_keypair, jwks_data): + async def test_expected_audiences_override(self, rsa_keypair, jwks_data, rsa_joserfc_key): """Test that expected_audiences overrides canonical_url for audience validation.""" - private_key, _ = rsa_keypair - # Token with custom audience - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "my-authkit-client-id", @@ -652,12 +889,8 @@ class TestResourceServerAuth: "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -681,12 +914,12 @@ class TestResourceServerAuth: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_expected_audiences_multiple_values(self, rsa_keypair, jwks_data): + async def test_expected_audiences_multiple_values( + self, rsa_keypair, jwks_data, rsa_joserfc_key + ): """Test that multiple expected_audiences work correctly.""" - private_key, _ = rsa_keypair - # Token with one of the expected audiences - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "secondary-client-id", @@ -694,12 +927,8 @@ class TestResourceServerAuth: "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -727,11 +956,11 @@ class TestResourceServerAuth: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_expected_audiences_defaults_to_canonical_url(self, rsa_keypair, jwks_data): + async def test_expected_audiences_defaults_to_canonical_url( + self, rsa_keypair, jwks_data, rsa_joserfc_key + ): """Test that without expected_audiences, canonical_url is used for audience validation.""" - private_key, _ = rsa_keypair - - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "https://mcp.example.com/mcp", @@ -739,12 +968,8 @@ class TestResourceServerAuth: "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -767,11 +992,11 @@ class TestResourceServerAuth: assert user.user_id == "user123" @pytest.mark.asyncio - async def test_expected_audiences_wrong_audience_rejected(self, rsa_keypair, jwks_data): + async def test_expected_audiences_wrong_audience_rejected( + self, rsa_keypair, jwks_data, rsa_joserfc_key + ): """Test that tokens with wrong audience are rejected even with expected_audiences.""" - private_key, _ = rsa_keypair - - payload = { + claims = { "sub": "user123", "iss": "https://auth.example.com", "aud": "wrong-client-id", # Not in expected_audiences list @@ -779,12 +1004,8 @@ class TestResourceServerAuth: "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -1085,11 +1306,11 @@ class TestMultipleAuthorizationServers: assert user.email == "user@example.com" @pytest.mark.asyncio - async def test_resource_server_multiple_as_different_jwks(self, rsa_keypair, jwks_data): + async def test_resource_server_multiple_as_different_jwks( + self, rsa_keypair, jwks_data, rsa_joserfc_key + ): """Test multiple AS with different JWKS (multi-IdP).""" - private_key, _ = rsa_keypair - - payload1 = { + claims1 = { "sub": "user123", "email": "user@workos.com", "iss": "https://workos.authkit.app", @@ -1097,14 +1318,10 @@ class TestMultipleAuthorizationServers: "exp": int(time.time()) + 3600, "iat": int(time.time()), } - token1 = jwt.encode( - payload1, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header1 = {"alg": "RS256", "kid": "test-key-1"} + token1 = jwt.encode(header1, claims1, rsa_joserfc_key) - payload2 = { + claims2 = { "sub": "user456", "email": "user@keycloak.com", "iss": "http://localhost:8080/realms/mcp-test", @@ -1112,12 +1329,8 @@ class TestMultipleAuthorizationServers: "exp": int(time.time()) + 3600, "iat": int(time.time()), } - token2 = jwt.encode( - payload2, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header2 = {"alg": "RS256", "kid": "test-key-1"} + token2 = jwt.encode(header2, claims2, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock() @@ -1159,11 +1372,11 @@ class TestMultipleAuthorizationServers: assert user2.email == "user@keycloak.com" @pytest.mark.asyncio - async def test_resource_server_rejects_unconfigured_as(self, rsa_keypair, jwks_data): + async def test_resource_server_rejects_unconfigured_as( + self, rsa_keypair, jwks_data, rsa_joserfc_key + ): """Test that tokens from unlisted AS are rejected.""" - private_key, _ = rsa_keypair - - payload = { + claims = { "sub": "user123", "email": "user@evil.com", "iss": "https://evil.com", # Not in configured list (unauthorized issuer) @@ -1171,12 +1384,8 @@ class TestMultipleAuthorizationServers: "exp": int(time.time()) + 3600, "iat": int(time.time()), } - token = jwt.encode( - payload, - private_key, - algorithm="RS256", - headers={"kid": "test-key-1"}, - ) + header = {"alg": "RS256", "kid": "test-key-1"} + token = jwt.encode(header, claims, rsa_joserfc_key) with patch("httpx.AsyncClient.get") as mock_get: mock_response = Mock()