Treat an access token as expired at its expiry time

Author: Kludex

src/mcp/client/auth/extensions/identity_assertion.py

@@ -167,7 +167,7 @@ async def _auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Req
                 self._expiry = calculate_token_expiry(self._tokens.expires_in) if self._tokens else None
                 self._initialized = True
 
-            if self._tokens and (self._expiry is None or time.time() <= self._expiry):
+            if self._tokens and (self._expiry is None or time.time() < self._expiry):
                 request.headers["Authorization"] = f"Bearer {self._tokens.access_token}"
             response = yield request
 

src/mcp/client/auth/oauth2.py

@@ -188,7 +188,7 @@ def is_token_valid(self) -> bool:
         return bool(
             self.current_tokens
             and self.current_tokens.access_token
-            and (not self.token_expiry_time or time.time() <= self.token_expiry_time)
+            and (not self.token_expiry_time or time.time() < self.token_expiry_time)
         )
 
     def can_refresh_token(self) -> bool:

tests/client/auth/extensions/test_identity_assertion.py

@@ -8,6 +8,7 @@
 import base64
 import json
 import urllib.parse
+from unittest import mock
 
 import httpx2
 import pytest
@@ -324,6 +325,21 @@ async def test_stored_token_is_reused_without_reauthorizing() -> None:
     assert requests[0].headers["Authorization"] == "Bearer cached"
 
 
+@pytest.mark.anyio
+async def test_stored_token_is_not_sent_at_its_expiry_instant() -> None:
+    """A stored token whose expiry equals the current time is already expired, so it is not attached."""
+    requests: list[httpx2.Request] = []
+    storage = InMemoryStorage(tokens=OAuthToken(access_token="cached", token_type="Bearer", expires_in=0))
+    auth = make_provider(storage)
+
+    transport = mock_transport(requests, rs_first_status=200)
+    with mock.patch("time.time", return_value=1_800_000_000.0):
+        async with httpx2.AsyncClient(transport=transport, auth=auth) as http:
+            await http.post(f"{RS}/mcp")
+
+    assert "Authorization" not in requests[0].headers
+
+
 @pytest.mark.anyio
 async def test_second_401_re_exchanges_without_refetching_asm() -> None:
     """ASM is discovered once; a later 401 mints a fresh assertion against the cached token endpoint."""

tests/client/test_auth.py

@@ -244,6 +244,11 @@ async def test_token_validity_checking(self, oauth_provider: OAuthClientProvider
         assert not context.is_token_valid()
         assert context.can_refresh_token()  # Can still refresh
 
+        # At the expiry instant itself the token is already expired (RFC 7519 §4.1.4)
+        with mock.patch("time.time", return_value=1_800_000_000.0):
+            context.token_expiry_time = 1_800_000_000.0
+            assert not context.is_token_valid()
+
         # Remove refresh token
         context.current_tokens.refresh_token = None
         assert not context.can_refresh_token()