security: enforce JWT audience + require sub in token verification

Closes the audience-binding gap (RFC 9068): tokens minted by the issuer for a
different resource are now rejected at /mcp, and subject-less tokens are refused.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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,

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():