fix: Make credential key generation stable and prevent cross-user credential leaks

This change updates the credential key generation to use a stable hash (SHA256) instead of Python's built-in hash, which can vary based on PYTHONHASHSEED. It also makes sure that temporary or exchanged OAuth2 fields are excluded from the key calculation. I also added when saving credentials, a copy of the AuthConfig is used to avoid modifying the original shared AuthConfig instance with user-specific exchanged credentials.

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 864599326

tests/unittests/auth/test_auth_config.py

@@ -12,6 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+from pathlib import Path
+import subprocess
+import sys
+
 from fastapi.openapi.models import OAuth2
 from fastapi.openapi.models import OAuthFlowAuthorizationCode
 from fastapi.openapi.models import OAuthFlows
@@ -107,3 +112,53 @@ def test_get_credential_key_with_extras(auth_config):
   assert original_key == key
   assert "extra_field" in auth_config.auth_scheme.model_extra
   assert "extra_field" in auth_config.raw_auth_credential.model_extra
+
+
+def test_credential_key_is_stable_across_python_hash_seed():
+  """Test AuthConfig key generation does not depend on PYTHONHASHSEED."""
+  repo_root = Path(__file__).resolve().parents[3]
+  pythonpath = str(repo_root / "src")
+  code = "\n".join([
+      "from fastapi.openapi.models import OAuth2",
+      "from fastapi.openapi.models import OAuthFlowAuthorizationCode",
+      "from fastapi.openapi.models import OAuthFlows",
+      "from google.adk.auth.auth_credential import AuthCredential",
+      "from google.adk.auth.auth_credential import AuthCredentialTypes",
+      "from google.adk.auth.auth_credential import OAuth2Auth",
+      "from google.adk.auth.auth_tool import AuthConfig",
+      "",
+      "auth_scheme = OAuth2(",
+      "  flows=OAuthFlows(",
+      "    authorizationCode=OAuthFlowAuthorizationCode(",
+      "      authorizationUrl='https://example.com/oauth2/authorize',",
+      "      tokenUrl='https://example.com/oauth2/token',",
+      "      scopes={'read': 'Read access'},",
+      "    )",
+      "  )",
+      ")",
+      "auth_cred = AuthCredential(",
+      "  auth_type=AuthCredentialTypes.OAUTH2,",
+      "  oauth2=OAuth2Auth(",
+      "    client_id='mock_client_id',",
+      "    client_secret='mock_client_secret',",
+      "  ),",
+      ")",
+      "print(AuthConfig(",
+      "  auth_scheme=auth_scheme,",
+      "  raw_auth_credential=auth_cred,",
+      ").credential_key)",
+  ])
+
+  def _run_with_seed(seed: str) -> str:
+    env = os.environ.copy()
+    env["PYTHONHASHSEED"] = seed
+    env["PYTHONPATH"] = os.pathsep.join(
+        [pythonpath, env.get("PYTHONPATH", "")]
+    ).strip(os.pathsep)
+    return subprocess.check_output(
+        [sys.executable, "-c", code],
+        env=env,
+        text=True,
+    ).strip()
+
+  assert _run_with_seed("0") == _run_with_seed("1")

tests/unittests/auth/test_auth_handler.py

@@ -348,10 +348,12 @@ class TestGenerateAuthRequest:
         exchanged_auth_credential=oauth2_credentials_with_auth_uri.model_copy(
             deep=True
         ),
+        credential_key="my_tool_tokens",
     )
     handler = AuthHandler(config)
     result = handler.generate_auth_request()
 
+    assert result.credential_key == "my_tool_tokens"
     assert (
         result.exchanged_auth_credential.oauth2.auth_uri
         == oauth2_credentials_with_auth_uri.oauth2.auth_uri
@@ -400,6 +402,31 @@ class TestGenerateAuthRequest:
     assert mock_generate_auth_uri.called
     assert result.exchanged_auth_credential == mock_credential
 
+  @patch("google.adk.auth.auth_handler.AuthHandler.generate_auth_uri")
+  def test_preserves_credential_key_on_generated_request(
+      self, mock_generate_auth_uri, oauth2_auth_scheme, oauth2_credentials
+  ):
+    """Test that AuthHandler preserves an explicit credential_key."""
+    mock_generate_auth_uri.return_value = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="mock_client_id",
+            client_secret="mock_client_secret",
+            auth_uri="https://example.com/generated",
+            state="generated_state",
+        ),
+    )
+
+    config = AuthConfig(
+        auth_scheme=oauth2_auth_scheme,
+        raw_auth_credential=oauth2_credentials,
+        credential_key="my_tool_tokens",
+    )
+    handler = AuthHandler(config)
+    result = handler.generate_auth_request()
+
+    assert result.credential_key == "my_tool_tokens"
+
 
 class TestGetAuthResponse:
   """Tests for the get_auth_response method."""

tests/unittests/auth/test_credential_manager.py

@@ -21,6 +21,8 @@ from fastapi.openapi.models import OAuth2
 from fastapi.openapi.models import OAuthFlowAuthorizationCode
 from fastapi.openapi.models import OAuthFlowImplicit
 from fastapi.openapi.models import OAuthFlows
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.agents.llm_agent import Agent
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.auth.auth_credential import AuthCredentialTypes
 from google.adk.auth.auth_credential import OAuth2Auth
@@ -33,8 +35,12 @@ from google.adk.auth.auth_tool import AuthConfig
 from google.adk.auth.credential_manager import CredentialManager
 from google.adk.auth.credential_manager import ServiceAccountCredentialExchanger
 from google.adk.auth.oauth2_discovery import AuthorizationServerMetadata
+from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.adk.tools.tool_context import ToolContext
 import pytest
 
+from .. import testing_utils
+
 
 class TestCredentialManager:
   """Test suite for CredentialManager."""
@@ -211,8 +217,30 @@ class TestCredentialManager:
   @pytest.mark.asyncio
   async def test_save_credential_with_service(self):
     """Test _save_credential with credential service."""
-    auth_config = Mock(spec=AuthConfig)
-    mock_credential = Mock(spec=AuthCredential)
+    auth_scheme = OAuth2(
+        flows=OAuthFlows(
+            authorizationCode=OAuthFlowAuthorizationCode(
+                authorizationUrl="https://example.com/oauth2/authorize",
+                tokenUrl="https://example.com/oauth2/token",
+                scopes={"read": "Read access"},
+            )
+        )
+    )
+    auth_config = AuthConfig(
+        auth_scheme=auth_scheme,
+        raw_auth_credential=AuthCredential(
+            auth_type=AuthCredentialTypes.OAUTH2,
+            oauth2=OAuth2Auth(
+                client_id="mock_client_id",
+                client_secret="mock_client_secret",
+            ),
+        ),
+        credential_key="test_key",
+    )
+    mock_credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(access_token="mock_access_token"),
+    )
 
     # Mock credential service
     credential_service = AsyncMock()
@@ -228,15 +256,39 @@ class TestCredentialManager:
     manager = CredentialManager(auth_config)
     await manager._save_credential(tool_context, mock_credential)
 
-    tool_context.save_credential.assert_called_once_with(auth_config)
-    assert auth_config.exchanged_auth_credential == mock_credential
+    tool_context.save_credential.assert_called_once()
+    saved_auth_config = tool_context.save_credential.call_args.args[0]
+    assert saved_auth_config.credential_key == "test_key"
+    assert saved_auth_config.exchanged_auth_credential == mock_credential
+    assert auth_config.exchanged_auth_credential is None
 
   @pytest.mark.asyncio
   async def test_save_credential_no_service(self):
     """Test _save_credential when no credential service is available."""
-    auth_config = Mock(spec=AuthConfig)
-    auth_config.exchanged_auth_credential = None
-    mock_credential = Mock(spec=AuthCredential)
+    auth_scheme = OAuth2(
+        flows=OAuthFlows(
+            authorizationCode=OAuthFlowAuthorizationCode(
+                authorizationUrl="https://example.com/oauth2/authorize",
+                tokenUrl="https://example.com/oauth2/token",
+                scopes={"read": "Read access"},
+            )
+        )
+    )
+    auth_config = AuthConfig(
+        auth_scheme=auth_scheme,
+        raw_auth_credential=AuthCredential(
+            auth_type=AuthCredentialTypes.OAUTH2,
+            oauth2=OAuth2Auth(
+                client_id="mock_client_id",
+                client_secret="mock_client_secret",
+            ),
+        ),
+        credential_key="test_key",
+    )
+    mock_credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(access_token="mock_access_token"),
+    )
 
     # Mock invocation context with no credential service
     invocation_context = Mock()
@@ -248,9 +300,86 @@ class TestCredentialManager:
     manager = CredentialManager(auth_config)
     await manager._save_credential(tool_context, mock_credential)
 
-    # Should not raise an error, and credential should be set in auth_config
-    # even when there's no credential service (config is updated regardless)
-    assert auth_config.exchanged_auth_credential == mock_credential
+    assert auth_config.exchanged_auth_credential is None
+
+  @pytest.mark.asyncio
+  async def test_request_credential_does_not_leak_across_users(self):
+    """Test that user-specific tokens are not cached on shared tool configs."""
+    auth_scheme = OAuth2(
+        flows=OAuthFlows(
+            authorizationCode=OAuthFlowAuthorizationCode(
+                authorizationUrl="https://example.com/oauth2/authorize",
+                tokenUrl="https://example.com/oauth2/token",
+                scopes={"read": "Read access"},
+            )
+        )
+    )
+    auth_config = AuthConfig(
+        auth_scheme=auth_scheme,
+        raw_auth_credential=AuthCredential(
+            auth_type=AuthCredentialTypes.OAUTH2,
+            oauth2=OAuth2Auth(
+                client_id="mock_client_id",
+                client_secret="mock_client_secret",
+            ),
+        ),
+        credential_key="shared_key",
+    )
+    manager = CredentialManager(auth_config)
+    agent = Agent(
+        name="root_agent",
+        model=testing_utils.MockModel.create(responses=[]),
+        tools=[],
+    )
+
+    session_service = InMemorySessionService()
+    session_a = await session_service.create_session(
+        app_name="test_app",
+        user_id="user_a",
+        session_id="session_a",
+    )
+    session_b = await session_service.create_session(
+        app_name="test_app",
+        user_id="user_b",
+        session_id="session_b",
+    )
+
+    invocation_context_a = InvocationContext(
+        session_service=session_service,
+        invocation_id="invocation_a",
+        agent=agent,
+        session=session_a,
+        credential_service=None,
+    )
+    invocation_context_b = InvocationContext(
+        session_service=session_service,
+        invocation_id="invocation_b",
+        agent=agent,
+        session=session_b,
+        credential_service=None,
+    )
+
+    tool_context_a = ToolContext(
+        invocation_context_a, function_call_id="call_a"
+    )
+    tool_context_b = ToolContext(
+        invocation_context_b, function_call_id="call_b"
+    )
+
+    tool_context_a.state["temp:" + auth_config.credential_key] = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            auth_uri="https://example.com/oauth2/authorize?x=y",
+            state="state_a",
+            access_token="token_a",
+            expires_at=9_999_999_999,
+        ),
+    )
+    await manager.get_auth_credential(tool_context_a)
+    await manager.request_credential(tool_context_b)
+
+    requested = tool_context_b.actions.requested_auth_configs["call_b"]
+    assert requested.exchanged_auth_credential.oauth2.access_token is None
 
   @pytest.mark.asyncio
   async def test_refresh_credential_oauth2(self):