OAuth2: key token caches by the full credential configuration

.github/workflows/test.yml

@@ -29,13 +29,16 @@ jobs:
         pip install -e "core[dev]"
         pip install -e plugins/communication_protocols/cli[dev]
         pip install -e plugins/communication_protocols/http[dev]
+        pip install -e plugins/communication_protocols/file[dev]
         pip install -e plugins/communication_protocols/mcp[dev]
         pip install -e plugins/communication_protocols/text[dev]
         pip install -e plugins/communication_protocols/socket[dev]
+        pip install -e plugins/communication_protocols/websocket[dev]
+        pip install -e plugins/communication_protocols/gql[dev]
 
     - name: Run tests with pytest
       run: |
-        pytest core/tests/ plugins/communication_protocols/cli/tests/ plugins/communication_protocols/http/tests/ plugins/communication_protocols/mcp/tests/ plugins/communication_protocols/text/tests/ plugins/communication_protocols/socket/tests/ --doctest-modules --junitxml=junit/test-results.xml --cov=core/src/utcp --cov-report=xml --cov-report=html
+        pytest core/tests/ plugins/communication_protocols/cli/tests/ plugins/communication_protocols/http/tests/ plugins/communication_protocols/file/tests/ plugins/communication_protocols/mcp/tests/ plugins/communication_protocols/text/tests/ plugins/communication_protocols/socket/tests/ plugins/communication_protocols/websocket/tests/ plugins/communication_protocols/gql/tests/ --doctest-modules --junitxml=junit/test-results.xml --cov=core/src/utcp --cov-report=xml --cov-report=html
 
     - name: Upload coverage reports to Codecov
       uses: codecov/codecov-action@v3

core/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp"
-version = "1.1.3"
+version = "1.1.4"
 authors = [
   { name = "UTCP Contributors" },
 ]

core/src/utcp/data/auth_implementations/oauth2_auth.py

@@ -1,3 +1,5 @@
+import json
+
 from utcp.data.auth import Auth
 from utcp.interfaces.serializer import Serializer
 from utcp.exceptions import UtcpSerializerValidationError
@@ -26,6 +28,21 @@ class OAuth2Auth(Auth):
     client_secret: str = Field(..., description="The OAuth2 client secret.")
     scope: Optional[str] = Field(None, description="The OAuth2 scope.")
 
+    def cache_key(self) -> str:
+        """The identity of this credential configuration, for caching and coalescing tokens.
+
+        Two ``OAuth2Auth`` values obtain the same token exactly when this key is
+        equal, so it includes everything that changes the token: the token
+        endpoint, client id, client secret and scope. Keying by ``client_id``
+        alone would let two configurations that share a client id but differ in
+        issuer, secret or scope receive each other's tokens. An absent scope is
+        normalised to ``""`` so ``None`` and ``""`` compare equal.
+
+        This is the single source of that rule for every communication protocol.
+        The key embeds the secret: use it only as a dictionary key, never log it.
+        """
+        return json.dumps([self.token_url, self.client_id, self.client_secret, self.scope or ""])
+
 
 class OAuth2AuthSerializer(Serializer[OAuth2Auth]):
     """REQUIRED

core/tests/data/test_oauth2_auth.py

@@ -0,0 +1,45 @@
+"""``OAuth2Auth.cache_key``: the identity of a credential configuration.
+
+This key is the single source of the rule that every communication protocol
+uses to cache and coalesce tokens, so its semantics are pinned here: two
+configurations share a key exactly when they would obtain the same token.
+"""
+
+from utcp.data.auth_implementations import OAuth2Auth
+
+
+def _auth(**overrides) -> OAuth2Auth:
+    fields = dict(
+        auth_type="oauth2",
+        token_url="https://issuer-a.example/token",
+        client_id="client",
+        client_secret="secret",
+        scope="read",
+    )
+    fields.update(overrides)
+    return OAuth2Auth(**fields)
+
+
+def test_identical_configurations_share_a_key():
+    assert _auth().cache_key() == _auth().cache_key()
+
+
+def test_every_token_affecting_field_changes_the_key():
+    base = _auth().cache_key()
+    assert _auth(token_url="https://issuer-b.example/token").cache_key() != base
+    assert _auth(client_id="other").cache_key() != base
+    assert _auth(client_secret="other").cache_key() != base
+    assert _auth(scope="write").cache_key() != base
+
+
+def test_same_client_id_at_a_different_issuer_does_not_share_a_key():
+    # The flaw this rule exists to prevent: a shared client_id must never let
+    # two configurations receive each other's tokens.
+    a = _auth(token_url="https://issuer-a.example/token")
+    b = _auth(token_url="https://issuer-b.example/token")
+    assert a.client_id == b.client_id
+    assert a.cache_key() != b.cache_key()
+
+
+def test_absent_scope_normalises_to_empty():
+    assert _auth(scope=None).cache_key() == _auth(scope="").cache_key()

plugins/communication_protocols/gql/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-gql"
-version = "1.1.4"
+version = "1.1.5"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -15,7 +15,7 @@ dependencies = [
     "pydantic>=2.0",
     "gql>=3.0",
     "aiohttp>=3.8",
-    "utcp>=1.1"
+    "utcp>=1.1.4"
 ]
 classifiers = [
     "Development Status :: 4 - Beta",

plugins/communication_protocols/gql/src/utcp_gql/gql_communication_protocol.py

@@ -80,8 +80,9 @@ async def _handle_oauth2(self, auth: OAuth2Auth) -> str:
         GHSA-8cp3-qxj6-px34 and GHSA-9qhg-99ww-9mqc.
         """
         client_id = auth.client_id
-        if client_id in self._oauth_tokens:
-            return self._oauth_tokens[client_id]["access_token"]
+        cache_key = auth.cache_key()
+        if cache_key in self._oauth_tokens:
+            return self._oauth_tokens[cache_key]["access_token"]
 
         ensure_secure_url(auth.token_url, context="OAuth2 token URL")
 
@@ -101,7 +102,7 @@ async def _handle_oauth2(self, auth: OAuth2Auth) -> str:
             ) as resp:
                 resp.raise_for_status()
                 token_response = await resp.json()
-                self._oauth_tokens[client_id] = token_response
+                self._oauth_tokens[cache_key] = token_response
                 return token_response["access_token"]
 
     async def _prepare_headers(

plugins/communication_protocols/gql/tests/test_gql_oauth_cache_isolation.py

@@ -0,0 +1,30 @@
+"""OAuth2 token cache is isolated per credential configuration.
+
+The token cache is keyed by ``OAuth2Auth.cache_key``, so two templates that
+share a ``client_id`` but differ in issuer, secret or scope never receive each
+other's tokens. Network-free: the cache is seeded directly, and the second
+configuration is only checked for absence.
+"""
+
+import pytest
+
+from utcp.data.auth_implementations import OAuth2Auth
+from utcp_gql.gql_communication_protocol import GraphQLCommunicationProtocol
+
+
+def _auth(token_url: str) -> OAuth2Auth:
+    return OAuth2Auth(
+        auth_type="oauth2", token_url=token_url, client_id="shared", client_secret="s", scope=""
+    )
+
+
+@pytest.mark.asyncio
+async def test_token_cache_is_isolated_per_credential_configuration():
+    proto = GraphQLCommunicationProtocol()
+    a = _auth("https://issuer-a.example/token")
+    b = _auth("https://issuer-b.example/token")  # same client_id, different issuer
+
+    proto._oauth_tokens[a.cache_key()] = {"access_token": "token-for-a"}
+
+    assert await proto._handle_oauth2(a) == "token-for-a"
+    assert b.cache_key() not in proto._oauth_tokens

plugins/communication_protocols/http/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-http"
-version = "1.1.12"
+version = "1.1.13"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -16,7 +16,7 @@ dependencies = [
     "authlib>=1.0",
     "aiohttp>=3.8",
     "pyyaml>=6.0",
-    "utcp>=1.1"
+    "utcp>=1.1.4"
 ]
 classifiers = [
     "Development Status :: 4 - Beta",

plugins/communication_protocols/http/src/utcp_http/http_communication_protocol.py

@@ -63,7 +63,7 @@ class HttpCommunicationProtocol(CommunicationProtocol):
 
     Attributes:
         _session: Optional aiohttp ClientSession for connection reuse.
-        _oauth_tokens: Cache of OAuth2 tokens by client_id.
+        _oauth_tokens: Cache of OAuth2 tokens keyed by the full credential configuration (``OAuth2Auth.cache_key``).
         _log: Logger function for debugging and error reporting.
     """
 
@@ -414,8 +414,9 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
         """
         client_id = auth_details.client_id
 
-        if client_id in self._oauth_tokens:
-            return self._oauth_tokens[client_id]["access_token"]
+        cache_key = auth_details.cache_key()
+        if cache_key in self._oauth_tokens:
+            return self._oauth_tokens[cache_key]["access_token"]
 
         # Reject obviously-internal or plain-HTTP non-loopback token
         # endpoints before any credential bytes leave the process.
@@ -440,7 +441,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_response = await response.json()
-                    self._oauth_tokens[client_id] = token_response
+                    self._oauth_tokens[cache_key] = token_response
                     return token_response["access_token"]
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with credentials in body failed: {e}. Trying Basic Auth header.")
@@ -463,7 +464,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_response = await response.json()
-                    self._oauth_tokens[client_id] = token_response
+                    self._oauth_tokens[cache_key] = token_response
                     return token_response["access_token"]
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with Basic Auth header also failed: {e}")

plugins/communication_protocols/http/src/utcp_http/sse_communication_protocol.py

@@ -480,8 +480,9 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
         endpoint itself.
         """
         client_id = auth_details.client_id
-        if client_id in self._oauth_tokens:
-            return self._oauth_tokens[client_id]["access_token"]
+        cache_key = auth_details.cache_key()
+        if cache_key in self._oauth_tokens:
+            return self._oauth_tokens[cache_key]["access_token"]
 
         # Reject obviously-internal or plain-HTTP non-loopback token
         # endpoints before any credential bytes leave the process.
@@ -499,7 +500,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_response = await response.json()
-                    self._oauth_tokens[client_id] = token_response
+                    self._oauth_tokens[cache_key] = token_response
                     return token_response["access_token"]
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with body failed: {e}. Trying Basic Auth.")
@@ -517,7 +518,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_response = await response.json()
-                    self._oauth_tokens[client_id] = token_response
+                    self._oauth_tokens[cache_key] = token_response
                     return token_response["access_token"]
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with header failed: {e}")

plugins/communication_protocols/http/src/utcp_http/streamable_http_communication_protocol.py

@@ -362,8 +362,9 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
         endpoint itself.
         """
         client_id = auth_details.client_id
-        if client_id in self._oauth_tokens:
-            return self._oauth_tokens[client_id]["access_token"]
+        cache_key = auth_details.cache_key()
+        if cache_key in self._oauth_tokens:
+            return self._oauth_tokens[cache_key]["access_token"]
 
         # Reject obviously-internal or plain-HTTP non-loopback token
         # endpoints before any credential bytes leave the process.
@@ -387,7 +388,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_data = await response.json()
-                    self._oauth_tokens[client_id] = token_data
+                    self._oauth_tokens[cache_key] = token_data
                     return token_data['access_token']
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with credentials in body failed: {e}. Trying Basic Auth header.")
@@ -409,7 +410,7 @@ async def _handle_oauth2(self, auth_details: OAuth2Auth) -> str:
                 ) as response:
                     response.raise_for_status()
                     token_data = await response.json()
-                    self._oauth_tokens[client_id] = token_data
+                    self._oauth_tokens[cache_key] = token_data
                     return token_data['access_token']
             except aiohttp.ClientError as e:
                 logger.error(f"OAuth2 with Basic Auth header also failed: {e}")

plugins/communication_protocols/http/tests/test_http_oauth_cache_isolation.py

@@ -0,0 +1,39 @@
+"""OAuth2 token caches are isolated per credential configuration.
+
+Each HTTP-family protocol keys its token cache by ``OAuth2Auth.cache_key``,
+so two templates that share a ``client_id`` but differ in issuer, secret or
+scope never receive each other's tokens. Network-free: the cache is seeded
+directly, and the second configuration is only checked for absence.
+"""
+
+import pytest
+
+from utcp.data.auth_implementations import OAuth2Auth
+from utcp_http.http_communication_protocol import HttpCommunicationProtocol
+from utcp_http.sse_communication_protocol import SseCommunicationProtocol
+from utcp_http.streamable_http_communication_protocol import StreamableHttpCommunicationProtocol
+
+
+def _auth(token_url: str) -> OAuth2Auth:
+    return OAuth2Auth(
+        auth_type="oauth2", token_url=token_url, client_id="shared", client_secret="s", scope=""
+    )
+
+
+@pytest.mark.parametrize(
+    "protocol_class",
+    [HttpCommunicationProtocol, SseCommunicationProtocol, StreamableHttpCommunicationProtocol],
+)
+@pytest.mark.asyncio
+async def test_token_cache_is_isolated_per_credential_configuration(protocol_class):
+    proto = protocol_class()
+    a = _auth("https://issuer-a.example/token")
+    b = _auth("https://issuer-b.example/token")  # same client_id, different issuer
+
+    proto._oauth_tokens[a.cache_key()] = {"access_token": "token-for-a"}
+
+    # A is served from the cache (proves the key is what the lookup uses)...
+    assert await proto._handle_oauth2(a) == "token-for-a"
+    # ...and B, sharing only the client_id, has no entry and so can never be
+    # handed A's token.
+    assert b.cache_key() not in proto._oauth_tokens

plugins/communication_protocols/mcp/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-mcp"
-version = "1.1.3"
+version = "1.1.4"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -14,7 +14,7 @@ requires-python = ">=3.11"
 dependencies = [
     "pydantic>=2.0",
     "mcp>=1.12,<2",
-    "utcp>=1.1",
+    "utcp>=1.1.4",
     "mcp-use>=1.3",
     "langchain>=0.3.27,<0.4.0",
 ]

plugins/communication_protocols/mcp/src/utcp_mcp/mcp_communication_protocol.py

@@ -249,12 +249,11 @@ def _owner_key(manual_call_template: 'McpCallTemplate', config_key: str) -> str:
     def _oauth_cache_key(auth: OAuth2Auth) -> str:
         """Key for the OAuth token cache and in-flight map.
 
-        Keyed by the FULL configuration, not ``client_id`` alone: two manuals may
-        share a client_id but point at different issuers, scopes or secrets, and
-        must not receive each other's tokens. Matches the HTTP plugin. Carries
-        the secret, so it is used only as a dict key and never logged.
+        Delegates to ``OAuth2Auth.cache_key`` in core, the single source of the
+        rule that a credential's identity is its full configuration, not
+        ``client_id`` alone. Carries the secret: dict key only, never logged.
         """
-        return json.dumps([auth.token_url, auth.client_id, auth.client_secret, auth.scope or ""])
+        return auth.cache_key()
 
     async def _ensure_mcp_client(self, manual_call_template: 'McpCallTemplate') -> MCPClient:
         """Return the MCPClient for this manual's configuration, creating it once.

plugins/communication_protocols/websocket/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "utcp-websocket"
-version = "1.1.4"
+version = "1.1.5"
 authors = [
   { name = "UTCP Contributors" },
 ]
@@ -14,7 +14,7 @@ requires-python = ">=3.10"
 dependencies = [
     "pydantic>=2.0",
     "aiohttp>=3.8",
-    "utcp>=1.1"
+    "utcp>=1.1.4"
 ]
 classifiers = [
     "Development Status :: 4 - Beta",