fix(auth): migrate credential storage to secret: scope (Phase 2)

Migrate existing credential writers to use the `secret:` prefix so
that OAuth tokens and credentials are never persisted to session
storage backends.

- Change BIGQUERY_TOKEN_CACHE_KEY to "secret:bigquery_token_cache"
- Update SessionStateCredentialService.save_credential and
  load_credential to prefix credential_key with State.SECRET_PREFIX
- Add backward-compatible fallback: load paths try the secret-prefixed
  key first, then fall back to the legacy unprefixed key so existing
  sessions migrate without re-authentication
- Update and add tests for prefixed keys and legacy fallback

Depends on google#5132 (Phase 1: secret: scope infrastructure)
Closes google#5112

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caohy1988

src/google/adk/auth/credential_service/session_state_credential_service.py

@@ -19,6 +19,7 @@
 from typing_extensions import override
 
 from ...agents.callback_context import CallbackContext
+from ...sessions.state import State
 from ...utils.feature_decorator import experimental
 from ..auth_credential import AuthCredential
 from ..auth_tool import AuthConfig
@@ -54,6 +55,13 @@ async def load_credential(
         Optional[AuthCredential]: the credential saved in the store.
 
     """
+    # Try secret-scoped key first, fall back to legacy unprefixed key
+    # so existing sessions migrate without re-authentication.
+    result = callback_context.state.get(
+        State.SECRET_PREFIX + auth_config.credential_key
+    )
+    if result is not None:
+      return result
     return callback_context.state.get(auth_config.credential_key)
 
   @override
@@ -78,6 +86,6 @@ async def save_credential(
         None
     """
 
-    callback_context.state[auth_config.credential_key] = (
+    callback_context.state[State.SECRET_PREFIX + auth_config.credential_key] = (
         auth_config.exchanged_auth_credential
     )

src/google/adk/tools/_google_credentials.py

@@ -171,11 +171,13 @@ async def get_valid_credentials(
             f" {self.credentials_config.external_access_token_key}."
         )
     # First, try to get credentials from the tool context
-    creds_json = (
-        tool_context.state.get(self.credentials_config._token_cache_key, None)
-        if self.credentials_config._token_cache_key
-        else None
-    )
+    cache_key = self.credentials_config._token_cache_key
+    creds_json = tool_context.state.get(cache_key, None) if cache_key else None
+    # Fall back to legacy unprefixed key so existing sessions migrate
+    # without re-authentication.
+    if creds_json is None and cache_key and cache_key.startswith("secret:"):
+      legacy_key = cache_key[len("secret:") :]
+      creds_json = tool_context.state.get(legacy_key, None)
     creds = (
         google.oauth2.credentials.Credentials.from_authorized_user_info(
             json.loads(creds_json), self.credentials_config.scopes

src/google/adk/tools/bigquery/bigquery_credentials.py

@@ -18,7 +18,7 @@
 from ...features import FeatureName
 from .._google_credentials import BaseGoogleCredentialsConfig
 
-BIGQUERY_TOKEN_CACHE_KEY = "bigquery_token_cache"
+BIGQUERY_TOKEN_CACHE_KEY = "secret:bigquery_token_cache"
 BIGQUERY_SCOPES = [
     "https://www.googleapis.com/auth/bigquery",
     "https://www.googleapis.com/auth/dataplex.read-write",

tests/unittests/auth/credential_service/test_session_state_credential_service.py

@@ -23,6 +23,7 @@
 from google.adk.auth.auth_credential import OAuth2Auth
 from google.adk.auth.auth_tool import AuthConfig
 from google.adk.auth.credential_service.session_state_credential_service import SessionStateCredentialService
+from google.adk.sessions.state import State
 import pytest
 
 
@@ -265,10 +266,11 @@ async def test_state_persistence_across_operations(
     # Save credential
     await credential_service.save_credential(auth_config, callback_context)
 
-    # Verify state contains the credential
-    assert auth_config.credential_key in callback_context.state
+    # Verify state contains the credential under secret: prefix
+    secret_key = State.SECRET_PREFIX + auth_config.credential_key
+    assert secret_key in callback_context.state
     assert (
-        callback_context.state[auth_config.credential_key]
+        callback_context.state[secret_key]
         == auth_config.exchanged_auth_credential
     )
 
@@ -279,9 +281,9 @@ async def test_state_persistence_across_operations(
     assert result is not None
 
     # Verify state still contains the credential
-    assert auth_config.credential_key in callback_context.state
+    assert secret_key in callback_context.state
     assert (
-        callback_context.state[auth_config.credential_key]
+        callback_context.state[secret_key]
         == auth_config.exchanged_auth_credential
     )
 
@@ -300,7 +302,7 @@ async def test_state_persistence_across_operations(
     await credential_service.save_credential(auth_config, callback_context)
 
     # Verify state was updated
-    assert callback_context.state[auth_config.credential_key] == new_credential
+    assert callback_context.state[secret_key] == new_credential
 
   @pytest.mark.asyncio
   async def test_credential_key_uniqueness(
@@ -344,13 +346,12 @@ async def test_credential_key_uniqueness(
     await credential_service.save_credential(auth_config1, callback_context)
     await credential_service.save_credential(auth_config2, callback_context)
 
-    # Verify both exist in state with different keys
-    assert "unique_key_1" in callback_context.state
-    assert "unique_key_2" in callback_context.state
-    assert (
-        callback_context.state["unique_key_1"]
-        != callback_context.state["unique_key_2"]
-    )
+    # Verify both exist in state with secret-prefixed keys
+    sk1 = State.SECRET_PREFIX + "unique_key_1"
+    sk2 = State.SECRET_PREFIX + "unique_key_2"
+    assert sk1 in callback_context.state
+    assert sk2 in callback_context.state
+    assert callback_context.state[sk1] != callback_context.state[sk2]
 
     # Load and verify both credentials
     result1 = await credential_service.load_credential(
@@ -379,10 +380,62 @@ async def test_direct_state_access(
             redirect_uri="https://direct.com/callback",
         ),
     )
-    callback_context.state[auth_config.credential_key] = test_credential
+    callback_context.state[State.SECRET_PREFIX + auth_config.credential_key] = (
+        test_credential
+    )
 
     # Load using the service
     result = await credential_service.load_credential(
         auth_config, callback_context
     )
     assert result == test_credential
+
+  @pytest.mark.asyncio
+  async def test_load_falls_back_to_legacy_unprefixed_key(
+      self, credential_service, auth_config, callback_context
+  ):
+    """Credentials stored under the old unprefixed key are still found."""
+    legacy_cred = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="legacy_client",
+            client_secret="legacy_secret",
+        ),
+    )
+    # Simulate a session persisted before the secret: migration
+    callback_context.state[auth_config.credential_key] = legacy_cred
+
+    result = await credential_service.load_credential(
+        auth_config, callback_context
+    )
+    assert result is not None
+    assert result.oauth2.client_id == "legacy_client"
+
+  @pytest.mark.asyncio
+  async def test_secret_key_takes_precedence_over_legacy(
+      self, credential_service, auth_config, callback_context
+  ):
+    """When both keys exist, the secret-prefixed key wins."""
+    old_cred = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="old_client",
+            client_secret="old_secret",
+        ),
+    )
+    new_cred = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="new_client",
+            client_secret="new_secret",
+        ),
+    )
+    callback_context.state[auth_config.credential_key] = old_cred
+    callback_context.state[State.SECRET_PREFIX + auth_config.credential_key] = (
+        new_cred
+    )
+
+    result = await credential_service.load_credential(
+        auth_config, callback_context
+    )
+    assert result.oauth2.client_id == "new_client"