Encrypt sensitive credential in session state with KMS

[guillaumeblaquiere]

Link to Issue and Description of Change

• Related: "Secret" session state scope #5112, but do not solve it exactly

Problem:
During authentication flows, sensitive credentials (including OAuth 2.0 access/refresh tokens, basic authentication passwords, API keys, and Service Account private keys) are persisted in plaintext inside the session state database. This exposes sensitive credentials to unauthorized access if the session database is compromised.

Solution:
Implemented a comprehensive, highly secure, and optimized envelope and direct encryption mechanism utilizing Google Cloud KMS to protect all sensitive credential types before persisting them, with a robust re-authentication fallback in case of decryption failures:

1. Envelope Encryption for Google OAuth 2 Credentials: Introduced a local symmetric Data Encryption Key (DEK) via Fernet to encrypt credentials locally, and wrapped (encrypted) the DEK using the Cloud KMS key (acting as the Key Encryption Key, KEK).
2. Direct KMS Encryption via Custom Pydantic Type (KmsEncryptedString): Created a custom type using Pydantic v2 BeforeValidator and PlainSerializer to automatically encrypt sensitive fields during serialization (model_dump()) and decrypt them during validation/deserialization (model_validate()). Used this type for:
   • HttpCredentials (password, token)
   • OAuth2Auth (access_token, refresh_token, client_secret)
   • ServiceAccountCredential (private_key)
   • AuthCredential (api_key)
3. In-Memory Caching (Zero Latency): Cached the KMS client and decrypted keys in-memory (_KMS_KEY_DEK_CACHE and _DEK_FERNET_CACHE). Deserialization requires calling Cloud KMS only once per session load, and subsequent encryptions/decryptions are processed locally in-memory (instantaneous).
4. Re-Authentication Fallback (No-Crash): Wrapped deserialization and decryption calls in try-except blocks across SessionStateCredentialService.load_credential, AuthHandler.get_auth_response, and GoogleCredentialsManager.get_valid_credentials. If decryption fails (e.g. KMS key destroyed or IAM permissions revoked), the system logs a warning, suppresses the exception, and returns None to gracefully trigger the standard re-authentication flow (re-prompting the user for credentials) instead of crashing the runner.
5. Backward Compatibility: Standard plaintext credentials and sessions without KMS configurations continue to load and save normally without errors.
6. Dependencies: Added google-cloud-kms under required dependencies in pyproject.toml.

---

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Summary of pytest results:
Added and updated tests in tests/unittests/auth/test_kms_credentials.py mocking the KMS client (avoiding network dependency) to assert:

• Local DEK generation, caching, and envelope encryption/decryption.
• Serialization and deserialization of KmsEncryptedString fields (verifying kms: prefixes and decryption back to plaintext).
• Backward compatibility when loading legacy plaintext credentials.
• Fallback behavior: verifying that KMS decryption failures (key destroyed or permissions missing) safely return None (triggering re-authentication) rather than crashing the session load.

All 210 authentication unit tests passed successfully:

pytest tests/unittests/auth/
====================== 210 passed, 77 warnings in 10.44s =======================