fix(e2e): bind MCP OAuth acceptance to the owned gateway and snapshot the stored token once per phase

Co-Authored-By: bot_apk <apk@cognition.ai>

.github/workflows/test-mcp-oauth-e2e.yml

@@ -1,23 +1,12 @@
 name: MCP OAuth happy path
 
 on:
-  pull_request:
-    paths:
-      - tests/e2e/idp.py
-      - tests/e2e/provider_edge.py
-      - tests/e2e/models.py
-      - tests/e2e/conftest.py
-      - .github/e2e-stack/assert_tests_ran.py
-      - tests/e2e/mcp/oauth_chat_client.py
-      - tests/e2e/mcp/oauth_gateway.py
-      - tests/e2e/mcp/test_mcp_oauth_happy_path_e2e.py
-      - .github/workflows/test-mcp-oauth-e2e.yml
   workflow_dispatch:
 
 permissions: {}
 
 concurrency:
-  group: mcp-oauth-${{ github.event.pull_request.number || github.ref }}
+  group: mcp-oauth-${{ github.ref }}
   cancel-in-progress: true
 
 jobs:
@@ -161,6 +150,10 @@ jobs:
         run: |
           uv run --no-sync python .github/e2e-stack/assert_tests_ran.py \
             "${RUNNER_TEMP}/mcp-oauth-private/results.xml" tests/e2e/mcp/test_mcp_oauth_happy_path_e2e.py
+      - name: Publish sanitized summary
+        if: always()
+        run: |
+          grep -E '^(FAILED|PASSED|ERROR|E   AssertionError|=+ .* =+)' "${RUNNER_TEMP}/mcp-oauth-private/pytest.log" || true
       - name: Remove private login and logs
         if: always()
         run: |

tests/e2e/mcp/oauth_gateway.py

@@ -26,6 +26,8 @@ from proxy_client import ProxyClient, build_proxy_client
 from psycopg.rows import class_row
 from pydantic import BaseModel, SecretStr, TypeAdapter, ValidationError
 
+INHERITED_ENV_PREFIXES: Final = ("REDIS_", "MICROSOFT_", "GOOGLE_", "GENERIC_", "PROXY_")
+
 
 class StoredOAuth(BaseModel):
     type: str
@@ -38,6 +40,7 @@ class CredentialRow:
 
 
 def stored_oauth(user_id: str, server_id: str) -> StoredOAuth:
+    """Read the encrypted credential because management APIs omit the plaintext token."""
     from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
 
     with psycopg.Connection[CredentialRow].connect(
@@ -68,14 +71,12 @@ class RpcMethod(BaseModel):
 
 @dataclass(slots=True)
 class OAuthObservation:
-    user_id: str
-    server_id: str = ""
     gateway_token: str = field(default="", repr=False)
-    _seen: tuple[tuple[str, bool, bool], ...] = field(default=(), init=False, repr=False)
+    _seen: tuple[tuple[str, str, bool], ...] = field(default=(), init=False, repr=False)
     _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
 
     def observe(self, url: str, headers: Mapping[str, str], body: bytes | None) -> None:
-        if not self.server_id or body is None or not url.endswith("/mcp"):
+        if body is None or not url.endswith("/mcp"):
             return
         try:
             operation: Final = RpcMethod.model_validate_json(body).method
@@ -83,21 +84,21 @@ class OAuthObservation:
             return
         if operation not in ("tools/list", "tools/call"):
             return
-        credential: Final = stored_oauth(self.user_id, self.server_id)
         received: Final = headers.get("authorization", "")
-        matches: Final = received == f"Bearer {credential.access_token.get_secret_value()}"
-        differs: Final = bool(received) and all(
-            value not in (self.gateway_token, f"Bearer {self.gateway_token}") for value in headers.values()
+        gateway_leaked: Final = any(
+            value in (self.gateway_token, f"Bearer {self.gateway_token}") for value in headers.values()
         )
         with self._lock:
-            self._seen = (*self._seen, (operation, matches, differs))
+            self._seen = (*self._seen, (operation, received, gateway_leaked))
 
-    def assert_forwarded(self) -> None:
+    def assert_forwarded(self, expected: StoredOAuth) -> None:
         with self._lock:
             snapshot: Final = self._seen
             self._seen = ()
         assert {item[0] for item in snapshot} == {"tools/list", "tools/call"}, "missing upstream observations"
-        assert all(item[1] and item[2] for item in snapshot), "upstream bearer did not match the user's stored token"
+        expected_header: Final = f"Bearer {expected.access_token.get_secret_value()}"
+        assert all(item[1] == expected_header for item in snapshot), "upstream bearer did not match the stored token"
+        assert all(not item[2] for item in snapshot), "gateway bearer leaked to the upstream"
 
 
 def available_port() -> int:
@@ -170,7 +171,7 @@ def owned_gateway(idp: Keycloak, directory: Path, cleanup: ExitStack) -> OAuthGa
         "    user_id_upsert: true\n"
     )
     environment: Final = {
-        **{key: value for key, value in os.environ.items() if not key.startswith("REDIS_")},
+        **{key: value for key, value in os.environ.items() if not key.startswith(INHERITED_ENV_PREFIXES)},
         **browser.environment(idp.discovery()),
         "PROXY_BASE_URL": base_url,
         "JWT_PUBLIC_KEY_URL": idp.jwks_url,

tests/e2e/mcp/test_mcp_oauth_happy_path_e2e.py

@@ -102,7 +102,7 @@ class TestMcpOauthHappyPath:
         alias: Final = f"e2elinear{unique_marker()}"
         tool: Final = f"{alias}-{LINEAR_READONLY_TOOL}"
         token: Final = idp.access_token(jwt_identity)
-        observation: Final = OAuthObservation(user_id=jwt_identity.user_id, gateway_token=token)
+        observation: Final = OAuthObservation(gateway_token=token)
         edge: Final = (
             start_provider_edge(
                 LiveEdge(observe_request=observation.observe),
@@ -145,13 +145,6 @@ class TestMcpOauthHappyPath:
         assert client.server_user_credentials(created.server_id) == (), (
             "scenario must start without upstream credentials"
         )
-        observation.server_id = created.server_id
-        client.proxy.update_team(
-            TeamUpdateBody(
-                team_id=jwt_identity.group,
-                object_permission=ObjectPermission(mcp_servers=[created.server_id]),
-            )
-        )
         unwrap(
             client.proxy.transport.post(
                 "/team/member_add",
@@ -162,6 +155,12 @@ class TestMcpOauthHappyPath:
                 response_type=NoBody,
             )
         )
+        client.proxy.update_team(
+            TeamUpdateBody(
+                team_id=jwt_identity.group,
+                object_permission=ObjectPermission(mcp_servers=[created.server_id]),
+            )
+        )
         headers: Final = {"x-litellm-api-key": f"Bearer {token}"} if route == "explicit_header_jwt" else {}
         resources.defer(
             lambda: client.revoke_user_token(
@@ -185,9 +184,9 @@ class TestMcpOauthHappyPath:
         assert len(credentials) == 1
         assert credentials[0].user_id == jwt_identity.user_id
         assert credentials[0].credential_type == "oauth2"
-        stored_oauth(jwt_identity.user_id, created.server_id)
+        first_stored_oauth: Final = stored_oauth(jwt_identity.user_id, created.server_id)
         if observed:
-            observation.assert_forwarded()
+            observation.assert_forwarded(first_stored_oauth)
         oauth_gateway.restart()
         fresh_token: Final = idp.access_token(jwt_identity)
         observation.gateway_token = fresh_token
@@ -203,6 +202,6 @@ class TestMcpOauthHappyPath:
             allow_upstream_consent=False,
         )
         assert_tool_result(second, tool)
-        stored_oauth(jwt_identity.user_id, created.server_id)
+        second_stored_oauth: Final = stored_oauth(jwt_identity.user_id, created.server_id)
         if observed:
-            observation.assert_forwarded()
+            observation.assert_forwarded(second_stored_oauth)