morningman · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026
@@ -319,9 +319,19 @@ Rules for `of(map)`:
   condition *holds*, so the lambda states the failure case — the opposite of
   Guava's `checkArgument`. Reversing it compiles cleanly and inverts the check.
 - **Mark secrets `sensitive = true`** and implement `toString()` as
-  `ConnectorPropertiesUtils.toMaskedString(this)`. That covers logs only; the
-  `SHOW CREATE CATALOG` masking is a separate, hand-maintained list in fe-core
-  (`DatasourcePrintableMap.SENSITIVE_KEY`).
+  `ConnectorPropertiesUtils.toMaskedString(this)`. That covers logs only. For
+  `SHOW CREATE CATALOG`, the audit log and the other places that print catalog
+  properties, also declare the keys in `ConnectorProvider#securitySurface()`:
+  pass `ConnectorPropertiesUtils.getSensitiveKeys(<properties class>)` (or the
+  metastore dispatcher's `MetaStoreProviders.sensitivePropertyKeys()`) to
+  `withSensitiveKeys(...)`, so the same mark drives both. A namespace passed
+  through to a client unenumerated (Trino's `trino.*`, Fluss's lake options)
+  is declared with `withSensitiveKeyPrefix(prefix, markers...)`. The engine
+  adds both to its masking list (`DatasourcePrintableMap`) when it registers
+  the provider. Storage credentials (`s3.secret_key`, ...) are already on that
+  list. A path is not a secret: don't mark a keytab or certificate *path*
+  `sensitive`, since the list is matched by key name across every catalog,
+  resource and storage vault.
 - **Never pass the raw map into anything that can reach user-visible output.**
   It holds the credentials. `ConnectorTableSchema`'s table properties are the
   trap: `SHOW CREATE TABLE` renders them unmasked, and the only thing keeping

@@ -21,6 +21,8 @@
 import org.apache.doris.connector.spi.Connector;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
+import org.apache.doris.foundation.property.ConnectorPropertiesUtils;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -78,4 +80,19 @@ public void validatePropertiesForUpdate(
     public Connector create(Map<String, String> properties, ConnectorContext context) {
         return new AdbcConnector(properties, context);
     }
+
+    /**
+     * The sensitive properties (the password), and the driver options under
+     * {@link AdbcCatalogProperties#DRIVER_OPTION_PREFIX}: each driver names its own credentials (an auth
+     * token, an authorization header, a private key), so a driver option is a secret when its name carries
+     * a credential-like word. Like {@link #getType()}, this must not touch an ADBC class: the engine asks
+     * for it before it discards a duplicate directory copy of the plugin.
+     */
+    @Override
+    public ConnectorSecuritySurface securitySurface() {
+        return ConnectorSecuritySurface.EMPTY
+                .withSensitiveKeys(ConnectorPropertiesUtils.getSensitiveKeys(AdbcCatalogProperties.class))
+                .withSensitiveKeyPrefix(AdbcCatalogProperties.DRIVER_OPTION_PREFIX,
+                        "key", "secret", "password", "token", "credential", "authorization");
+    }
 }
@@ -75,6 +75,10 @@ void constructingTheProviderLoadsNoAdbcClass() throws Exception {
                 "getType() must be answerable without any ADBC class");
         Assertions.assertTrue(((ConnectorProvider) provider).isStandaloneCatalogType(),
                 "adbc is a catalog type a user writes in CREATE CATALOG");
+        // The engine reads the security surface when it registers the provider, before it can tell that a
+        // directory copy duplicates the built-in one, so it must be answerable without any ADBC class too.
+        Assertions.assertFalse(((ConnectorProvider) provider).securitySurface().sensitiveKeys().isEmpty(),
+                "securitySurface() must be answerable without any ADBC class");
 
         Assertions.assertEquals(List.of(), loader.blockedNames(),
                 "Provider discovery must not load any org.apache.arrow.adbc class");

@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.adbc;
+
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface.SensitiveKeyPrefix;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+/**
+ * Pins the ADBC connector's secrets: the password, and the driver options passed through under
+ * {@code adbc.} whose names carry a credential-like word, since each ADBC driver names its own credentials.
+ */
+public class AdbcConnectorProviderSecuritySurfaceTest {
+
+    @Test
+    public void declaresThePasswordAndTheCredentialLikeDriverOptions() {
+        ConnectorSecuritySurface surface = new AdbcConnectorProvider().securitySurface();
+        Assertions.assertEquals(Collections.singletonList("password"), surface.sensitiveKeys());
+        Assertions.assertEquals(1, surface.sensitiveKeyPrefixes().size());
+        SensitiveKeyPrefix driverOptions = surface.sensitiveKeyPrefixes().get(0);
+        // A driver option keeps its own adbc. name behind the pass-through prefix.
+        Assertions.assertTrue(driverOptions.covers("adbc.adbc.flight.sql.authorization_header"));
+        Assertions.assertTrue(driverOptions.covers("adbc.adbc.snowflake.sql.client_option.auth_token"));
+        Assertions.assertTrue(driverOptions.covers("adbc.adbc.snowflake.sql.client_option.jwt_private_key"));
+        Assertions.assertFalse(driverOptions.covers("adbc.adbc.snowflake.sql.db"));
+        Assertions.assertFalse(driverOptions.covers("uri"));
+        Assertions.assertTrue(surface.ssrfEndpointKeys().isEmpty());
+    }
+}
@@ -21,8 +21,10 @@
 import org.apache.doris.foundation.property.ConnectorProperty;
 import org.apache.doris.foundation.property.ParamRules;
 
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Everything a user writes in {@code CREATE CATALOG} for an Elasticsearch catalog, bound and checked.
@@ -172,6 +174,19 @@ public static EsCatalogProperties of(Map<String, String> properties) {
         return p;
     }
 
+    /**
+     * The names of the properties marked {@code sensitive}, each in both spellings a catalog may use: as
+     * declared and with {@link #ES_PROPERTIES_PREFIX}.
+     */
+    public static Set<String> sensitiveKeys() {
+        Set<String> keys = new HashSet<>();
+        for (String key : ConnectorPropertiesUtils.getSensitiveKeys(EsCatalogProperties.class)) {
+            keys.add(key);
+            keys.add(ES_PROPERTIES_PREFIX + key);
+        }
+        return keys;
+    }
+
     /**
      * Returns the properties with {@link #ES_PROPERTIES_PREFIX} removed from every key that carries it.
      *

@@ -20,6 +20,7 @@
 import org.apache.doris.connector.spi.Connector;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
 
 import java.util.Map;
 import java.util.Optional;
@@ -56,4 +57,10 @@ public void validateProperties(Map<String, String> properties) {
     public Connector create(Map<String, String> properties, ConnectorContext context) {
         return new EsConnector(properties, context);
     }
+
+    /** The engine masks the sensitive properties (the password) wherever it shows catalog properties. */
+    @Override
+    public ConnectorSecuritySurface securitySurface() {
+        return ConnectorSecuritySurface.EMPTY.withSensitiveKeys(EsCatalogProperties.sensitiveKeys());
+    }
 }
@@ -17,9 +17,13 @@
 
 package org.apache.doris.connector.es;
 
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.Optional;
 
 /** What this connector declares to the engine before any catalog of its type is initialized. */
@@ -41,4 +45,13 @@ public void anEsCatalogIsNotForceInitializedForEventSync() {
         // idle es catalog just to look for one.
         Assertions.assertFalse(new EsConnectorProvider().providesEventSource());
     }
+
+    @Test
+    public void thePasswordIsASecretInBothSpellings() {
+        // A catalog may write any key with the legacy elasticsearch. prefix, so the engine masks both.
+        ConnectorSecuritySurface surface = new EsConnectorProvider().securitySurface();
+        Assertions.assertEquals(new HashSet<>(Arrays.asList("password", "elasticsearch.password")),
+                new HashSet<>(surface.sensitiveKeys()));
+        Assertions.assertTrue(surface.sensitiveKeyPrefixes().isEmpty());
+    }
 }
@@ -20,6 +20,7 @@
 import org.apache.doris.connector.spi.Connector;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
 
 import java.util.Map;
 
@@ -48,4 +49,18 @@ public void validateProperties(Map<String, String> properties) {
     public Connector create(Map<String, String> properties, ConnectorContext context) {
         return new FlussConnector(FlussCatalogProperties.of(properties), context);
     }
+
+    /**
+     * The fluss client's SASL credentials, a small and stable set of keys, and the lake options under
+     * {@link FlussCatalogProperties#LAKE_OPTION_PREFIX}: those are open-ended (new paimon / filesystem
+     * credential names may arrive without a Doris release), so a lake option is a secret when its name
+     * carries a credential-like word, and endpoints and other operational settings stay visible.
+     */
+    @Override
+    public ConnectorSecuritySurface securitySurface() {
+        return ConnectorSecuritySurface.ofSensitiveKeys("fluss.client.security.sasl.username",
+                        "fluss.client.security.sasl.password", "fluss.client.security.sasl.jaas.config")
+                .withSensitiveKeyPrefix(FlussCatalogProperties.LAKE_OPTION_PREFIX,
+                        "key", "secret", "password", "token", "credential");
+    }
 }
@@ -21,11 +21,14 @@
 import org.apache.doris.connector.spi.ConnectorCapability;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface.SensitiveKeyPrefix;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.ServiceLoader;
@@ -111,4 +114,20 @@ public void declaresNestedColumnPruneButNotFieldIdAccessPath() throws IOExceptio
             connector.close();
         }
     }
+
+    @Test
+    public void declaresTheSaslCredentialsAndTheCredentialLikeLakeOptions() {
+        ConnectorSecuritySurface surface = new FlussConnectorProvider().securitySurface();
+        Assertions.assertEquals(Arrays.asList("fluss.client.security.sasl.username",
+                "fluss.client.security.sasl.password", "fluss.client.security.sasl.jaas.config"),
+                surface.sensitiveKeys());
+        Assertions.assertEquals(1, surface.sensitiveKeyPrefixes().size());
+        SensitiveKeyPrefix lake = surface.sensitiveKeyPrefixes().get(0);
+        Assertions.assertTrue(lake.covers("fluss.lake.paimon.s3.secret-key"));
+        Assertions.assertTrue(lake.covers("fluss.lake.paimon.rest.token"));
+        Assertions.assertTrue(lake.covers("fluss.lake.paimon.azure.oauth2_client_secret"));
+        Assertions.assertFalse(lake.covers("fluss.lake.paimon.s3.endpoint"));
+        Assertions.assertFalse(lake.covers("fluss.lake.paimon.warehouse"));
+        Assertions.assertFalse(lake.covers("fluss.bootstrap.servers"));
+    }
 }
@@ -17,6 +17,7 @@
 
 package org.apache.doris.connector.hive;
 
+import org.apache.doris.connector.metastore.spi.MetaStoreProviders;
 import org.apache.doris.connector.spi.Connector;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
@@ -86,8 +87,10 @@ public void validatePropertiesForUpdate(
     public ConnectorSecuritySurface securitySurface() {
         // Every key the HMS client may take its thrift endpoint from: hive.metastore.uris, its documented alias
         // uri, and Hive 3's metastore.thrift.uris, which the client reads first. The engine's SSRF gate rejects
-        // an intranet target there at CREATE / ALTER CATALOG.
+        // an intranet target there at CREATE / ALTER CATALOG. The secrets are the metastore's sensitive
+        // properties (none today: a keytab is named by its path).
         return ConnectorSecuritySurface.ofSsrfEndpoints(HiveCatalogProperties.HIVE_METASTORE_URIS,
-                HiveCatalogProperties.URI, HiveCatalogProperties.METASTORE_THRIFT_URIS);
+                HiveCatalogProperties.URI, HiveCatalogProperties.METASTORE_THRIFT_URIS)
+                .withSensitiveKeys(MetaStoreProviders.sensitivePropertyKeys());
     }
 }
@@ -42,4 +42,14 @@ public void declaresMetastoreUriAsSsrfEndpoint() {
         Assertions.assertTrue(surface.jdbcDriverUrlKeys().isEmpty());
         Assertions.assertNull(surface.jdbcDriverUrlFlavorKey());
     }
+
+    @Test
+    public void declaresNoKeytabPathAsASecret() {
+        // The secrets come from the metastore's sensitive properties; the keytabs are named by path, and
+        // masking hadoop.kerberos.keytab by name would also hide it in every storage vault and resource.
+        ConnectorSecuritySurface surface = new HiveConnectorProvider().securitySurface();
+        Assertions.assertFalse(surface.sensitiveKeys().contains("hive.metastore.client.keytab"));
+        Assertions.assertFalse(surface.sensitiveKeys().contains("hadoop.kerberos.keytab"));
+        Assertions.assertTrue(surface.sensitiveKeyPrefixes().isEmpty());
+    }
 }
@@ -17,6 +17,7 @@
 
 package org.apache.doris.connector.iceberg;
 
+import org.apache.doris.connector.metastore.spi.MetaStoreProviders;
 import org.apache.doris.connector.spi.Connector;
 import org.apache.doris.connector.spi.ConnectorContext;
 import org.apache.doris.connector.spi.ConnectorProvider;
@@ -90,6 +91,8 @@ public void validateCreateTable(Map<String, String> properties) {
     public ConnectorSecuritySurface securitySurface() {
         // iceberg.jdbc.driver_url is a jar the connector loads, but only live on a jdbc-flavored catalog
         // (iceberg.catalog.type = jdbc); on a REST/HMS/... catalog it is dead config the gate must not touch.
-        return ConnectorSecuritySurface.ofJdbcDriverUrls("iceberg.catalog.type", "iceberg.jdbc.driver_url");
+        // The secrets are those of every metastore flavor: the engine masks by key name alone.
+        return ConnectorSecuritySurface.ofJdbcDriverUrls("iceberg.catalog.type", "iceberg.jdbc.driver_url")
+                .withSensitiveKeys(MetaStoreProviders.sensitivePropertyKeys());
     }
 }
@@ -39,4 +39,19 @@ public void declaresJdbcDriverUrlGatedByFlavor() {
         Assertions.assertEquals("iceberg.catalog.type", surface.jdbcDriverUrlFlavorKey());
         Assertions.assertTrue(surface.ssrfEndpointKeys().isEmpty());
     }
+
+    @Test
+    public void declaresTheSecretsOfEveryMetastoreFlavor() {
+        // iceberg.jdbc.password was printed in clear by SHOW CREATE CATALOG before it was declared here:
+        // fe-core's own list does not know it.
+        ConnectorSecuritySurface surface = new IcebergConnectorProvider().securitySurface();
+        for (String key : new String[] {"iceberg.jdbc.password", "iceberg.rest.oauth2.token",
+                "iceberg.rest.oauth2.credential", "aws.glue.session-token", "dlf.secret_key"}) {
+            Assertions.assertTrue(surface.sensitiveKeys().contains(key), key + " missing from "
+                    + surface.sensitiveKeys());
+        }
+        Assertions.assertFalse(surface.sensitiveKeys().contains("hadoop.kerberos.keytab"),
+                "a keytab is a path, not a secret");
+        Assertions.assertFalse(surface.sensitiveKeys().contains("iceberg.jdbc.user"));
+    }
 }
@@ -23,8 +23,10 @@
 import org.apache.doris.foundation.security.JdbcDriverUrlSecurity;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.UnaryOperator;
 
 /**
@@ -304,6 +306,19 @@ private void checkBoolean(String key) {
         }
     }
 
+    /**
+     * The names of the properties marked {@code sensitive}, each in both spellings a catalog may use: as
+     * declared and with {@link #JDBC_PROPERTIES_PREFIX}.
+     */
+    public static Set<String> sensitiveKeys() {
+        Set<String> keys = new HashSet<>();
+        for (String key : ConnectorPropertiesUtils.getSensitiveKeys(JdbcCatalogProperties.class)) {
+            keys.add(key);
+            keys.add(JDBC_PROPERTIES_PREFIX + key);
+        }
+        return keys;
+    }
+
     /**
      * Returns the properties with {@link #JDBC_PROPERTIES_PREFIX} removed from every key that carries
      * it, leaving an unprefixed key already present untouched -- the short spelling wins, as it did in

@@ -54,6 +54,8 @@ public void validateProperties(Map<String, String> properties) {
     public ConnectorSecuritySurface securitySurface() {
         // driver_url (documented alias: jdbc.driver_url) is the jar the connector loads into the FE JVM;
         // the engine's operator gate applies jdbc_driver_secure_path / the mandatory rule on CREATE + ALTER.
-        return ConnectorSecuritySurface.ofJdbcDriverUrls(null, "driver_url", "jdbc.driver_url");
+        // The engine masks the sensitive properties (the password) wherever it shows catalog properties.
+        return ConnectorSecuritySurface.ofJdbcDriverUrls(null, "driver_url", "jdbc.driver_url")
+                .withSensitiveKeys(JdbcCatalogProperties.sensitiveKeys());
     }
 }
@@ -23,6 +23,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.HashSet;
 
 /**
  * Pins the JDBC connector's driver_url declaration: the engine's operator gate checks exactly these keys,
@@ -37,4 +38,13 @@ public void declaresDriverUrlKeysAlwaysLive() {
         Assertions.assertNull(surface.jdbcDriverUrlFlavorKey());
         Assertions.assertTrue(surface.ssrfEndpointKeys().isEmpty());
     }
+
+    @Test
+    public void declaresThePasswordInBothSpellings() {
+        // A catalog may write any key with the jdbc. prefix, so the engine has to mask both spellings.
+        ConnectorSecuritySurface surface = new JdbcConnectorProvider().securitySurface();
+        Assertions.assertEquals(new HashSet<>(Arrays.asList("password", "jdbc.password")),
+                new HashSet<>(surface.sensitiveKeys()));
+        Assertions.assertTrue(surface.sensitiveKeyPrefixes().isEmpty());
+    }
 }