fix: handle case-insensitive retrieval in GoogleSecretManagerSettingsSource (#730)

Author: ezwiefel

docs/index.md

@@ -2452,6 +2452,9 @@ For nested models, Secret Manager supports the `env_nested_delimiter` setting as
 ### Important Notes
 
 1. **Case Sensitivity**: By default, secret names are case-sensitive.
+    *   If you set `case_sensitive=False`, `pydantic-settings` will attempt to resolve secrets in a case-insensitive manner. It prioritizes exact matches over case-insensitive matches. For some examples of this, imagine `case_sensitive=False` and the model attribute is named `my_secret`:
+        * If Google Secret Manager has both `MY_SECRET` and `my_secret` defined - the value of `my_secret` will be returned.
+        * If Google Secret Manager has `MY_SECRET`, `My_Secret`, and `my_Secret` defined - a warning will be raised and the value of `my_Secret` will be returned - as the secret names are first sorted in ASCII sort order (where lowercased letters are greater than upper case letters) and the last one is chosen (which would be `my_Secret` in this case).
 2. **Secret Naming**: Create secrets in Google Secret Manager with names that match your field names (including any prefix). According the [Secret Manager documentation](https://cloud.google.com/secret-manager/docs/creating-and-accessing-secrets#create-a-secret), a secret name can contain uppercase and lowercase letters, numerals, hyphens, and underscores. The maximum allowed length for a name is 255 characters.
 3. **Secret Versions**: The GoogleSecretManagerSettingsSource uses the "latest" version of secrets.
 

pydantic_settings/sources/providers/gcp.py

@@ -1,5 +1,6 @@
 from __future__ import annotations as _annotations
 
+import warnings
 from collections.abc import Iterator, Mapping
 from functools import cached_property
 from typing import TYPE_CHECKING
@@ -47,36 +48,69 @@ def __init__(self, secret_client: SecretManagerServiceClient, project_id: str, c
     def _gcp_project_path(self) -> str:
         return self._secret_client.common_project_path(self._project_id)
 
+    def _select_case_insensitive_secret(self, lower_name: str, candidates: list[str]) -> str:
+        if len(candidates) == 1:
+            return candidates[0]
+
+        # Sort to ensure deterministic selection (prefer lowercase / ASCII last)
+        candidates.sort()
+        winner = candidates[-1]
+        warnings.warn(
+            f"Secret collision: Found multiple secrets {candidates} normalizing to '{lower_name}'. "
+            f"Using '{winner}' for case-insensitive lookup.",
+            UserWarning,
+            stacklevel=2,
+        )
+        return winner
+
     @cached_property
-    def _secret_names(self) -> list[str]:
-        rv: list[str] = []
+    def _secret_name_map(self) -> dict[str, str]:
+        mapping: dict[str, str] = {}
+        # Group secrets by normalized name to detect collisions
+        normalized_groups: dict[str, list[str]] = {}
 
         secrets = self._secret_client.list_secrets(parent=self._gcp_project_path)
         for secret in secrets:
             name = self._secret_client.parse_secret_path(secret.name).get('secret', '')
+            mapping[name] = name
+
             if not self._case_sensitive:
-                name = name.lower()
-            rv.append(name)
-        return rv
+                lower_name = name.lower()
+                if lower_name not in normalized_groups:
+                    normalized_groups[lower_name] = []
+                normalized_groups[lower_name].append(name)
+
+        if not self._case_sensitive:
+            for lower_name, candidates in normalized_groups.items():
+                mapping[lower_name] = self._select_case_insensitive_secret(lower_name, candidates)
+
+        return mapping
+
+    @property
+    def _secret_names(self) -> list[str]:
+        return list(self._secret_name_map.keys())
 
     def _secret_version_path(self, key: str, version: str = 'latest') -> str:
         return self._secret_client.secret_version_path(self._project_id, key, version)
 
     def __getitem__(self, key: str) -> str | None:
-        if not self._case_sensitive:
-            key = key.lower()
-        if key not in self._loaded_secrets:
-            # If we know the key isn't available in secret manager, raise a key error
-            if key not in self._secret_names:
-                raise KeyError(key)
+        if key in self._loaded_secrets:
+            return self._loaded_secrets[key]
+
+        gcp_secret_name = self._secret_name_map.get(key)
+        if gcp_secret_name is None and not self._case_sensitive:
+            gcp_secret_name = self._secret_name_map.get(key.lower())
 
+        if gcp_secret_name:
             try:
                 self._loaded_secrets[key] = self._secret_client.access_secret_version(
-                    name=self._secret_version_path(key)
+                    name=self._secret_version_path(gcp_secret_name)
                 ).payload.data.decode('UTF-8')
             except Exception:
                 # If we can't access the secret, we return None
                 self._loaded_secrets[key] = None
+        else:
+            raise KeyError(key)
 
         return self._loaded_secrets[key]
 

tests/test_source_gcp_secret_manager.py

@@ -208,7 +208,7 @@ def settings_customise_sources(
                 )
 
         with pytest.raises(ValidationError):
-            _ = Settings()
+            _ = Settings()  # type: ignore
 
     def test_pydantic_base_settings_with_default_value(self, mock_secret_client):
         class Settings(BaseSettings):
@@ -242,3 +242,118 @@ def test_secret_manager_mapping_list_secrets_error(self, secret_manager_mapping,
 
         with pytest.raises(Exception, match='Permission denied'):
             _ = secret_manager_mapping._secret_names
+
+    @pytest.mark.parametrize(
+        'case_sensitive, secret_name_in_gcp, requested_key, expected_value',
+        [
+            (True, 'test-secret', 'test-secret', 'test-value'),
+            (True, 'TEST-SECRET', 'TEST-SECRET', 'test-value'),
+            (True, 'testSecret', 'testSecret', 'test-value'),
+            (True, 'TEST-SECRET', 'test-secret', None),
+            (True, 'test-secret', 'TEST_SECRET', None),
+            (False, 'test-secret', 'TEST-SECRET', 'test-value'),
+            (False, 'TEST-SECRET', 'test-secret', 'test-value'),
+            (False, 'TEST-SECRET', 'TEST-SECRET', 'test-value'),
+            (False, 'testSecret', 'testSecret', 'test-value'),
+            (False, 'testSecret', 'TESTSECRET', 'test-value'),
+        ],
+    )
+    def test_secret_manager_mapping_retrieval_cases(
+        self, mocker, case_sensitive, secret_name_in_gcp, requested_key, expected_value
+    ):
+        """
+        Tests various combinations of case sensitivity and secret naming.
+        """
+        client = mocker.Mock(spec=SecretManagerServiceClient)
+        client.common_project_path.return_value = 'projects/test-project'
+        client.secret_version_path = (
+            lambda project, secret, version: f'projects/{project}/secrets/{secret}/versions/{version}'
+        )
+        client.parse_secret_path = SecretManagerServiceClient.parse_secret_path
+
+        # Mock list_secrets to return the specific secret name
+        secret = mocker.Mock()
+        secret.name = f'projects/test-project/secrets/{secret_name_in_gcp}'
+        client.list_secrets.return_value = [secret]
+
+        secret_response = mocker.Mock()
+        secret_response.payload.data.decode.return_value = 'test-value'
+
+        def mock_access_secret_version(name: str):
+            # GCP is always case-sensitive
+            if name == f'projects/test-project/secrets/{secret_name_in_gcp}/versions/latest':
+                return secret_response
+            raise Exception(f'Secret not found or access denied: {name}')
+
+        client.access_secret_version = mock_access_secret_version
+
+        mapping = GoogleSecretManagerMapping(client, project_id='test-project', case_sensitive=case_sensitive)
+
+        if expected_value is None:
+            # Depending on implementation, it might raise KeyError or return None if we try to access it via .get() or handled access
+            # The Mapping __getitem__ implementation in pydantic-settings currently returns None if access fails
+            # OR raises KeyError if the key isn't in the list at all.
+
+            # If the key is not in _secret_names, it raises KeyError.
+            # If it IS in _secret_names but access fails, it returns None.
+
+            # For case (True, 'TEST-SECRET', 'test-secret', None):
+            # _secret_names will be ['TEST-SECRET']. 'test-secret' is not in there. KeyError expected.
+            try:
+                val = mapping[requested_key]
+                assert val == expected_value
+            except KeyError:
+                assert expected_value is None
+        else:
+            assert mapping[requested_key] == expected_value
+
+    @pytest.mark.parametrize(
+        'case_sensitive, requested_key, expected_value',
+        [
+            (True, 'TEST-SECRET', 'UPPER_VAL'),
+            (True, 'test-secret', 'lower_val'),
+            # Case insensitive collision with "Prefer Exact Match" logic:
+            (False, 'TEST-SECRET', 'UPPER_VAL'),  # Exact match exists, prefer it
+            (False, 'test-secret', 'lower_val'),  # Exact match exists, prefer it
+            (False, 'Test-Secret', 'lower_val'),  # No exact match, fallback to 'lower_val' (last loaded)
+        ],
+    )
+    def test_secret_manager_mapping_collision(self, mocker, case_sensitive, requested_key, expected_value):
+        client = mocker.Mock(spec=SecretManagerServiceClient)
+        client.common_project_path.return_value = 'projects/test-project'
+        client.secret_version_path = (
+            lambda project, secret, version: f'projects/{project}/secrets/{secret}/versions/{version}'
+        )
+        client.parse_secret_path = SecretManagerServiceClient.parse_secret_path
+
+        # Mock list_secrets with colliding names
+        secrets = []
+        for name in ['TEST-SECRET', 'test-secret']:
+            s = mocker.Mock()
+            s.name = f'projects/test-project/secrets/{name}'
+            secrets.append(s)
+        client.list_secrets.return_value = secrets
+
+        def mock_access_secret_version(name: str):
+            # name format: projects/test-project/secrets/{SECRET_ID}/versions/latest
+            if '/secrets/TEST-SECRET/' in name:
+                resp = mocker.Mock()
+                resp.payload.data.decode.return_value = 'UPPER_VAL'
+                return resp
+            elif '/secrets/test-secret/' in name:
+                resp = mocker.Mock()
+                resp.payload.data.decode.return_value = 'lower_val'
+                return resp
+            raise Exception(f'Secret not found: {name}')
+
+        client.access_secret_version = mock_access_secret_version
+
+        mapping = GoogleSecretManagerMapping(client, project_id='test-project', case_sensitive=case_sensitive)
+
+        if not case_sensitive:
+            with pytest.warns(UserWarning, match='Secret collision'):
+                _ = mapping._secret_name_map
+        else:
+            _ = mapping._secret_name_map
+
+        assert mapping[requested_key] == expected_value