Fixes 4362: preserve secret: prefix during Python SDK serialization (#28625)

CustomSecretStr serialization resolved `secret:<id>` external references
through the configured secrets manager, stripping the `secret:` prefix from
model_dump / model_dump_json output. On create/update the SDK then sent the
resolved (or, with the default DB manager, prefix-stripped) value to the
server, so external references were stored as plain secrets instead of being
preserved.

handle_secret now serializes with skip_secret_manager=True so the raw stored
reference survives transport. Secret resolution still happens at use-time via
the direct get_secret_value() call when building a connection, so ingestion
keeps resolving secret:<id> as before.

Fixes open-metadata/openmetadata-collate#4362

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: pmbrull

ingestion/src/metadata/ingestion/models/custom_pydantic.py

@@ -205,10 +205,15 @@ def handle_secret(value: Any, handler, info: SerializationInfo) -> str:
     Handle the secret value in the model.
     """
     if not (info.context is not None and info.context.get("mask_secrets", False)):
+        # Serialization must preserve the raw stored value. For external secret
+        # references (`secret:<id>`) this keeps the reference intact instead of
+        # resolving it, so the payload sent to the server keeps the reference and
+        # is not silently turned into a plain secret. Resolution against the
+        # secrets manager happens at use-time through a direct get_secret_value()
+        # call (e.g. when building a connection).
         if info.mode == "json":
-            # short circuit the json serialization and return the actual value
-            return value.get_secret_value()
-        return handler(value.get_secret_value())
+            return value.get_secret_value(skip_secret_manager=True)
+        return handler(value.get_secret_value(skip_secret_manager=True))
     return str(value)  # use pydantic's logic to mask the secret
 
 

ingestion/tests/unit/models/test_custom_pydantic.py

@@ -1,7 +1,10 @@
+import json
 import uuid
 from typing import List, Optional
 from unittest import TestCase
 
+import pytest
+
 from metadata.generated.schema.api.data.createDashboardDataModel import (
     CreateDashboardDataModelRequest,
 )
@@ -21,14 +24,30 @@
     TableConstraint,
     TableType,
 )
+from metadata.generated.schema.entity.services.connections.database.common.basicAuth import (
+    BasicAuth,
+)
+from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
+    MysqlConnection,
+)
+from metadata.generated.schema.security.secrets.secretsManagerClientLoader import (
+    SecretsManagerClientLoader,
+)
+from metadata.generated.schema.security.secrets.secretsManagerProvider import (
+    SecretsManagerProvider,
+)
 from metadata.generated.schema.type.basic import (
     EntityExtension,
     EntityName,
     FullyQualifiedEntityName,
     Markdown,
 )
 from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.ingestion.connections.builders import get_password_secret
 from metadata.ingestion.models.custom_pydantic import BaseModel, CustomSecretStr
+from metadata.utils.secrets.secrets_manager import SecretsManager
+from metadata.utils.secrets.secrets_manager_factory import SecretsManagerFactory
+from metadata.utils.singleton import Singleton
 
 
 class CustomPydanticValidationTest(TestCase):
@@ -837,6 +856,112 @@ class ComplexSecretModel(BaseModel):
         )
 
 
+class TestExternalSecretReferenceSerialization:
+    """Regression tests for https://github.com/open-metadata/openmetadata-collate/issues/4362.
+
+    Values prefixed with ``secret:`` are external secret references: the server
+    resolves them against an external secret manager instead of persisting the
+    raw value. The prefix MUST survive SDK serialization. If it is stripped, the
+    reference is sent as a plain secret and ends up stored in Collate's internal
+    secret manager rather than being resolved externally.
+
+    The SDK serializes create/update payloads with
+    ``model_dump_json(context={"mask_secrets": False})`` (see the PUT path in
+    ``metadata.ingestion.ometa.ometa_api``), so these tests exercise that exact
+    serialization with the default ``DBSecretsManager`` active, as it is once an
+    ometa client has been instantiated.
+    """
+
+    EXTERNAL_SECRET_REFERENCE = "secret:/external/path/to/db/password"
+
+    @pytest.fixture(autouse=True)
+    def db_secrets_manager(self):
+        Singleton.clear_all()
+        SecretsManagerFactory(SecretsManagerProvider.db, SecretsManagerClientLoader.noop)
+        yield
+        Singleton.clear_all()
+
+    def test_prefix_preserved_for_custom_secret_str(self):
+        class Connection(BaseModel):
+            password: CustomSecretStr
+
+        connection = Connection(password=self.EXTERNAL_SECRET_REFERENCE)
+
+        serialized = json.loads(connection.model_dump_json(context={"mask_secrets": False}))
+
+        assert serialized["password"] == self.EXTERNAL_SECRET_REFERENCE
+
+    def test_prefix_preserved_for_typed_service_connection(self):
+        connection = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password=self.EXTERNAL_SECRET_REFERENCE),
+            hostPort="localhost:3306",
+        )
+
+        serialized = json.loads(connection.model_dump_json(context={"mask_secrets": False}, by_alias=True))
+
+        assert serialized["authType"]["password"] == self.EXTERNAL_SECRET_REFERENCE
+
+
+RESOLVED_EXTERNAL_SECRET = "resolved-external-password"
+
+
+class _FakeExternalSecretsManager(SecretsManager):
+    """Test double for an external secrets manager (e.g. AWS, Azure).
+
+    Resolves any reference id to a fixed value, so a test can tell real
+    resolution apart from mere ``secret:`` prefix stripping.
+    """
+
+    def get_string_value(self, secret_id: str) -> str:
+        return RESOLVED_EXTERNAL_SECRET
+
+
+class TestExternalSecretReferenceResolution:
+    """Guards the connect-time resolution path against the #4362 fix.
+
+    Preserving the ``secret:`` reference during serialization must NOT change how
+    ingestion resolves that reference when building a connection. Resolution runs
+    through a direct ``get_secret_value()`` call (see
+    ``metadata.ingestion.connections.builders.get_password_secret``), not through
+    serialization, so it stays intact. An external secrets manager is active
+    here, as it would be on a hybrid ingestion runner.
+    """
+
+    EXTERNAL_SECRET_REFERENCE = "secret:/external/path/to/db/password"
+
+    @pytest.fixture(autouse=True)
+    def external_secrets_manager(self):
+        Singleton.clear_all()
+        factory = SecretsManagerFactory(SecretsManagerProvider.db, SecretsManagerClientLoader.noop)
+        factory.secrets_manager = _FakeExternalSecretsManager()
+        yield
+        Singleton.clear_all()
+
+    def test_reference_resolves_for_connection_use(self):
+        connection = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password=self.EXTERNAL_SECRET_REFERENCE),
+            hostPort="localhost:3306",
+        )
+
+        password = get_password_secret(connection)
+
+        assert password.get_secret_value() == RESOLVED_EXTERNAL_SECRET
+
+    def test_reference_is_not_resolved_into_serialized_payload(self):
+        connection = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password=self.EXTERNAL_SECRET_REFERENCE),
+            hostPort="localhost:3306",
+        )
+
+        serialized = json.loads(connection.model_dump_json(context={"mask_secrets": False}, by_alias=True))
+
+        assert serialized["authType"]["password"] == self.EXTERNAL_SECRET_REFERENCE
+        assert serialized["authType"]["password"] != RESOLVED_EXTERNAL_SECRET
+
+
 class DashboardDataModelTransformationTest(TestCase):
     """Test DashboardDataModel transformations with nested children and reserved keywords."""