keys) {
}
/**
- * Whether a property key must be masked in any user-visible catalog rendering.
- *
- * Most connectors register exact aliases. Fluss lake options are deliberately open-ended and
- * namespaced, however: new paimon/filesystem credential names may arrive without a Doris release.
- * Match credential-like tails only inside that namespace so endpoints and other operational settings
- * remain visible. Fluss client SASL credentials use a small, stable set of exact keys.
+ * Registers a namespace of secret keys that a connector provider declared with
+ * {@link ConnectorSecuritySurface#withSensitiveKeyPrefix}. {@code ConnectorPluginManager} registers them
+ * with the provider at FE startup, as it does the provider's exact keys through
+ * {@link #registerSensitiveKeys}.
+ */
+ public static void registerSensitiveKeyPrefix(SensitiveKeyPrefix keyPrefix) {
+ SENSITIVE_KEY_PREFIXES.add(keyPrefix);
+ }
+
+ /**
+ * Whether a property key must be masked in any user-visible catalog rendering: it is on the list, or it
+ * falls in a namespace of secret keys. The list and the namespaces are the engine's own plus those of
+ * every connector and filesystem provider registered at FE startup.
*/
public static boolean isSensitiveKey(Object key) {
if (!(key instanceof String)) {
@@ -175,19 +203,12 @@ public static boolean isSensitiveKey(Object key) {
if (SENSITIVE_KEY.contains(stringKey)) {
return true;
}
- String lower = stringKey.toLowerCase(Locale.ROOT);
- if (lower.equals("fluss.client.security.sasl.username")
- || lower.equals("fluss.client.security.sasl.password")
- || lower.equals("fluss.client.security.sasl.jaas.config")) {
- return true;
- }
- String lakePrefix = "fluss.lake.paimon.";
- if (!lower.startsWith(lakePrefix)) {
- return false;
+ for (SensitiveKeyPrefix keyPrefix : SENSITIVE_KEY_PREFIXES) {
+ if (keyPrefix.covers(stringKey)) {
+ return true;
+ }
}
- String tail = lower.substring(lakePrefix.length());
- return tail.contains("key") || tail.contains("secret") || tail.contains("password")
- || tail.contains("token") || tail.contains("credential");
+ return false;
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java
index e660a0425855c8..1d536cf9d69f4a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java
@@ -17,10 +17,13 @@
package org.apache.doris.connector;
+import org.apache.doris.common.util.DatasourcePrintableMap;
import org.apache.doris.connector.spi.Connector;
import org.apache.doris.connector.spi.ConnectorConfFile;
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.apache.doris.datasource.CatalogFactory;
import org.apache.doris.extension.loader.ApiVersionGate;
import org.apache.doris.extension.loader.ClassLoadingPolicy;
@@ -188,19 +191,22 @@ void loadBuiltins(ClassLoader classLoader) {
providerClass.getName(), e);
continue;
}
+ ConnectorSecuritySurface securitySurface;
try {
- // Snapshot self-reported metadata before publishing the provider
- // so one throwing implementation is rejected cleanly instead of
- // aborting startup or being active without an inventory row.
+ // Snapshot self-reported metadata (the security surface included) before publishing the
+ // provider so one throwing implementation is rejected cleanly instead of
+ // aborting startup or being active without an inventory row. A surface built through the
+ // provider's own ServiceLoader fails with ServiceConfigurationError, as construction can.
+ securitySurface = provider.securitySurface();
PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, provider);
- } catch (RuntimeException e) {
+ } catch (ServiceConfigurationError | RuntimeException e) {
LOG.warn("Skip built-in connector provider {}: self-reported metadata failed",
providerClass.getName(), e);
continue;
}
// Deliberately outside that catch: registerDiscovered's fail-loud
// IllegalStateException reports a build error, not a "skip this one" condition.
- if (registerDiscovered(provider, true)) {
+ if (registerDiscovered(provider, securitySurface, true)) {
LOG.info("Registered built-in connector provider: {}", provider.getType());
}
}
@@ -212,15 +218,19 @@ void loadBuiltins(ClassLoader classLoader) {
* {@link ConnectorProvider#supports} matches case-insensitively and catalog types are lower-cased before
* routing). Refusing a reserved name here is what makes it impossible for a plugin to shadow a catalog type
* the engine implements itself — routing order then cannot matter. The engine names it claims for
- * {@code CREATE TABLE ... ENGINE=} go through the same two checks.
+ * {@code CREATE TABLE ... ENGINE=} go through the same two checks. An admitted provider's declared secrets
+ * (the keys and key namespaces on its {@code securitySurface()}) join the engine's masking list.
*
+ * @param securitySurface the provider's {@code securitySurface()}, which the caller reads before it
+ * publishes anything: it is plugin code that may throw
* @param failFast {@code true} for the classpath batch: two providers claiming one type name there is a
* build error and must fail loud. {@code false} for the plugin-directory batch: that is a
* deployment accident, so the offender is skipped and logged, preserving
* {@link #loadPlugins}'s partial-success contract (one bad plugin dir must not stop FE).
* @return true if the provider was admitted
*/
- boolean registerDiscovered(ConnectorProvider provider, boolean failFast) {
+ boolean registerDiscovered(ConnectorProvider provider, ConnectorSecuritySurface securitySurface,
+ boolean failFast) {
String type = provider.getType();
Set engineNames = provider.acceptedCreateTableEngineNames();
String problem = typeNameProblem(type);
@@ -242,10 +252,19 @@ boolean registerDiscovered(ConnectorProvider provider, boolean failFast) {
for (String engineName : engineNames) {
claimedEngineNames.add(engineName.toLowerCase());
}
+ registerSecrets(securitySurface);
providers.add(provider);
return true;
}
+ /** Adds the secret keys and key namespaces a provider declares to the engine's masking list. */
+ private static void registerSecrets(ConnectorSecuritySurface securitySurface) {
+ DatasourcePrintableMap.registerSensitiveKeys(securitySurface.sensitiveKeys());
+ for (SensitiveKeyPrefix keyPrefix : securitySurface.sensitiveKeyPrefixes()) {
+ DatasourcePrintableMap.registerSensitiveKeyPrefix(keyPrefix);
+ }
+ }
+
private static String typeNameProblem(String type) {
if (type == null || type.trim().isEmpty()) {
return "getType() returned a blank type name";
@@ -314,9 +333,23 @@ public void loadPlugins(List pluginRoots) {
runtimeManager.discard(handle.getPluginName());
continue;
}
+ ConnectorProvider provider = handle.getFactory();
+ // Snapshot the security surface before publishing anything: it is plugin code, and one bad plugin
+ // dir must not stop FE. A connector may build it through its own ServiceLoader, as Hive, Iceberg and
+ // Paimon do for their metastore backends, and a descriptor naming a backend that cannot be loaded
+ // fails that with ServiceConfigurationError, an Error.
+ ConnectorSecuritySurface securitySurface;
+ try {
+ securitySurface = provider.securitySurface();
+ } catch (RuntimeException | LinkageError | ServiceConfigurationError e) {
+ runtimeManager.discard(handle.getPluginName());
+ LOG.warn("Skip connector plugin '{}' from {}: securitySurface() failed",
+ handle.getPluginName(), handle.getPluginDir(), e);
+ continue;
+ }
// The inventory row is written only for a provider that was actually admitted, so
// information_schema.extensions never lists a connector the routing table cannot reach.
- if (registerDiscovered(handle.getFactory(), false)) {
+ if (registerDiscovered(provider, securitySurface, false)) {
PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle);
loadConnectorConfig(handle);
LOG.info("Loaded connector plugin: name={}, pluginDir={}, jarCount={}",
@@ -548,6 +581,7 @@ public void validateCreateTable(String catalogType, Map properti
*/
public void registerProvider(ConnectorProvider provider) {
providers.add(0, provider);
+ registerSecrets(provider.securitySurface());
}
private List providerNames() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilderForEncryption.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilderForEncryption.java
index c1552b58f83afc..a3b4227fc39780 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilderForEncryption.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilderForEncryption.java
@@ -115,6 +115,15 @@ public LogicalPlan visitCreateCatalog(DorisParser.CreateCatalogContext ctx) {
return super.visitCreateCatalog(ctx);
}
+ // alter catalog clause (ALTER CATALOG ... SET PROPERTIES(...))
+ @Override
+ public LogicalPlan visitAlterCatalogProperties(DorisParser.AlterCatalogPropertiesContext ctx) {
+ DorisParser.PropertyItemListContext properties = ctx.propertyItemList();
+ encryptProperty(visitPropertyItemList(properties), properties.start.getStartIndex(),
+ properties.stop.getStopIndex());
+ return super.visitAlterCatalogProperties(ctx);
+ }
+
// create repository clause (CREATE [READ ONLY] REPOSITORY ... WITH ... PROPERTIES(...))
@Override
public LogicalPlan visitCreateRepository(DorisParser.CreateRepositoryContext ctx) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterCatalogPropertiesCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterCatalogPropertiesCommand.java
index 042a552f2d127f..0229b690f2d62a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterCatalogPropertiesCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterCatalogPropertiesCommand.java
@@ -30,7 +30,7 @@
/**
* Represents the command for ALTER CATALOG ... SET PROPERTIES.
*/
-public class AlterCatalogPropertiesCommand extends AlterCatalogCommand {
+public class AlterCatalogPropertiesCommand extends AlterCatalogCommand implements NeedAuditEncryption {
private final Map newProperties;
public AlterCatalogPropertiesCommand(String catalogName, Map properties) {
@@ -50,4 +50,9 @@ public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
public R accept(PlanVisitor visitor, C context) {
return visitor.visitAlterCatalogPropertiesCommand(this, context);
}
+
+ @Override
+ public boolean needAuditEncryption() {
+ return true;
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/util/DatasourcePrintableMapTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/util/DatasourcePrintableMapTest.java
index 304b7b18573122..0960cbdea77cb5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/common/util/DatasourcePrintableMapTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/common/util/DatasourcePrintableMapTest.java
@@ -17,6 +17,8 @@
package org.apache.doris.common.util;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -226,6 +228,7 @@ public void testFlussNamespacedCredentialsAreMaskedWithoutEnumeratingProviders()
properties.put("fluss.lake.paimon.oss.secret-key", "oss-secret");
properties.put("fluss.lake.paimon.azure.oauth2_client_secret", "azure-secret");
properties.put("fluss.lake.paimon.rest.token", "rest-token");
+ properties.put("FLUSS.LAKE.PAIMON.GCS.Secret-Key", "upper-case-secret");
properties.put("fluss.lake.paimon.s3.endpoint", "http://minio:9000");
String result = new DatasourcePrintableMap<>(properties, "=", false, false, true).toString();
@@ -236,6 +239,7 @@ public void testFlussNamespacedCredentialsAreMaskedWithoutEnumeratingProviders()
Assertions.assertFalse(result.contains("oss-secret"), result);
Assertions.assertFalse(result.contains("azure-secret"), result);
Assertions.assertFalse(result.contains("rest-token"), result);
+ Assertions.assertFalse(result.contains("upper-case-secret"), result);
Assertions.assertTrue(result.contains("fluss.lake.paimon.s3.endpoint = http://minio:9000"), result);
}
@@ -333,4 +337,29 @@ public void testRegisterSensitiveKeysMasksNewlyRegisteredKey() {
Assertions.assertTrue(result.contains(
"PROVIDER_CONTRIBUTED_SECRET_KEY = " + DatasourcePrintableMap.PASSWORD_MASK));
}
+
+ @Test
+ public void testRegisterSensitiveKeyPrefixMasksTheKeysItCovers() {
+ // Simulates connector providers declaring pass-through namespaces at startup; the list is
+ // process-wide, hence prefixes no other test uses.
+ ConnectorSecuritySurface surface = ConnectorSecuritySurface
+ .ofSensitiveKeyPrefix("prefix_test_passthrough.", "Password", "token")
+ .withSensitiveKeyPrefix("prefix_test_all.");
+ surface.sensitiveKeyPrefixes().forEach(DatasourcePrintableMap::registerSensitiveKeyPrefix);
+
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey("prefix_test_passthrough.connection-password"));
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey("PREFIX_TEST_PASSTHROUGH.auth.TOKEN"));
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey("prefix_test_passthrough.connection-url"));
+ // Only the rest after the prefix is searched, and only under the prefix.
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey("other.prefix_test_passthrough.password"));
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey("prefix_test_all.endpoint"));
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey("prefix_test_al.endpoint"));
+
+ Map testMap = new HashMap<>();
+ testMap.put("prefix_test_passthrough.connection-password", "super_secret_value");
+ testMap.put("prefix_test_passthrough.connection-url", "jdbc:x://host");
+ String result = new DatasourcePrintableMap<>(testMap, "=", false, false, true).toString();
+ Assertions.assertFalse(result.contains("super_secret_value"), result);
+ Assertions.assertTrue(result.contains("prefix_test_passthrough.connection-url = jdbc:x://host"), result);
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorPluginManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorPluginManagerTest.java
index 643b7bbeeeee09..f7883fd2542644 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorPluginManagerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorPluginManagerTest.java
@@ -17,12 +17,18 @@
package org.apache.doris.connector;
+import org.apache.doris.common.util.DatasourcePrintableMap;
import org.apache.doris.connector.spi.Connector;
import org.apache.doris.connector.spi.ConnectorContext;
import org.apache.doris.connector.spi.ConnectorMetadata;
import org.apache.doris.connector.spi.ConnectorProvider;
+import org.apache.doris.connector.spi.ConnectorSecuritySurface;
import org.apache.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connectorconf.testplugins.BackendDiscoveringConnectorProvider;
import org.apache.doris.datasource.CatalogFactory;
+import org.apache.doris.extension.loader.PluginRegistry;
+import org.apache.doris.pluginapiversion.testplugins.VersionProbeConnectorProvider;
+import org.apache.doris.utframe.PluginJarWriter;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -39,6 +45,7 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -174,10 +181,10 @@ void duplicateTypeNameOnClasspathFailsLoud() {
// Type names are the identity CREATE CATALOG routes on and the anchor of source-prefixed namespaces.
// Two providers claiming one name on the classpath is a build error, and the winner would be decided by
// ServiceLoader order — silently, differently per build.
- Assertions.assertTrue(manager.registerDiscovered(createProvider("dup", true, "first"), true));
+ Assertions.assertTrue(admit(createProvider("dup", true, "first"), true));
IllegalStateException e = Assertions.assertThrows(IllegalStateException.class,
- () -> manager.registerDiscovered(createProvider("DUP", true, "second"), true),
+ () -> admit(createProvider("DUP", true, "second"), true),
"a classpath duplicate must fail loud, and case must not be a way around it");
Assertions.assertTrue(e.getMessage().contains("already claimed"), e.getMessage());
}
@@ -187,8 +194,8 @@ void duplicateTypeNameInPluginDirectoryIsSkippedButDoesNotStopFe() {
// Same conflict, different blame: two plugin directories shipping one type name is a deployment
// accident. loadPlugins promises partial success — one bad plugin dir must not keep FE from starting —
// so the offender is skipped and the incumbent keeps serving.
- Assertions.assertTrue(manager.registerDiscovered(createProvider("dup", true, "first"), false));
- Assertions.assertFalse(manager.registerDiscovered(createProvider("dup", true, "second"), false),
+ Assertions.assertTrue(admit(createProvider("dup", true, "first"), false));
+ Assertions.assertFalse(admit(createProvider("dup", true, "second"), false),
"the second claimant must be refused, not silently appended");
Assertions.assertEquals(Collections.singletonList("dup"), manager.getRegisteredTypes(),
@@ -205,10 +212,10 @@ void providerClaimingAnEngineBuiltinCatalogTypeIsRefused() {
// creates. Refusing it at registration means the shadowing case cannot arise at all.
for (String builtin : new String[] {"doris", "test", "lakesoul"}) {
Assertions.assertTrue(CatalogFactory.isBuiltinCatalogType(builtin), builtin);
- Assertions.assertFalse(manager.registerDiscovered(createProvider(builtin, true, "x"), false),
+ Assertions.assertFalse(admit(createProvider(builtin, true, "x"), false),
"a plugin must not be allowed to claim engine built-in type '" + builtin + "'");
Assertions.assertThrows(IllegalStateException.class,
- () -> manager.registerDiscovered(createProvider(builtin.toUpperCase(), true, "x"),
+ () -> admit(createProvider(builtin.toUpperCase(), true, "x"),
true),
"on the classpath the same violation must fail loud, case-insensitively");
}
@@ -220,16 +227,77 @@ void providerClaimingAnEngineBuiltinCatalogTypeIsRefused() {
void blankTypeNameIsRefused() {
// getType() is now the sole admission ticket for third-party code; a blank name would otherwise sit in
// the registry and match nothing, or match everything the moment someone compares loosely.
- Assertions.assertFalse(manager.registerDiscovered(createProvider(" ", true, "x"), false),
+ Assertions.assertFalse(admit(createProvider(" ", true, "x"), false),
"a blank type name must be refused");
Assertions.assertTrue(manager.getRegisteredTypes().isEmpty());
}
+ @Test
+ void admittedProviderSecretKeysAreMaskedInShownCatalogProperties() {
+ // SHOW CREATE CATALOG masks only the keys on DatasourcePrintableMap's list. A connector's own secret
+ // keys must reach that list through its security surface, so fe-core never has to name them. The
+ // list is process-wide, hence key names no other test uses.
+ String token = "test.connector.admitted.bearer-token";
+ String passThroughSecret = "test.connector.admitted.passthrough.client-secret";
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey(token));
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey(passThroughSecret));
+
+ Assertions.assertTrue(admit(createProvider("secret_type",
+ ConnectorSecuritySurface.ofSensitiveKeys(token)
+ .withSensitiveKeyPrefix("test.connector.admitted.passthrough.", "secret")), false));
+
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey(token.toUpperCase()),
+ "declared keys must match case-insensitively, like the engine's own list");
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey(passThroughSecret.toUpperCase()),
+ "a declared namespace covers the credential-like keys under it, case-insensitively");
+ Map properties = new LinkedHashMap<>();
+ properties.put("type", "secret_type");
+ properties.put(token, "token-value");
+ properties.put(passThroughSecret, "client-secret-value");
+ properties.put("test.connector.admitted.passthrough.endpoint", "http://host:1");
+ // The rendering SHOW CREATE CATALOG uses (CatalogMgr#showCreateCatalog).
+ String shown = new DatasourcePrintableMap<>(properties, "=", true, true, true, true).toString();
+ Assertions.assertFalse(shown.contains("token-value"), shown);
+ Assertions.assertFalse(shown.contains("client-secret-value"), shown);
+ Assertions.assertTrue(shown.contains("\"test.connector.admitted.passthrough.endpoint\" = \"http://host:1\""),
+ "a key under the namespace without a marker stays visible: " + shown);
+ Assertions.assertTrue(shown.contains("\"" + token + "\" = \"" + DatasourcePrintableMap.PASSWORD_MASK + "\""),
+ shown);
+ Assertions.assertTrue(shown.contains("\"type\" = \"secret_type\""), "undeclared keys stay visible: " + shown);
+ }
+
+ @Test
+ void refusedProviderDoesNotExtendTheMaskingList() {
+ // Only a provider that will serve catalogs contributes keys; a refused claimant leaves no trace.
+ String token = "test.connector.refused.bearer-token";
+ Assertions.assertTrue(admit(createProvider("dup", true, "first"), false));
+ Assertions.assertFalse(admit(createProvider("dup",
+ ConnectorSecuritySurface.ofSensitiveKeys(token)
+ .withSensitiveKeyPrefix("test.connector.refused.passthrough.")), false));
+
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey(token));
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey("test.connector.refused.passthrough.x"));
+ }
+
+ @Test
+ void registerProviderMasksDeclaredSecretKeysToo() {
+ // Tests stand in for a real plugin through registerProvider, so it must mask like the admission path.
+ String token = "test.connector.override.api-key";
+ manager.registerProvider(createProvider("override_type",
+ ConnectorSecuritySurface.ofSsrfEndpoints("override.uri").withSensitiveKeys(token)
+ .withSensitiveKeyPrefix("test.connector.override.passthrough.", "password")));
+
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey(token));
+ Assertions.assertTrue(DatasourcePrintableMap.isSensitiveKey("test.connector.override.passthrough.password"));
+ Assertions.assertFalse(DatasourcePrintableMap.isSensitiveKey("override.uri"),
+ "an endpoint key is policed, not masked");
+ }
+
@Test
void registerProviderStillShadowsADiscoveredType() {
// registerProvider exists to stand in for a real plugin in tests (several rely on shadowing a real
// type name). The uniqueness check must not reach it, or those tests lose their only seam.
- Assertions.assertTrue(manager.registerDiscovered(createProvider("iceberg", true, "real"), false));
+ Assertions.assertTrue(admit(createProvider("iceberg", true, "real"), false));
manager.registerProvider(createProvider("iceberg", true, "override"));
Connector connector = manager.createConnector("iceberg", Collections.emptyMap(), testContext);
@@ -274,6 +342,62 @@ private static int kernelApiMajor() throws IOException {
return Integer.parseInt(declared.substring(0, declared.indexOf('.')));
}
+ @Test
+ void classpathProviderWhoseSecuritySurfaceThrowsIsSkippedNotFatal(@TempDir Path tempDir) throws Exception {
+ // A connector's security surface is plugin code. When it throws, FE skips the provider, as it does a
+ // filesystem provider whose sensitivePropertyKeys() throws, and starts.
+ Path jar = createClasspathProviderJar(tempDir.resolve("throwing-surface.jar"), kernelApiMajor() + ".0",
+ ThrowingSurfaceClasspathProvider.class);
+ ConnectorPluginManager loading = new ConnectorPluginManager();
+ try (URLClassLoader loader = providerClassLoader(jar, ThrowingSurfaceClasspathProvider.class)) {
+ Assertions.assertDoesNotThrow(() -> loading.loadBuiltins(loader));
+ }
+ Assertions.assertFalse(loading.getRegisteredTypes().contains("throwing-surface-test"));
+ }
+
+ @Test
+ void classpathProviderWhoseSecuritySurfaceFailsInItsOwnServiceLoaderIsSkippedNotFatal(@TempDir Path tempDir)
+ throws Exception {
+ // Hive, Iceberg and Paimon read their metastore backends through a ServiceLoader to build their security
+ // surfaces. A service descriptor naming a backend that is not there fails that lookup with
+ // ServiceConfigurationError, an Error: FE must skip the provider and start all the same.
+ Path jar = createClasspathProviderJar(tempDir.resolve("backend-discovering.jar"), kernelApiMajor() + ".0",
+ BackendDiscoveringConnectorProvider.class);
+ Path leftBehind = writeBackendDescriptorJar(tempDir.resolve("left-behind.jar"));
+ ConnectorPluginManager loading = new ConnectorPluginManager();
+ try (URLClassLoader loader =
+ providerClassLoader(jar, BackendDiscoveringConnectorProvider.class, leftBehind)) {
+ Assertions.assertDoesNotThrow(() -> loading.loadBuiltins(loader));
+ }
+ Assertions.assertFalse(loading.getRegisteredTypes().contains(BackendDiscoveringConnectorProvider.TYPE));
+ }
+
+ @Test
+ void pluginWhoseSecuritySurfaceFailsInItsOwnServiceLoaderIsSkippedNotFatal(@TempDir Path tempDir)
+ throws Exception {
+ // The same failure from a plugin directory, as a jar left in lib/ by an earlier release causes when its
+ // descriptor names a backend the plugin no longer ships. loadPlugins promises partial success: the plugin
+ // is skipped, and the plugin next to it still loads.
+ String apiVersion = kernelApiMajor() + ".0";
+ Path root = tempDir.resolve("connector-plugins");
+ PluginJarWriter.writePluginRoot(root, "backend-discovering", BackendDiscoveringConnectorProvider.class,
+ ConnectorProvider.class, "Doris-Connector-Plugin-Api-Version", apiVersion,
+ Collections.singletonList(BackendDiscoveringConnectorProvider.Backend.class));
+ writeBackendDescriptorJar(root.resolve("backend-discovering").resolve("lib").resolve("left-behind.jar"));
+ PluginJarWriter.writePluginRoot(root, "version-probe", VersionProbeConnectorProvider.class,
+ ConnectorProvider.class, "Doris-Connector-Plugin-Api-Version", apiVersion, Collections.emptyList());
+ ConnectorPluginManager loading = new ConnectorPluginManager();
+ try {
+ Assertions.assertDoesNotThrow(() -> loading.loadPlugins(Collections.singletonList(root)));
+ Assertions.assertFalse(loading.getRegisteredTypes().contains(BackendDiscoveringConnectorProvider.TYPE));
+ Assertions.assertTrue(loading.getRegisteredTypes().contains("version_probe"),
+ loading.getRegisteredTypes().toString());
+ } finally {
+ // loadPlugins writes inventory rows into a process-wide singleton.
+ PluginRegistry.getInstance().clearForTest();
+ }
+ }
+
@Test
void incompatibleClasspathProviderIsRejectedBeforeConstruction(@TempDir Path tempDir) throws Exception {
System.clearProperty(CLASSPATH_PROVIDER_CONSTRUCTED);
@@ -294,9 +418,9 @@ void duplicateCreateTableEngineNameIsRefused() {
// Engine names route CREATE TABLE ... ENGINE= the same way type names route CREATE CATALOG. Two
// plugins answering to one engine name would make the statement mean whichever registered first, so
// the conflict is refused where it can still be refused: at registration.
- Assertions.assertTrue(manager.registerDiscovered(
+ Assertions.assertTrue(admit(
createProviderWithEngines("first_type", "shared_engine"), false));
- Assertions.assertFalse(manager.registerDiscovered(
+ Assertions.assertFalse(admit(
createProviderWithEngines("second_type", "SHARED_ENGINE"), false),
"a second claimant of the same engine name must be refused, case-insensitively");
@@ -304,7 +428,7 @@ void duplicateCreateTableEngineNameIsRefused() {
"the refused provider must not end up in the registry");
IllegalStateException e = Assertions.assertThrows(IllegalStateException.class,
- () -> manager.registerDiscovered(createProviderWithEngines("third_type", "shared_engine"), true),
+ () -> admit(createProviderWithEngines("third_type", "shared_engine"), true),
"on the classpath the same conflict must fail loud");
Assertions.assertTrue(e.getMessage().contains("already claimed"), e.getMessage());
}
@@ -315,7 +439,7 @@ void providerClaimingAnEngineReservedEngineNameIsRefused() {
// owe the user a specific "use X instead" message from InternalCatalog. A plugin claiming one would
// silently take over a statement the engine answers for.
for (String reserved : new String[] {"olap", "mysql", "odbc", "broker"}) {
- Assertions.assertFalse(manager.registerDiscovered(
+ Assertions.assertFalse(admit(
createProviderWithEngines("t_" + reserved, reserved.toUpperCase()), false),
"a plugin must not be allowed to claim reserved engine name '" + reserved + "'");
}
@@ -327,10 +451,10 @@ void providerClaimingAnEngineReservedEngineNameIsRefused() {
void refusedProviderDoesNotLeaveItsTypeOrEngineNameClaimed() {
// The checks run before anything is claimed, so a provider rejected for one reason must not poison the
// name it never got. Otherwise a bad plugin directory could permanently disable a good one.
- Assertions.assertFalse(manager.registerDiscovered(createProviderWithEngines("good_type", "olap"), false),
+ Assertions.assertFalse(admit(createProviderWithEngines("good_type", "olap"), false),
"rejected for the reserved engine name");
- Assertions.assertTrue(manager.registerDiscovered(createProviderWithEngines("good_type", "good_engine"),
+ Assertions.assertTrue(admit(createProviderWithEngines("good_type", "good_engine"),
false),
"the type name must still be free after the earlier provider was refused");
Assertions.assertEquals(Collections.singletonList("good_type"), manager.getRegisteredTypes());
@@ -372,14 +496,36 @@ public Connector create(Map properties, ConnectorContext context
}
}
+ public static class ThrowingSurfaceClasspathProvider implements ConnectorProvider {
+ @Override
+ public String getType() {
+ return "throwing-surface-test";
+ }
+
+ @Override
+ public Connector create(Map properties, ConnectorContext context) {
+ return null;
+ }
+
+ @Override
+ public ConnectorSecuritySurface securitySurface() {
+ throw new IllegalStateException("the connector could not build its security surface");
+ }
+ }
+
private static Path createClasspathProviderJar(Path jarPath, String apiVersion) throws IOException {
+ return createClasspathProviderJar(jarPath, apiVersion, ClasspathProvider.class);
+ }
+
+ private static Path createClasspathProviderJar(Path jarPath, String apiVersion,
+ Class providerClass) throws IOException {
Manifest manifest = new Manifest();
manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION, "1.0");
manifest.getMainAttributes().putValue("Doris-Connector-Plugin-Api-Version", apiVersion);
- String classEntry = ClasspathProvider.class.getName().replace('.', '/') + ".class";
+ String classEntry = providerClass.getName().replace('.', '/') + ".class";
try (JarOutputStream jar = new JarOutputStream(Files.newOutputStream(jarPath), manifest)) {
jar.putNextEntry(new JarEntry(classEntry));
- try (InputStream bytes = ClasspathProvider.class.getClassLoader().getResourceAsStream(classEntry)) {
+ try (InputStream bytes = providerClass.getClassLoader().getResourceAsStream(classEntry)) {
Assertions.assertNotNull(bytes, "provider class bytes");
byte[] buffer = new byte[8192];
int read;
@@ -389,17 +535,27 @@ private static Path createClasspathProviderJar(Path jarPath, String apiVersion)
}
jar.closeEntry();
jar.putNextEntry(new JarEntry("META-INF/services/" + ConnectorProvider.class.getName()));
- jar.write((ClasspathProvider.class.getName() + "\n").getBytes(StandardCharsets.UTF_8));
+ jar.write((providerClass.getName() + "\n").getBytes(StandardCharsets.UTF_8));
jar.closeEntry();
}
return jarPath;
}
private static URLClassLoader providerClassLoader(Path jarPath) throws IOException {
- return new URLClassLoader(new URL[] {jarPath.toUri().toURL()}, ConnectorProvider.class.getClassLoader()) {
+ return providerClassLoader(jarPath, ClasspathProvider.class);
+ }
+
+ private static URLClassLoader providerClassLoader(Path jarPath, Class providerClass,
+ Path... moreJars) throws IOException {
+ URL[] urls = new URL[moreJars.length + 1];
+ urls[0] = jarPath.toUri().toURL();
+ for (int i = 0; i < moreJars.length; i++) {
+ urls[i + 1] = moreJars[i].toUri().toURL();
+ }
+ return new URLClassLoader(urls, ConnectorProvider.class.getClassLoader()) {
@Override
protected Class loadClass(String name, boolean resolve) throws ClassNotFoundException {
- if (name.equals(ClasspathProvider.class.getName())) {
+ if (name.equals(providerClass.getName())) {
synchronized (getClassLoadingLock(name)) {
Class loaded = findLoadedClass(name);
if (loaded == null) {
@@ -416,10 +572,49 @@ protected Class loadClass(String name, boolean resolve) throws ClassNotFoundE
};
}
+ /**
+ * Writes a jar holding only a service descriptor for {@link BackendDiscoveringConnectorProvider.Backend} that
+ * names a class no jar has, as a jar left behind by an earlier release would.
+ */
+ private static Path writeBackendDescriptorJar(Path jarPath) throws IOException {
+ Files.createDirectories(jarPath.getParent());
+ try (JarOutputStream jar = new JarOutputStream(Files.newOutputStream(jarPath))) {
+ jar.putNextEntry(new JarEntry(
+ "META-INF/services/" + BackendDiscoveringConnectorProvider.Backend.class.getName()));
+ jar.write("org.apache.doris.connectorconf.testplugins.RemovedBackend\n".getBytes(StandardCharsets.UTF_8));
+ jar.closeEntry();
+ }
+ return jarPath;
+ }
+
+ /** Admits {@code provider} as both loading paths do: with its security surface read up front. */
+ private boolean admit(ConnectorProvider provider, boolean failFast) {
+ return manager.registerDiscovered(provider, provider.securitySurface(), failFast);
+ }
+
private static ConnectorProvider createProvider(String type) {
return createProvider(type, true, "");
}
+ private static ConnectorProvider createProvider(String type, ConnectorSecuritySurface surface) {
+ return new ConnectorProvider() {
+ @Override
+ public String getType() {
+ return type;
+ }
+
+ @Override
+ public Connector create(Map properties, ConnectorContext context) {
+ return new TaggedConnector("");
+ }
+
+ @Override
+ public ConnectorSecuritySurface securitySurface() {
+ return surface;
+ }
+ };
+ }
+
private static ConnectorProvider createProvider(String type, boolean standalone, String tag) {
return new ConnectorProvider() {
@Override
diff --git a/fe/fe-core/src/test/java/org/apache/doris/connectorconf/testplugins/BackendDiscoveringConnectorProvider.java b/fe/fe-core/src/test/java/org/apache/doris/connectorconf/testplugins/BackendDiscoveringConnectorProvider.java
new file mode 100644
index 00000000000000..5f8c5e8f968b34
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/connectorconf/testplugins/BackendDiscoveringConnectorProvider.java
@@ -0,0 +1,69 @@
+// 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.connectorconf.testplugins;
+
+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.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+
+/**
+ * A connector provider that builds its security surface from backends it discovers through its own
+ * {@link ServiceLoader}, the way the Hive, Iceberg and Paimon providers declare their metastore backends' secret
+ * keys through {@code MetaStoreProviders}. Its class bytes are copied into a temporary jar, so a test can load it
+ * the way FE loads a shipped connector.
+ *
+ * When a service descriptor in the provider's jars names a backend that cannot be loaded, the lookup throws
+ * {@link java.util.ServiceConfigurationError}, which is an {@link Error} and neither a {@link RuntimeException}
+ * nor a {@link LinkageError}.
+ *
+ *
It lives outside {@code org.apache.doris.connector.} for the reason {@link ConfProbeConnectorProviderA} does.
+ */
+public class BackendDiscoveringConnectorProvider implements ConnectorProvider {
+
+ public static final String TYPE = "backend_discovering";
+
+ /** What the provider's backends implement; its jars register them under this interface's name. */
+ public interface Backend {
+ String sensitiveKey();
+ }
+
+ @Override
+ public String getType() {
+ return TYPE;
+ }
+
+ @Override
+ public Connector create(Map properties, ConnectorContext context) {
+ throw new UnsupportedOperationException(TYPE + " exists to be admitted or skipped at load time");
+ }
+
+ @Override
+ public ConnectorSecuritySurface securitySurface() {
+ List keys = new ArrayList<>();
+ for (Backend backend : ServiceLoader.load(Backend.class, getClass().getClassLoader())) {
+ keys.add(backend.sensitiveKey());
+ }
+ return ConnectorSecuritySurface.EMPTY.withSensitiveKeys(keys);
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/parser/EncryptSQLTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/parser/EncryptSQLTest.java
index 5ccb3ca9400524..e047f38e0d58cc 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/parser/EncryptSQLTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/parser/EncryptSQLTest.java
@@ -161,6 +161,20 @@ public void testEncryption() throws Exception {
+ ")";
parseAndCheck(sql, res);
+ // ALTER CATALOG sets the same secrets as CREATE CATALOG, so its audit text is masked the same way.
+ sql = "ALTER CATALOG ctl SET PROPERTIES("
+ + " \"s3.endpoint\" = \"s3.us-east-1.amazonaws.com\","
+ + " \"password\" = \"abc\","
+ + " \"s3.secret_key\" = \"abc\""
+ + ")";
+
+ res = "ALTER CATALOG ctl SET PROPERTIES("
+ + " \"s3.endpoint\" = \"s3.us-east-1.amazonaws.com\","
+ + " \"password\" = \"*XXX\","
+ + " \"s3.secret_key\" = \"*XXX\""
+ + ")";
+ parseAndCheck(sql, res);
+
sql = "CREATE TABLE mysql_tbl("
+ " k1 DATE,"
+ " k2 INT,"
diff --git a/regression-test/suites/audit/test_external_catalog_credential_masking.groovy b/regression-test/suites/audit/test_external_catalog_credential_masking.groovy
index b7b1bd4aebc530..b13febbe1a96eb 100644
--- a/regression-test/suites/audit/test_external_catalog_credential_masking.groovy
+++ b/regression-test/suites/audit/test_external_catalog_credential_masking.groovy
@@ -18,18 +18,22 @@
suite("test_external_catalog_credential_masking", "p0,nonConcurrent") {
String credentialCatalog = "audit_iceberg_credential_masking"
String tokenCatalog = "audit_iceberg_token_masking"
+ String jdbcCatalog = "audit_iceberg_jdbc_password_masking"
// Low-entropy placeholders keep this negative masking test distinct from real credentials.
String accessKey = "aaaaaaaaaaaaaaaaaaaa"
String secretKey = "bbbbbbbbbbbbbbbbbbbb"
String oauthCredential = "cccccccccccccccccccc"
String oauthToken = "dddddddddddddddddddd"
+ String jdbcPassword = "eeeeeeeeeeeeeeeeeeee"
+ String alteredJdbcPassword = "ffffffffffffffffffff"
- def waitForAuditStatement = { String catalogName ->
+ // The audit log files CREATE CATALOG under stmt_type OTHER and ALTER CATALOG under ALTER.
+ def waitForAuditStatement = { String catalogName, String statement = "create catalog", String stmtType = "OTHER" ->
String query = """
select stmt
from __internal_schema.audit_log
- where stmt_type = 'OTHER'
- and lower(stmt) like '%create catalog ${catalogName}%'
+ where stmt_type = '${stmtType}'
+ and lower(stmt) like '%${statement} ${catalogName}%'
and instr(stmt, '*XXX') > 0
order by time asc
limit 1
@@ -51,6 +55,7 @@ suite("test_external_catalog_credential_masking", "p0,nonConcurrent") {
try {
sql """drop catalog if exists ${credentialCatalog}"""
sql """drop catalog if exists ${tokenCatalog}"""
+ sql """drop catalog if exists ${jdbcCatalog}"""
sql """truncate table __internal_schema.audit_log"""
sql """
@@ -76,6 +81,38 @@ suite("test_external_catalog_credential_masking", "p0,nonConcurrent") {
'iceberg.rest.oauth2.token' = '${oauthToken}'
)
"""
+ // iceberg.jdbc.password is on no list fe-core keeps: only the iceberg connector knows it, and
+ // declares it on its security surface. The catalog is never initialized, so nothing connects.
+ sql """
+ create catalog ${jdbcCatalog} properties (
+ 'type' = 'iceberg',
+ 'iceberg.catalog.type' = 'jdbc',
+ 'uri' = 'jdbc:mysql://127.0.0.1:1/iceberg',
+ 'iceberg.jdbc.catalog_name' = 'audit_masking',
+ 'iceberg.jdbc.user' = 'iceberg',
+ 'iceberg.jdbc.password' = '${jdbcPassword}',
+ 'warehouse' = 's3://audit-masking/warehouse',
+ 's3.access_key' = '${accessKey}',
+ 's3.secret_key' = '${secretKey}',
+ 's3.endpoint' = 'http://127.0.0.1:1',
+ 's3.region' = 'us-east-1'
+ )
+ """
+ String shownJdbc = sql("show create catalog ${jdbcCatalog}")[0][1].toString()
+ assertFalse(shownJdbc.contains(jdbcPassword))
+ assertTrue(shownJdbc.contains('"iceberg.jdbc.password" = "*XXX"'))
+ assertTrue(shownJdbc.contains('"iceberg.jdbc.user" = "iceberg"'))
+
+ // ALTER CATALOG sets the same secrets, so its statement text must be masked in the audit log too.
+ sql """
+ alter catalog ${jdbcCatalog} set properties (
+ 'iceberg.jdbc.password' = '${alteredJdbcPassword}',
+ 's3.secret_key' = '${secretKey}'
+ )
+ """
+ shownJdbc = sql("show create catalog ${jdbcCatalog}")[0][1].toString()
+ assertFalse(shownJdbc.contains(alteredJdbcPassword))
+ assertTrue(shownJdbc.contains('"iceberg.jdbc.password" = "*XXX"'))
sql """call flush_audit_log()"""
// Audit statements are independently persisted after parsing, so they must use the
@@ -91,9 +128,20 @@ suite("test_external_catalog_credential_masking", "p0,nonConcurrent") {
String tokenStmt = waitForAuditStatement(tokenCatalog)
assertFalse(tokenStmt.contains(oauthToken))
assertTrue(tokenStmt.contains('"iceberg.rest.oauth2.token" = "*XXX"'))
+
+ String jdbcCreateStmt = waitForAuditStatement(jdbcCatalog)
+ assertFalse(jdbcCreateStmt.contains(jdbcPassword))
+ assertTrue(jdbcCreateStmt.contains('"iceberg.jdbc.password" = "*XXX"'))
+
+ String jdbcAlterStmt = waitForAuditStatement(jdbcCatalog, "alter catalog", "ALTER")
+ assertFalse(jdbcAlterStmt.contains(alteredJdbcPassword))
+ assertFalse(jdbcAlterStmt.contains(secretKey))
+ assertTrue(jdbcAlterStmt.contains('"iceberg.jdbc.password" = "*XXX"'))
+ assertTrue(jdbcAlterStmt.contains('"s3.secret_key" = "*XXX"'))
} finally {
sql """drop catalog if exists ${credentialCatalog}"""
sql """drop catalog if exists ${tokenCatalog}"""
+ sql """drop catalog if exists ${jdbcCatalog}"""
}
})
}