Fix secret handling for map-valued connection properties

common/src/main/java/org/openmetadata/annotations/PasswordAnnotator.java

@@ -35,6 +35,37 @@ public void propertyField(
         && "password".equals(propertyNode.get("format").asText())) {
       field.annotate(PasswordField.class);
     }
+    if (isPasswordValuedMap(propertyNode)) {
+      annotateAdditionalPropertiesAccessor(field);
+    }
+  }
+
+  /**
+   * A property whose schema is an object with password-formatted {@code additionalProperties} holds
+   * a map of secrets (e.g. Kafka {@code consumerConfig}). jsonschema2pojo generates a holder class
+   * whose only member is the {@code additionalProperties} map; the {@code format} lives on the
+   * property schema, never on the generated member, so without this the map is invisible to the
+   * secret walkers.
+   * ADR:2026-10-30-password-format-on-additionalproperties-makes-a-map-of-secrets
+   */
+  private boolean isPasswordValuedMap(JsonNode propertyNode) {
+    JsonNode additionalProperties = propertyNode.get("additionalProperties");
+    return additionalProperties != null
+        && additionalProperties.isObject()
+        && additionalProperties.get("format") != null
+        && "password".equals(additionalProperties.get("format").asText());
+  }
+
+  private void annotateAdditionalPropertiesAccessor(JFieldVar field) {
+    if (field.type() instanceof JDefinedClass mapHolderClass) {
+      JFieldVar additionalProperties = mapHolderClass.fields().get("additionalProperties");
+      if (additionalProperties != null) {
+        additionalProperties.annotate(PasswordField.class);
+      }
+      mapHolderClass.methods().stream()
+          .filter(method -> "getAdditionalProperties".equals(method.name()))
+          .forEach(method -> method.annotate(PasswordField.class));
+    }
   }
 
   /** Add {@link PasswordField} annotation to getter methods */

docs/decisions/2026-10-24-a-masked-connection-secret-is-kept-only-while-the-endpoint-is-unchanged.md

@@ -1,7 +1,7 @@
 # A masked connection secret is restored only while the connection still points at the same endpoint
 
 - **Status:** Accepted
-- **Revisions:** v1 2026-10-24 (initial); v2 2026-10-24 (fail-closed: exempt known-safe fields rather than enumerate endpoint fields); v3 2026-10-30 (ingestion pipeline sources); v4 2026-10-30 (a storage manifest no schema matches)
+- **Revisions:** v1 2026-10-24 (initial); v2 2026-10-24 (fail-closed: exempt known-safe fields rather than enumerate endpoint fields); v3 2026-10-30 (ingestion pipeline sources); v4 2026-10-30 (a storage manifest no schema matches); v5 2026-11-03 (typed-over password-map entries count as retargets)
 - **Deciders:** backend security
 - **Guard:** `ConnectionSecretsGuardTest`; `IngestionPipelineBuilderTest`; `IngestionPipelineResourceIT.test_dbtTokenMustBeReenteredWhenTheSourceMoves`, `IngestionPipelineResourceIT.test_storageManifestMatchingNoSourceSchemaIsRefused`
 - **Related:** open-metadata/openmetadata-collate#7797, open-metadata/OpenMetadata#36016
@@ -60,3 +60,25 @@ manifest source that matches none of the source schemas fails closed instead of
   returned and decrypted, and the rest of the pipeline stays readable so it can be repaired.
 - **Re-stored by a secrets-manager migration**, which reads each pipeline back as above before
   encrypting it again: such a source is dropped and has to be entered again.
+
+## Amendment (v5, 2026-11-03)
+
+Once `additionalProperties: {format: password}` made every string entry of a password-valued map
+mask to the same placeholder (ADR:2026-10-30-password-format-on-additionalproperties-makes-a-map-of-secrets),
+the masked comparison could no longer see an edit to a non-secret map entry — an OAuth token
+endpoint URL changed next to its masked client secret. The guard now also treats a typed
+(plaintext) value at a masked password-map entry as a change, unless it still equals the stored
+entry. The same holds for any value other than the mask: an entry is kept only when it arrives as
+the stored value itself or decrypts locally to the stored cleartext, so a different `secret:`
+reference, which the server never fetches, or another Fernet token counts as a change. The check is scoped to password-map entries:
+a typed value at a scalar password field is a re-entered secret, which this record has always
+accepted. The check covers both connections and the ingestion pipeline sources of v3, where a dbt
+HTTP source's `dbtHttpHeaders` is such a map.
+
+The cost is deliberate: rotating one secret *inside* the map (a new
+`sasl.oauthbearer.client.secret`) while the connection's other secrets stay masked reads exactly
+like a redirect — both are typed values at masked map entries — and is refused, so the caller
+must re-enter every secret of the connection. That is the same rule the unmasked maps produced
+before they were declared secret, and the only sound distinction (marking which keys are secrets)
+is the per-key declaration ADR:2026-10-30-password-format-on-additionalproperties-makes-a-map-of-secrets
+leaves open. Pinned by `ConnectionSecretsGuardTest.rotatingOneSecretInsideTheConfigMapRequiresReenteringTheOthers`.

docs/decisions/2026-10-30-password-format-on-additionalproperties-makes-a-map-of-secrets.md

@@ -0,0 +1,55 @@
+# A schema property whose additionalProperties declare format: password is a map of secrets
+
+- **Status:** Accepted
+- **Revisions:** v1 2026-10-30 (initial)
+- **Deciders:** backend security
+- **Guard:** `PasswordEntityMaskerTest`, `SecretsManagerLifecycleTest`, `KafkaConnectionMapSecretsTest`
+- **Related:** ADR:2026-10-24-a-masked-connection-secret-is-kept-only-while-the-endpoint-is-unchanged, #33477 (secrets inside arrays)
+
+## Context
+
+Map-valued connection properties — `consumerConfig` and `schemaRegistryConfig` on Kafka and
+Redpanda, `dbtHttpHeaders` on a dbt HTTP pipeline, `headers` on a CustomLLM service — are the only
+place for credentials such as `sasl.oauthbearer.client.secret`, `ssl.key.password` or a GitLab
+`PRIVATE-TOKEN`. The secret framework had no notion of a map of secrets: `PasswordAnnotator` only
+marked properties that are themselves `format: password`, and the walkers in `SecretsManager` and
+`PasswordEntityMasker` had no branch for `java.util.Map`, so map entries were stored unencrypted,
+never reached an external secrets manager, and were returned in clear to any authenticated user
+(ANT-2026-INNF1N4V). #33477 closed the same gap for secrets inside arrays but not inside maps.
+
+## Decision
+
+- A JSON-schema property whose `additionalProperties` declare `format: password` is a **map of
+  secrets**. `PasswordAnnotator` puts `@PasswordField` on the generated holder class's
+  `getAdditionalProperties()`, and every walker handles the map entry by entry.
+- Every **string** entry is masked on read, encrypted at rest, and — with an external secrets
+  manager — stored as its own secret. Non-string entries (`session.timeout.ms: 6000`) keep their
+  type and are neither encrypted nor masked: the maps also carry numbers and booleans that are not
+  secrets.
+- Each entry's secret id is `<entity secret path>/<field>/additionalproperties/<entry key>`,
+  lowercased and cleaned by the provider's id rules (dots become `_` under the default
+  configuration), and mask/unmask keys include the entry key, so a `*********` sent back by a
+  client is restored for that exact entry and `deleteSecrets` derives the same id.
+- Ingestion clients resolve `secret:` references inside such maps themselves; the server never
+  fetches secrets. String-valued maps generate as `dict[str, CustomSecretStr]` and unwrap through
+  `get_secret_value()` (dbt); untyped maps stay `dict[str, Any]` and the connector resolves
+  references before use (`messaging/kafka/connection.py`).
+- The mask is a **blanket**: harmless settings in the map (`security.protocol`, `group.id`) show
+  as `*********` too, because JSON Schema cannot mark individual keys of a free-form map. This is
+  the accepted cost of declaring the map at all; the alternative (typed fields for every possible
+  key) does not exist for maps like `dbtHttpHeaders`.
+
+## Consequences
+
+- `connectionOptions`/`connectionArguments`, Trino `proxies` and NATS `additionalConfig` are
+  deliberately **not** declared password maps: doing so would mask every option they carry, not
+  only credentials. Declaring them needs a per-case decision.
+- `ConnectionSecretsGuard` compares masked configs, and a masked map hides edits to non-secret
+  entries (a token endpoint URL) — so the guard treats a typed value at a masked map entry as a
+  retarget; see ADR:2026-10-24-a-masked-connection-secret-is-kept-only-while-the-endpoint-is-unchanged.
+  Because a rotated map secret looks exactly like a retargeted setting, rotating one entry
+  requires re-entering every secret of the connection.
+- Rows saved before this record keep their map values in plaintext until the service or pipeline
+  is saved again; credentials already stored this way should be rotated.
+- Revisit if a per-key secret declaration (typed fields for known keys, refused in the map) is
+  wanted for the Kafka maps, where masking harmless settings is the visible cost.

ingestion/src/metadata/ingestion/source/database/dbt/dbt_config.py

@@ -120,14 +120,31 @@ def _(config: DbtLocalConfig):
         raise DBTConfigException(f"Error fetching dbt files from local: {exc}")  # noqa: B904
 
 
+def _request_headers(config: DbtHttpConfig) -> dict:
+    """dbtHttpHeaders values are CustomSecretStr (format: password in the schema). With an
+    external secrets manager the server hands out `secret:` references, resolved here; under
+    the default DB provider a header starting with `secret:` is the literal value and is kept."""
+    if not config.dbtHttpHeaders:
+        return {}
+    from metadata.utils.secrets.db_secrets_manager import (  # pylint: disable=import-outside-toplevel
+        DBSecretsManager,
+    )
+    from metadata.utils.secrets.secrets_manager_factory import (  # pylint: disable=import-outside-toplevel,cyclic-import
+        SecretsManagerFactory,
+    )
+
+    literal = isinstance(SecretsManagerFactory().get_secrets_manager(), DBSecretsManager)
+    return {key: value.get_secret_value(skip_secret_manager=literal) for key, value in config.dbtHttpHeaders.items()}
+
+
 @get_dbt_details.register
 def _(config: DbtHttpConfig):  # noqa: C901
     try:
         verify_ssl_fn = get_verify_ssl_fn(config.dbtVerifySSL) if config.dbtVerifySSL else lambda _: None
         ssl_verify = verify_ssl_fn(config.dbtSSLConfig) if config.dbtVerifySSL else None
         if ssl_verify is None:
             ssl_verify = True
-        request_headers = dict(config.dbtHttpHeaders) if config.dbtHttpHeaders else {}
+        request_headers = _request_headers(config)
 
         manifest_url = config.dbtManifestHttpPath
         logger.debug(f"Requesting [dbtManifestHttpPath] to: {manifest_url}")

ingestion/src/metadata/ingestion/source/messaging/kafka/connection.py

@@ -33,6 +33,7 @@
 )
 from metadata.ingestion.connections.connection import BaseConnection
 from metadata.ingestion.connections.test_connections import test_connection_steps
+from metadata.ingestion.models.custom_pydantic import SECRET
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.utils.constants import THREE_MIN
 from metadata.utils.logger import ingestion_logger
@@ -59,6 +60,47 @@ class SchemaRegistryException(Exception):  # noqa: N818
 CONSUMER_ONLY_CONFIG_KEYS = ("group.id", "enable.auto.commit", "auto.offset.reset")
 
 
+def _resolve_secret_value(value: str) -> str:
+    """Resolve a `secret:` reference against the configured secrets manager.
+
+    Imported lazily, as in custom_pydantic, to avoid a circular import.
+    """
+    from metadata.utils.secrets.db_secrets_manager import (  # pylint: disable=import-outside-toplevel
+        DBSecretsManager,
+    )
+    from metadata.utils.secrets.secrets_manager_factory import (  # pylint: disable=import-outside-toplevel,cyclic-import
+        SecretsManagerFactory,
+    )
+
+    secrets_manager = SecretsManagerFactory().get_secrets_manager()
+    # The server only hands out references with an external store; under the default DB
+    # provider a value starting with `secret:` is the literal setting and must reach the client.
+    if not secrets_manager or isinstance(secrets_manager, DBSecretsManager):
+        return value
+    secret_id = value.replace(SECRET, "", 1)
+    try:
+        resolved = secrets_manager.get_string_value(secret_id)
+    except Exception as exc:  # pylint: disable=broad-except
+        logger.error(
+            "Secret value [%s] not present in the configured secrets manager: %s",
+            secret_id,
+            exc,
+        )
+        return value
+    return resolved if resolved is not None else value
+
+
+def _resolve_secret_values(config: dict) -> dict:
+    """consumerConfig/schemaRegistryConfig entries are encrypted per entry on the server.
+    With an external secrets manager the API hands the client `secret:` references, which
+    typed password fields resolve through CustomSecretStr but a free-form map cannot, so
+    resolve them here before the dict reaches librdkafka."""
+    return {
+        key: _resolve_secret_value(value) if isinstance(value, str) and value.startswith(SECRET) else value
+        for key, value in config.items()
+    }
+
+
 class KafkaClient:
     def __init__(self, admin_client, schema_registry_client, consumer_factory) -> None:
         self.admin_client = admin_client
@@ -84,8 +126,8 @@ def get_connection(connection: KafkaConnectionConfig | RedpandaConnection) -> Ka
     """
     Create connection
     """
-    consumer_config = deepcopy(connection.consumerConfig) or {}
-    schema_registry_config = deepcopy(connection.schemaRegistryConfig) or {}
+    consumer_config = _resolve_secret_values(deepcopy(connection.consumerConfig) or {})
+    schema_registry_config = _resolve_secret_values(deepcopy(connection.schemaRegistryConfig) or {})
 
     if connection.saslUsername or connection.saslPassword or connection.saslMechanism:
         if connection.saslUsername:

ingestion/tests/unit/source/messaging/kafka/test_connection.py

@@ -108,3 +108,70 @@ def test_close_consumer_releases_a_built_consumer():
         client.close_consumer()
 
     mock_consumer.return_value.close.assert_called_once()
+
+
+def test_secret_references_in_config_maps_are_resolved():
+    """Map entries arrive as `secret:` references when an external secrets manager is
+    configured; they must be resolved before the dict reaches librdkafka."""
+    from metadata.ingestion.source.messaging.kafka.connection import get_connection
+
+    config = _kafka_config(
+        consumerConfig={
+            "sasl.oauthbearer.client.secret": "secret:/om/messaging/prod_kafka/consumerconfig/additionalproperties/sasl_oauthbearer_client_secret",
+            "session.timeout.ms": 6000,
+        },
+        schemaRegistryConfig={
+            "basic.auth.user.info": "secret:/om/messaging/prod_kafka/schemaregistryconfig/additionalproperties/basic_auth_user_info",
+        },
+        schemaRegistryURL="https://registry.example.com",
+    )
+    secrets_manager = MagicMock()
+    secrets_manager.get_string_value.side_effect = lambda secret_id: f"resolved:{secret_id}"
+
+    with (
+        patch(f"{CONNECTION_MODULE}.AdminClient") as mock_admin,
+        patch(f"{CONNECTION_MODULE}.SchemaRegistryClient") as mock_registry,
+        patch("metadata.utils.secrets.secrets_manager_factory.SecretsManagerFactory") as mock_factory,
+    ):
+        mock_factory.return_value.get_secrets_manager.return_value = secrets_manager
+        get_connection(config)
+
+    admin_config = mock_admin.call_args.args[0]
+    assert admin_config["sasl.oauthbearer.client.secret"].startswith("resolved:/om/messaging")
+    assert admin_config["session.timeout.ms"] == 6000
+    registry_config = mock_registry.call_args.args[0]
+    assert registry_config["basic.auth.user.info"].startswith("resolved:/om/messaging")
+
+
+def test_config_maps_pass_through_when_no_secrets_manager_is_configured():
+    from metadata.ingestion.source.messaging.kafka.connection import get_connection
+
+    config = _kafka_config(consumerConfig={"ssl.key.password": "secret:/om/messaging/prod/key"})
+
+    with (
+        patch(f"{CONNECTION_MODULE}.AdminClient") as mock_admin,
+        patch("metadata.utils.secrets.secrets_manager_factory.SecretsManagerFactory") as mock_factory,
+    ):
+        mock_factory.return_value.get_secrets_manager.return_value = None
+        get_connection(config)
+
+    admin_config = mock_admin.call_args.args[0]
+    assert admin_config["ssl.key.password"] == "secret:/om/messaging/prod/key"
+
+
+def test_literal_secret_prefixed_values_pass_through_under_the_default_db_provider():
+    """The default provider never hands out references, so a literal `secret:` value is kept."""
+    from metadata.ingestion.source.messaging.kafka.connection import get_connection
+    from metadata.utils.secrets.db_secrets_manager import DBSecretsManager
+
+    config = _kafka_config(consumerConfig={"ssl.key.password": "secret:passphrase"})
+
+    with (
+        patch(f"{CONNECTION_MODULE}.AdminClient") as mock_admin,
+        patch("metadata.utils.secrets.secrets_manager_factory.SecretsManagerFactory") as mock_factory,
+    ):
+        mock_factory.return_value.get_secrets_manager.return_value = DBSecretsManager()
+        get_connection(config)
+
+    admin_config = mock_admin.call_args.args[0]
+    assert admin_config["ssl.key.password"] == "secret:passphrase"

ingestion/tests/unit/test_dbt_http_config.py

@@ -126,6 +126,51 @@ def test_headers_forwarded_to_all_optional_fetch_calls(self):
             _, kwargs = actual_call
             assert kwargs["headers"] == {"Authorization": "Bearer tok"}
 
+    def test_secret_reference_header_is_resolved_via_secrets_manager(self):
+        """Header values arrive as `secret:` references when an external secrets manager
+        is configured; they must be resolved before the request is made."""
+        config = _base_config(
+            dbtHttpHeaders={
+                "PRIVATE-TOKEN": "secret:/om/pipeline/dbt-http/dbthttpheaders/additionalproperties/private_token"
+            }
+        )
+        manifest_resp = _make_json_response(MANIFEST_JSON)
+        secrets_manager = MagicMock()
+        secrets_manager.get_string_value.return_value = "glpat-resolved"
+
+        with (
+            patch(
+                "metadata.ingestion.source.database.dbt.dbt_config.requests.get",
+                return_value=manifest_resp,
+            ) as mock_get,
+            patch("metadata.utils.secrets.secrets_manager_factory.SecretsManagerFactory") as mock_factory,
+        ):
+            mock_factory.return_value.get_secrets_manager.return_value = secrets_manager
+            list(get_dbt_details(config))
+
+        _, kwargs = mock_get.call_args
+        assert kwargs["headers"] == {"PRIVATE-TOKEN": "glpat-resolved"}
+
+    def test_literal_secret_prefixed_header_is_kept_under_the_default_db_provider(self):
+        """The default provider never hands out references, so a literal `secret:` header is kept."""
+        from metadata.utils.secrets.db_secrets_manager import DBSecretsManager
+
+        config = _base_config(dbtHttpHeaders={"Authorization": "secret:literal-token"})
+        manifest_resp = _make_json_response(MANIFEST_JSON)
+
+        with (
+            patch(
+                "metadata.ingestion.source.database.dbt.dbt_config.requests.get",
+                return_value=manifest_resp,
+            ) as mock_get,
+            patch("metadata.utils.secrets.secrets_manager_factory.SecretsManagerFactory") as mock_factory,
+        ):
+            mock_factory.return_value.get_secrets_manager.return_value = DBSecretsManager()
+            list(get_dbt_details(config))
+
+        _, kwargs = mock_get.call_args
+        assert kwargs["headers"] == {"Authorization": "secret:literal-token"}
+
 
 class TestDbtHttpConfigSSLVerify:
     """verifySSL enum values produce correct verify= argument to requests.get()."""