diff --git a/src/quant_platform_kit/cloud/gcp_provider.py b/src/quant_platform_kit/cloud/gcp_provider.py index 3f0b87ee..c4054f60 100644 --- a/src/quant_platform_kit/cloud/gcp_provider.py +++ b/src/quant_platform_kit/cloud/gcp_provider.py @@ -6,13 +6,40 @@ from __future__ import annotations import os - +import re +from dataclasses import dataclass, field # ══════════════════════════════════════════════════════════════════════ # Secret Store — GCP Secret Manager # ══════════════════════════════════════════════════════════════════════ +_CONCRETE_SECRET_VERSION = re.compile( + r"^projects/([^/]+)/secrets/([^/]+)/versions/([1-9][0-9]*)$" +) + + +@dataclass(frozen=True, slots=True) +class SecretValueWithMetadata: + """Payload from one read, plus a concrete version name when the response proves it. + + ``value`` is omitted from repr and this object is not serialized or logged here. + """ + + value: str = field(repr=False) + version_name: str | None + + +def _concrete_version_name(resource_name: object, secret_name: str) -> str | None: + """Keep a concrete version resource name; never include a payload in failures.""" + if not isinstance(resource_name, str): + return None + match = _CONCRETE_SECRET_VERSION.fullmatch(resource_name) + if match is None or match.group(2) != secret_name: + return None + return resource_name + + class GcpSecretStore: """Read-only secret access via GCP Secret Manager.""" @@ -25,6 +52,20 @@ def get_secret(self, secret_name: str, *, project_id: str | None = None) -> str: response = client.access_secret_version(request={"name": name}) return response.payload.data.decode("utf-8") + def get_secret_with_metadata( + self, secret_name: str, *, project_id: str | None = None + ) -> SecretValueWithMetadata: + """Read latest once and keep its concrete version name when the response proves it.""" + import google.cloud.secretmanager_v1 as secret_manager + + pid = project_id or _resolve_project_id() + client = secret_manager.SecretManagerServiceClient() + name = f"projects/{pid}/secrets/{secret_name}/versions/latest" + response = client.access_secret_version(request={"name": name}) + value = response.payload.data.decode("utf-8") + version_name = _concrete_version_name(getattr(response, "name", None), secret_name) + return SecretValueWithMetadata(value=value, version_name=version_name) + class GcpSecretStoreReadWrite: """Read-write secret access for token rotation scenarios.""" diff --git a/tests/test_cloud_abstraction.py b/tests/test_cloud_abstraction.py index ebf10b4b..7a1e52d1 100644 --- a/tests/test_cloud_abstraction.py +++ b/tests/test_cloud_abstraction.py @@ -693,3 +693,113 @@ def test_normal_document_operations_keep_original_sdk_calls(self): self.store.delete("items", "one") self.document.delete.assert_called_once_with() self.client.transaction.assert_not_called() + + +class GcpSecretMetadataTests(unittest.TestCase): + """Mock SDK checks for one-shot read-only secret version metadata.""" + + SYNTHETIC = "synthetic-r10-snapshot-value" + + def _response(self, name, payload=None): + from types import SimpleNamespace + + data = self.SYNTHETIC.encode("utf-8") if payload is None else payload + return SimpleNamespace(payload=SimpleNamespace(data=data), name=name) + + def _read( + self, + response, + *, + secret_name="lb-read-token", + project_id="my-project", + reader="metadata", + responses=None, + ): + import sys + from types import ModuleType + from unittest.mock import MagicMock, patch + + client = MagicMock() + client.access_secret_version.side_effect = list(responses) if responses is not None else [response] + google = ModuleType("google") + google_cloud = ModuleType("google.cloud") + secret_manager = ModuleType("google.cloud.secretmanager_v1") + secret_manager.SecretManagerServiceClient = MagicMock(return_value=client) + google.cloud = google_cloud + google_cloud.secretmanager_v1 = secret_manager + modules = { + "google": google, + "google.cloud": google_cloud, + "google.cloud.secretmanager_v1": secret_manager, + } + from quant_platform_kit.cloud.gcp_provider import GcpSecretStore + + with patch.dict(sys.modules, modules): + store = GcpSecretStore() + if reader == "get": + result = store.get_secret(secret_name, project_id=project_id) + else: + result = store.get_secret_with_metadata(secret_name, project_id=project_id) + request_name = f"projects/{project_id}/secrets/{secret_name}/versions/latest" + client.access_secret_version.assert_called_once_with(request={"name": request_name}) + return result + + def test_one_access_keeps_payload_and_concrete_version(self): + from quant_platform_kit.cloud.gcp_provider import SecretValueWithMetadata + + version_name = "projects/987654321/secrets/lb-read-token/versions/7" + result = self._read(self._response(version_name)) + self.assertIsInstance(result, SecretValueWithMetadata) + self.assertEqual(result.value, self.SYNTHETIC) + self.assertEqual(result.version_name, version_name) + self.assertNotIn(self.SYNTHETIC, repr(result)) + + def test_later_latest_change_is_not_read(self): + first = self._response("projects/987654321/secrets/lb-read-token/versions/7") + rotated = self._response( + "projects/987654321/secrets/lb-read-token/versions/8", + payload=b"synthetic-r10-rotated-value", + ) + result = self._read(first, responses=(first, rotated)) + self.assertEqual(result.version_name, "projects/987654321/secrets/lb-read-token/versions/7") + self.assertEqual(result.value, self.SYNTHETIC) + self.assertNotEqual(result.value, "synthetic-r10-rotated-value") + + def test_missing_alias_or_wrong_secret_keeps_payload_without_version(self): + names = ( + None, + "", + "projects/987654321/secrets/lb-read-token/versions/latest", + "projects/987654321/secrets/other-secret/versions/7", + "projects/987654321/secrets/lb-read-token/versions/0", + "projects/987654321/secrets/lb-read-token/versions/01", + "not-a-resource-name", + ) + for name in names: + with self.subTest(name=name): + result = self._read(self._response(name)) + self.assertEqual(result.value, self.SYNTHETIC) + self.assertIsNone(result.version_name) + self.assertNotIn(self.SYNTHETIC, repr(result)) + + def test_decode_error_stays_unicode_error_without_added_wrapper(self): + response = self._response( + "projects/987654321/secrets/lb-read-token/versions/7", + payload=b"\xff", + ) + with self.assertRaises(UnicodeDecodeError) as caught: + self._read(response) + self.assertNotIn(self.SYNTHETIC, str(caught.exception)) + + def test_get_secret_still_returns_plaintext_from_latest(self): + result = self._read( + self._response("projects/987654321/secrets/lb-read-token/versions/7"), + reader="get", + ) + self.assertEqual(result, self.SYNTHETIC) + self.assertIsInstance(result, str) + + def test_read_write_store_has_no_metadata_method(self): + from quant_platform_kit.cloud.gcp_provider import GcpSecretStoreReadWrite + + self.assertFalse(hasattr(GcpSecretStoreReadWrite, "get_secret_with_metadata"))