Encrypt sensitive credential in session state with KMS

docs/guides/README.md

@@ -20,6 +20,9 @@ This directory contains specific developer guides for the ADK Python implementat
 ### Tools
 * [to_mcp_server](tools/mcp_tool/agent_to_mcp/index.md) - Expose an ADK agent as an MCP server so any MCP host can drive it as a single tool (the MCP counterpart of to_a2a).
 
+### Security
+* [Credentials Encryption](credentials_encryption.md) - Securely encrypting sensitive session credentials using GCP Secret Manager.
+
 ### Workflows
 * [Workflow](workflow/workflow/index.md) - Graph-based orchestration of complex, multi-step agent interactions.
 * [Workflow Graphs](workflow/graph/index.md) - Understanding nodes, edges, and graph structures in workflows.

docs/guides/credentials_encryption.md

@@ -0,0 +1,78 @@
+# Session Credentials Encryption Guide
+
+To prevent sensitive credentials from being stored in plaintext inside the session state database, ADK supports encrypting them using Google Cloud KMS. This applies to both **Google Cloud credentials** (using envelope encryption) and **generic credentials** (using direct KMS encryption).
+
+---
+
+## Supported Credential Types
+
+When a KMS key is configured, the following fields are automatically encrypted before being saved to the session database:
+* **API Keys** (`apiKey`)
+* **HTTP Authentication** (`password`, `token`)
+* **OAuth 2.0 Credentials** (`access_token`, `refresh_token`, `client_secret` for both GCP and third-party integrations)
+* **GCP Service Account Keys** (`private_key`)
+
+---
+
+## How It Works
+
+### 1. Google OAuth 2 Credentials (Envelope Encryption)
+* **Data Encryption Key (DEK)**: A local 256-bit symmetric key (Fernet) is generated locally to encrypt the sensitive fields (`access_token`, `refresh_token`, `client_secret`).
+* **Key Encryption Key (KEK)**: The Google Cloud KMS key acts as the KEK and is used to encrypt (wrap) the local DEK.
+* **Storage**: The session stores the locally encrypted credentials, the public reference of the KMS key (`kms_key_name`), and the encrypted DEK (`wrapped_dek`).
+
+### 2. Generic Credentials (`KmsEncryptedString`)
+* A custom Pydantic v2 type `KmsEncryptedString` is defined for sensitive model fields.
+* **Serialization (`model_dump()`)**: When saving to session state, the field value is encrypted via Cloud KMS and prefixed with `kms:`.
+* **Deserialization (`model_validate()`)**: When loading, the field value starting with `kms:` is decrypted back to plaintext.
+
+### 3. In-Memory Caching (Zero Latency)
+* To prevent performing a slow GCP KMS network request on every field encryption or decryption, the resolved plaintext DEK (envelope) and direct decrypted values are cached in-memory.
+* Decryption is processed **exactly once** per session load, and subsequent accesses are processed locally in-memory (instantaneous).
+
+### 4. Robust Re-Authentication Fallback
+If KMS decryption fails for any reason (e.g. the KMS key version has been **destroyed**, IAM permissions are **denied**, or the KMS service is **unreachable**):
+* ADK **does not crash the session**.
+* It logs a warning detailing the decryption failure.
+* It returns `None` for the loaded credentials, causing ADK to gracefully fall back to the standard authentication/exchanger flow. This prompts the user for re-authentication (asking for the API Key, OAuth, or password again) as if the credentials had expired or were lost.
+
+---
+
+## Configuration
+
+Set the environment variable `GOOGLE_CREDENTIAL_KMS_KEY` to point to your GCP KMS CryptoKey (optionally pinning a specific version):
+
+```bash
+export GOOGLE_CREDENTIAL_KMS_KEY="projects/{project_id}/locations/{location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}/cryptoKeyVersions/{version_id}"
+```
+
+Alternatively, you can configure it programmatically on any `CredentialsConfig` (like `BigQueryCredentialsConfig`):
+
+```python
+oauth_credentials_config = BigQueryCredentialsConfig(
+    client_id=client_id,
+    client_secret=client_secret,
+    scopes=scopes,
+    kms_key_name="projects/{project_id}/locations/{location}/keyRings/{key_ring_name}/cryptoKeys/{key_name}/cryptoKeyVersions/{version_id}"
+)
+```
+
+---
+
+## Required IAM Permissions
+
+The Service Account running the ADK Agent / Runner must be granted the appropriate permissions to call the Cloud KMS API.
+
+### KMS Permissions
+* **Role**: `Cloud KMS CryptoKey Encrypter/Decrypter` (`roles/cloudkms.cryptoKeyEncrypterDecrypter`)
+* **Scope**: Must be granted on the specified CryptoKey or KeyRing.
+
+Example `gcloud` command to grant access:
+
+```bash
+gcloud kms keys add-iam-policy-binding {key_name} \
+    --location={location} \
+    --keyring={key_ring_name} \
+    --member="serviceAccount:{agent_service_account}@{project_id}.iam.gserviceaccount.com" \
+    --role="roles/cloudkms.cryptoKeyEncrypterDecrypter"
+```

pyproject.toml

@@ -37,6 +37,7 @@ dependencies = [
   "click>=8.1.8,<9",
   "fastapi>=0.133,<1",
   "google-auth[pyopenssl]>=2.47",
+  "google-cloud-kms>=3,<4",
   "google-genai>=2.12.1,<3",
   "graphviz>=0.20.2,<1",
   "httpx>=0.27,<1",
@@ -315,7 +316,7 @@ known_third_party = [ "a2a", "google.adk" ]
 #   hel/serie/strin -> substrings in test fixtures; te -> local variable;
 #   rouge -> the ROUGE metric; unparseable -> valid spelling variant;
 #   re-use/re-used -> intentional hyphenation; lamda -> Google LaMDA project.
-ignore-words-list = "hel,serie,strin,te,rouge,unparseable,re-use,re-used,lamda"
+ignore-words-list = "hel,serie,strin,te,rouge,unparseable,re-use,re-used,lamda,astroid"
 # CHANGELOG.md is generated from commit messages; lockfiles, notebooks, JSON
 # fixtures, bundled JS/source maps, and the vendored CLI browser bundle are
 # generated or data files, not prose we own.

src/google/adk/auth/__init__.py

@@ -34,4 +34,4 @@ def __getattr__(name: str) -> type[AuthHandler]:
     from .auth_handler import AuthHandler
 
     return AuthHandler
-  raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+  raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/google/adk/auth/_kms_encryptor.py

@@ -0,0 +1,198 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import base64
+import logging
+from typing import Dict
+from typing import Tuple
+
+from cryptography.fernet import Fernet
+
+logger = logging.getLogger("google_adk." + __name__)
+
+# Cache mapping KMS key name -> tuple of (plaintext_dek: bytes, wrapped_dek: str)
+_KMS_KEY_DEK_CACHE: Dict[str, Tuple[bytes, str]] = {}
+
+# Cache mapping wrapped_dek (str) -> Fernet instance
+_DEK_FERNET_CACHE: Dict[str, Fernet] = {}
+
+# KMS Client cache
+_KMS_CLIENT_CACHE: Dict[str, any] = {}
+
+
+def _get_kms_client(kms_key_name: str):
+  """Gets or creates a cached Google Cloud KMS client."""
+  if kms_key_name not in _KMS_CLIENT_CACHE:
+    from google.cloud import kms
+
+    _KMS_CLIENT_CACHE[kms_key_name] = kms.KeyManagementServiceClient()
+  return _KMS_CLIENT_CACHE[kms_key_name]
+
+
+def _get_or_create_dek(kms_key_name: str) -> Tuple[bytes, str]:
+  """Gets the cached DEK for a KMS key, or generates and wraps a new one."""
+  if kms_key_name not in _KMS_KEY_DEK_CACHE:
+    try:
+      # Generate a new 32-byte Fernet key
+      plaintext_dek = Fernet.generate_key()
+
+      # Wrap (encrypt) the DEK using Cloud KMS
+      client = _get_kms_client(kms_key_name)
+      response = client.encrypt(
+          request={
+              "name": kms_key_name,
+              "plaintext": plaintext_dek,
+          }
+      )
+      wrapped_dek = base64.b64encode(response.ciphertext).decode("utf-8")
+      _KMS_KEY_DEK_CACHE[kms_key_name] = (plaintext_dek, wrapped_dek)
+      # Also populate the Fernet cache for this wrapped DEK
+      _DEK_FERNET_CACHE[wrapped_dek] = Fernet(plaintext_dek)
+    except Exception as e:
+      logger.error(
+          "Failed to generate and wrap DEK using KMS key %s: %s",
+          kms_key_name,
+          e,
+      )
+      raise e
+
+  return _KMS_KEY_DEK_CACHE[kms_key_name]
+
+
+def _get_fernet_for_wrapped_dek(kms_key_name: str, wrapped_dek: str) -> Fernet:
+  """Gets the cached Fernet instance for a wrapped DEK, unwrapping it with KMS if needed."""
+  if wrapped_dek not in _DEK_FERNET_CACHE:
+    try:
+      # Unwrap (decrypt) the DEK using Cloud KMS
+      client = _get_kms_client(kms_key_name)
+      ciphertext_bytes = base64.b64decode(wrapped_dek.encode("utf-8"))
+      response = client.decrypt(
+          request={
+              "name": kms_key_name,
+              "ciphertext": ciphertext_bytes,
+          }
+      )
+      plaintext_dek = response.plaintext
+      _DEK_FERNET_CACHE[wrapped_dek] = Fernet(plaintext_dek)
+    except Exception as e:
+      logger.error("Failed to unwrap DEK using KMS key %s: %s", kms_key_name, e)
+      raise e
+
+  return _DEK_FERNET_CACHE[wrapped_dek]
+
+
+def encrypt_credentials(
+    kms_key_name: str,
+    token: str | None,
+    refresh_token: str | None,
+    client_secret: str | None,
+) -> Tuple[str | None, str | None, str | None, str | None]:
+  """Encrypts the sensitive credential fields using envelope encryption.
+
+  Returns a tuple of (encrypted_token, encrypted_refresh_token, encrypted_client_secret, wrapped_dek).
+  """
+  try:
+    plaintext_dek, wrapped_dek = _get_or_create_dek(kms_key_name)
+    fernet = _DEK_FERNET_CACHE[wrapped_dek]
+
+    enc_token = (
+        fernet.encrypt(token.encode("utf-8")).decode("utf-8") if token else None
+    )
+    enc_refresh = (
+        fernet.encrypt(refresh_token.encode("utf-8")).decode("utf-8")
+        if refresh_token
+        else None
+    )
+    enc_secret = (
+        fernet.encrypt(client_secret.encode("utf-8")).decode("utf-8")
+        if client_secret
+        else None
+    )
+
+    return enc_token, enc_refresh, enc_secret, wrapped_dek
+  except Exception as e:
+    logger.error("Failed to encrypt credentials: %s", e)
+    raise e
+
+
+def decrypt_credentials(
+    kms_key_name: str,
+    encrypted_token: str | None,
+    encrypted_refresh_token: str | None,
+    encrypted_client_secret: str | None,
+    wrapped_dek: str | None,
+) -> Tuple[str | None, str | None, str | None]:
+  """Decrypts the sensitive credential fields using the wrapped DEK."""
+  if not wrapped_dek:
+    # Backward compatibility
+    return encrypted_token, encrypted_refresh_token, encrypted_client_secret
+
+  try:
+    fernet = _get_fernet_for_wrapped_dek(kms_key_name, wrapped_dek)
+
+    dec_token = (
+        fernet.decrypt(encrypted_token.encode("utf-8")).decode("utf-8")
+        if encrypted_token
+        else None
+    )
+    dec_refresh = (
+        fernet.decrypt(encrypted_refresh_token.encode("utf-8")).decode("utf-8")
+        if encrypted_refresh_token
+        else None
+    )
+    dec_secret = (
+        fernet.decrypt(encrypted_client_secret.encode("utf-8")).decode("utf-8")
+        if encrypted_client_secret
+        else None
+    )
+
+    return dec_token, dec_refresh, dec_secret
+  except Exception as e:
+    logger.error("Failed to decrypt credentials: %s", e)
+    raise e
+
+
+def encrypt_value(kms_key_name: str, plaintext: str) -> str:
+  """Fallback/Direct encryption helper."""
+  try:
+    client = _get_kms_client(kms_key_name)
+    response = client.encrypt(
+        request={
+            "name": kms_key_name,
+            "plaintext": plaintext.encode("utf-8"),
+        }
+    )
+    return base64.b64encode(response.ciphertext).decode("utf-8")
+  except Exception as e:
+    logger.error("Failed to encrypt value: %s", e)
+    raise e
+
+
+def decrypt_value(kms_key_name: str, ciphertext: str) -> str:
+  """Fallback/Direct decryption helper."""
+  try:
+    client = _get_kms_client(kms_key_name)
+    ciphertext_bytes = base64.b64decode(ciphertext.encode("utf-8"))
+    response = client.decrypt(
+        request={
+            "name": kms_key_name,
+            "ciphertext": ciphertext_bytes,
+        }
+    )
+    return response.plaintext.decode("utf-8")
+  except Exception as e:
+    logger.error("Failed to decrypt value: %s", e)
+    raise e