diff --git a/src/intervals_mcp_server/auth.py b/src/intervals_mcp_server/auth.py index 3fcab70..fb1e70e 100644 --- a/src/intervals_mcp_server/auth.py +++ b/src/intervals_mcp_server/auth.py @@ -40,16 +40,17 @@ class AuthentikTokenVerifier: try: key = self._jwks.get_signing_key_from_jwt(token).key - # Validate issuer + signature + expiry strictly. Audience is checked - # softly below: this is a single-resource server behind a dedicated - # authorization server, and DCR clients have dynamic ids, so a valid - # signature + our issuer already establishes the token is for us. + # Validate issuer + signature + expiry strictly, and bind the token to + # this resource server (RFC 9068): PyJWT enforces that the ``aud`` + # claim contains at least one of our accepted audiences. ``sub`` is + # required so subject-less tokens are rejected outright. claims = jwt.decode( token, key, algorithms=_ALGORITHMS, issuer=self._issuer, - options={"require": ["exp", "iat", "iss"], "verify_aud": False}, + audience=self._audience, + options={"require": ["exp", "iat", "iss", "aud", "sub"]}, ) except Exception as exc: # noqa: BLE001 - any failure means unauthenticated logger.debug("Token verification failed: %s", exc) @@ -57,13 +58,6 @@ class AuthentikTokenVerifier: aud = claims.get("aud") auds = aud if isinstance(aud, list) else ([aud] if aud else []) - if auds and not any(a in self._audience for a in auds): - logger.warning( - "Token audience %s not in accepted %s; accepting (single resource).", - auds, - self._audience, - ) - resource = auds[0] if auds else self._audience[0] return AccessToken( token=token, diff --git a/tests/test_auth.py b/tests/test_auth.py index c8a9eac..afcb116 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -3,8 +3,8 @@ Tests for intervals_mcp_server.auth — native OAuth token verification. Covers the real security logic: a valid JWT (RS256 or Better Auth's EdDSA) yields an AccessToken with the right claims; tampered / expired / wrong-issuer / wrong-key -tokens yield None; audience is checked softly (single resource); and build_auth -only enables auth when the environment is configured. +/ wrong-audience / subject-less tokens yield None (audience is enforced per +RFC 9068); and build_auth only enables auth when the environment is configured. """ import asyncio @@ -12,10 +12,8 @@ import time import types import jwt -import pytest from cryptography.hazmat.primitives.asymmetric import ed25519, rsa -from intervals_mcp_server import auth as auth_mod from intervals_mcp_server.auth import AuthentikTokenVerifier, _audience_variants, build_auth ISSUER = "https://auth.example/application/o/x/" @@ -76,18 +74,41 @@ def test_expired_token_rejected(): assert asyncio.run(v.verify_token(tok)) is None -def test_wrong_audience_tolerated_single_resource(): - # Audience is checked softly: this is a single-resource server behind a - # dedicated authorization server with dynamic (DCR) client ids, so a genuine - # token — correct issuer + signature + not expired — is accepted even if its - # audience differs. The trust boundary is issuer + signature (asserted by the - # wrong-issuer / wrong-key tests below); a mismatch is logged, not rejected. +def test_wrong_audience_rejected(): + # Audience is enforced (RFC 9068): a token minted by the same issuer but for a + # different resource/audience must be rejected. This closes the confused-deputy + # / token-replay leg where a token for another resource was accepted at /mcp. priv, pub = _keypair() v = _verifier(pub, ["https://someone-else"]) tok = _token(priv, aud="https://some-other-resource") - result = asyncio.run(v.verify_token(tok)) - assert result is not None - assert result.subject == "user-123" + assert asyncio.run(v.verify_token(tok)) is None + + +def test_missing_audience_rejected(): + # ``aud`` is a required claim; a token without it is rejected outright. + priv, pub = _keypair() + v = _verifier(pub, [RESOURCE]) + now = int(time.time()) + tok = jwt.encode( + {"iss": ISSUER, "exp": now + 3600, "iat": now, "sub": "user-123"}, + priv, + algorithm="RS256", + ) + assert asyncio.run(v.verify_token(tok)) is None + + +def test_subjectless_token_rejected(): + # ``sub`` is required: subject-less tokens must be rejected rather than + # falling through to an env-credential fallback. + priv, pub = _keypair() + v = _verifier(pub, [RESOURCE]) + now = int(time.time()) + tok = jwt.encode( + {"iss": ISSUER, "aud": RESOURCE, "exp": now + 3600, "iat": now}, + priv, + algorithm="RS256", + ) + assert asyncio.run(v.verify_token(tok)) is None def test_eddsa_token_accepted():