Require every UI API endpoint to declare its authorization

storm-client/src/jvm/org/apache/storm/Config.java

@@ -1240,6 +1240,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the keystore that the nimbus TLS server uses.
      */
     @IsString
+    @Password
     public static final String NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD = "nimbus.thrift.tls.server.keystore.password";
 
     /**
@@ -1258,6 +1259,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the truststore that the nimbus TLS server uses.
      */
     @IsString
+    @Password
     public static final String NIMBUS_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD = "nimbus.thrift.tls.server.truststore.password";
 
     /**
@@ -1272,6 +1274,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the keystore that the nimbus TLS client uses.
      */
     @IsString
+    @Password
     public static final String NIMBUS_THRIFT_TLS_CLIENT_KEYSTORE_PASSWORD = "nimbus.thrift.tls.client.keystore.password";
 
     /**
@@ -1296,6 +1299,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the truststore that the nimbus TLS client uses.
      */
     @IsString
+    @Password
     public static final String NIMBUS_THRIFT_TLS_CLIENT_TRUSTSTORE_PASSWORD = "nimbus.thrift.tls.client.truststore.password";
 
     /**
@@ -1351,6 +1355,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the keystore that the supervisor TLS server uses.
      */
     @IsString
+    @Password
     public static final String SUPERVISOR_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD = "supervisor.thrift.tls.server.keystore.password";
 
     /**
@@ -1363,6 +1368,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the truststore that the supervisor TLS server uses.
      */
     @IsString
+    @Password
     public static final String SUPERVISOR_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD = "supervisor.thrift.tls.server.truststore.password";
 
     /**
@@ -1375,6 +1381,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the keystore that the supervisor TLS client uses.
      */
     @IsString
+    @Password
     public static final String SUPERVISOR_THRIFT_TLS_CLIENT_KEYSTORE_PASSWORD = "supervisor.thrift.tls.client.keystore.password";
 
     /**
@@ -1387,6 +1394,7 @@ public class Config extends HashMap<String, Object> {
      * The password of the truststore that the supervisor TLS client uses.
      */
     @IsString
+    @Password
     public static final String SUPERVISOR_THRIFT_TLS_CLIENT_TRUSTSTORE_PASSWORD = "supervisor.thrift.tls.client.truststore.password";
 
     /**
@@ -1534,6 +1542,7 @@ public class Config extends HashMap<String, Object> {
      * Keystore password for ZooKeeper client connection over SSL.
      */
     @IsString
+    @Password
     public static final String STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD = "storm.zookeeper.ssl.keystore.password";
     /**
      * Truststore location for ZooKeeper client connection over SSL.
@@ -1544,6 +1553,7 @@ public class Config extends HashMap<String, Object> {
      * Truststore password for ZooKeeper client connection over SSL.
      */
     @IsString
+    @Password
     public static final String STORM_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD = "storm.zookeeper.ssl.truststore.password";
     /**
      * Enable or disable hostname verification.
@@ -1735,6 +1745,7 @@ public class Config extends HashMap<String, Object> {
      * deny access from workers.
      */
     @IsString
+    @Password
     public static final String STORM_ZOOKEEPER_AUTH_PAYLOAD = "storm.zookeeper.auth.payload";
     /**
      * What Network Topography detection classes should we use. Given a list of supervisor hostnames (or IP addresses), this class would
@@ -1925,6 +1936,7 @@ public class Config extends HashMap<String, Object> {
      * Netty based messaging: Specifies the truststore password when TLS is enabled.
      */
     @IsString
+    @Password
     public static final String STORM_MESSAGING_NETTY_TLS_TRUSTSTORE_PASSWORD = "storm.messaging.netty.tls.truststore.password";
 
     /**
@@ -1937,6 +1949,7 @@ public class Config extends HashMap<String, Object> {
      * Netty based messaging: Specifies the keystore password when TLS is enabled.
      */
     @IsString
+    @Password
     public static final String STORM_MESSAGING_NETTY_TLS_KEYSTORE_PASSWORD = "storm.messaging.netty.tls.keystore.password";
 
     /**
@@ -1949,6 +1962,7 @@ public class Config extends HashMap<String, Object> {
      * Netty based messaging: Specifies the client truststore password when TLS is enabled.
      */
     @IsString
+    @Password
     public static final String STORM_MESSAGING_NETTY_TLS_CLIENT_TRUSTSTORE_PASSWORD =
         "storm.messaging.netty.tls.client.truststore.password";
 
@@ -1962,6 +1976,7 @@ public class Config extends HashMap<String, Object> {
      * Netty based messaging: Specifies the client keystore password when TLS is enabled.
      */
     @IsString
+    @Password
     public static final String STORM_MESSAGING_NETTY_TLS_CLIENT_KEYSTORE_PASSWORD =
         "storm.messaging.netty.tls.client.keystore.password";
 

storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java

@@ -25,6 +25,7 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.function.BooleanSupplier;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import org.apache.storm.Config;
@@ -42,6 +43,7 @@ public class ConfigUtils {
     public static final double RFC1889_ALPHA = 1.0 / 16.0;
 
     private static final Set<String> passwordConfigKeys = new HashSet<>();
+    private static final Pattern CREDENTIAL_KEY_NAME = Pattern.compile("(?i)(password|passwd|secret)");
 
     static {
         for (Class<?> clazz : ConfigValidation.getConfigClasses()) {
@@ -89,6 +91,41 @@ public Object transformEntry(String key, Object value) {
         return Maps.transformEntries(conf, maskPasswords);
     }
 
+    /**
+     * Mask credential values before a config map is served over an API. This covers what
+     * {@link #maskPasswords(Map)} covers, plus string values whose key name denotes a secret: plugins read their
+     * own keys straight out of the config map, so those keys are declared by no annotated field and the annotation
+     * scan cannot see them. Only string values are considered, so timeouts and class lists whose names merely
+     * mention credentials keep their value.
+     *
+     * @param conf the config to mask
+     * @return a view of the config with credential values replaced
+     */
+    public static Map<String, Object> maskCredentials(final Map<String, Object> conf) {
+        Maps.EntryTransformer<String, Object, Object> maskCredentials = new Maps.EntryTransformer<String, Object, Object>() {
+            @Override
+            public Object transformEntry(String key, Object value) {
+                if (passwordConfigKeys.contains(key)) {
+                    return "*****";
+                }
+                return value instanceof String && CREDENTIAL_KEY_NAME.matcher(key).find() ? "*****" : value;
+            }
+        };
+        return Maps.transformEntries(conf, maskCredentials);
+    }
+
+    /**
+     * Whether a config key holds a credential, and therefore whether {@link #maskCredentials(Map)} would replace its
+     * value. Callers that read a config back from a daemon use this to tell which entries carry no usable value and
+     * must be taken from their own configuration instead.
+     *
+     * @param key the config key
+     * @return true when the key denotes a credential
+     */
+    public static boolean isCredentialKey(String key) {
+        return passwordConfigKeys.contains(key) || CREDENTIAL_KEY_NAME.matcher(key).find();
+    }
+
     public static boolean isLocalMode(Map<String, Object> conf) {
         String mode = (String) conf.get(Config.STORM_CLUSTER_MODE);
         if (mode != null) {

storm-client/test/jvm/org/apache/storm/utils/ConfigUtilsTest.java

@@ -195,4 +195,118 @@ public void upstreamFeedbackFreqSecs_rejectsNegative() {
         assertThrows(IllegalArgumentException.class, () -> ConfigUtils.upstreamFeedbackFreqSecs(
             mockMap(Config.TOPOLOGY_UPSTREAM_FEEDBACK_FREQ_SECS, -1)));
     }
+
+    @Test
+    public void maskPasswords_masksClusterZookeeperCredentials() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD, "zk-user:zk-secret");
+        conf.put(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD, "topo-user:topo-secret");
+
+        Map<String, Object> masked = ConfigUtils.maskPasswords(conf);
+
+        assertEquals("*****", masked.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD));
+        assertEquals("*****", masked.get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD));
+    }
+
+    @Test
+    public void maskPasswords_masksThriftTlsStorePasswords() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD, "nimbus-ks");
+        conf.put(Config.NIMBUS_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD, "nimbus-ts");
+        conf.put(Config.NIMBUS_THRIFT_TLS_CLIENT_KEYSTORE_PASSWORD, "client-ks");
+        conf.put(Config.NIMBUS_THRIFT_TLS_CLIENT_TRUSTSTORE_PASSWORD, "client-ts");
+        conf.put(Config.SUPERVISOR_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD, "sup-ks");
+        conf.put(Config.SUPERVISOR_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD, "sup-ts");
+
+        Map<String, Object> masked = ConfigUtils.maskPasswords(conf);
+
+        assertEquals("*****", masked.get(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.NIMBUS_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.NIMBUS_THRIFT_TLS_CLIENT_KEYSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.NIMBUS_THRIFT_TLS_CLIENT_TRUSTSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.SUPERVISOR_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.SUPERVISOR_THRIFT_TLS_SERVER_TRUSTSTORE_PASSWORD));
+    }
+
+    @Test
+    public void maskPasswords_masksZookeeperAndNettyTlsStorePasswords() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD, "zk-ks");
+        conf.put(Config.STORM_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD, "zk-ts");
+        conf.put(Config.STORM_MESSAGING_NETTY_TLS_KEYSTORE_PASSWORD, "netty-ks");
+        conf.put(Config.STORM_MESSAGING_NETTY_TLS_TRUSTSTORE_PASSWORD, "netty-ts");
+
+        Map<String, Object> masked = ConfigUtils.maskPasswords(conf);
+
+        assertEquals("*****", masked.get(Config.STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.STORM_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.STORM_MESSAGING_NETTY_TLS_KEYSTORE_PASSWORD));
+        assertEquals("*****", masked.get(Config.STORM_MESSAGING_NETTY_TLS_TRUSTSTORE_PASSWORD));
+    }
+
+    @Test
+    public void maskCredentials_masksKeysThatOnlyPluginsDeclare() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put("storm.daemon.metrics.reporter.plugin.prometheus.basic_auth_password", "plugin-secret");
+        conf.put("storm.zookeeper.auth.password", "zk-pass");
+        conf.put("some.plugin.shared_secret", "shared");
+
+        Map<String, Object> masked = ConfigUtils.maskCredentials(conf);
+
+        assertEquals("*****", masked.get("storm.daemon.metrics.reporter.plugin.prometheus.basic_auth_password"));
+        assertEquals("*****", masked.get("storm.zookeeper.auth.password"));
+        assertEquals("*****", masked.get("some.plugin.shared_secret"));
+    }
+
+    @Test
+    public void maskCredentials_masksTheAnnotatedKeysAsWell() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD, "zk-user:zk-secret");
+
+        assertEquals("*****", ConfigUtils.maskCredentials(conf).get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD));
+    }
+
+    @Test
+    public void maskCredentials_leavesNonStringValuesAlone() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put("task.credentials.poll.secs", 30);
+        conf.put("nimbus.credential.renewers.freq.secs", 600);
+        conf.put("topology.auto-credentials", Collections.singletonList("org.example.AutoCreds"));
+        conf.put("nimbus.seeds", Collections.singletonList("nimbus1"));
+
+        Map<String, Object> masked = ConfigUtils.maskCredentials(conf);
+
+        assertEquals(30, masked.get("task.credentials.poll.secs"));
+        assertEquals(600, masked.get("nimbus.credential.renewers.freq.secs"));
+        assertEquals(Collections.singletonList("org.example.AutoCreds"), masked.get("topology.auto-credentials"));
+        assertEquals(Collections.singletonList("nimbus1"), masked.get("nimbus.seeds"));
+    }
+
+    @Test
+    public void isCredentialKey_recognisesAnnotatedAndPluginDeclaredKeys() {
+        assertTrue(ConfigUtils.isCredentialKey(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD));
+        assertTrue(ConfigUtils.isCredentialKey(Config.NIMBUS_THRIFT_TLS_CLIENT_KEYSTORE_PASSWORD));
+        assertTrue(ConfigUtils.isCredentialKey("storm.daemon.metrics.reporter.plugin.prometheus.basic_auth_password"));
+        assertTrue(ConfigUtils.isCredentialKey("some.plugin.shared_secret"));
+    }
+
+    @Test
+    public void isCredentialKey_ignoresKeysThatOnlyMentionCredentials() {
+        assertFalse(ConfigUtils.isCredentialKey("task.credentials.poll.secs"));
+        assertFalse(ConfigUtils.isCredentialKey(Config.TOPOLOGY_AUTO_CREDENTIALS));
+        assertFalse(ConfigUtils.isCredentialKey(Config.NIMBUS_THRIFT_TLS_CLIENT_KEYSTORE_PATH));
+        assertFalse(ConfigUtils.isCredentialKey(Config.TOPOLOGY_NAME));
+    }
+
+    @Test
+    public void maskPasswords_keepsOrdinaryValues() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.STORM_ZOOKEEPER_SERVERS, Collections.singletonList("zk1"));
+        conf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PATH, "/etc/storm/nimbus.jks");
+
+        Map<String, Object> masked = ConfigUtils.maskPasswords(conf);
+
+        assertEquals(Collections.singletonList("zk1"), masked.get(Config.STORM_ZOOKEEPER_SERVERS));
+        assertEquals("/etc/storm/nimbus.jks", masked.get(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PATH));
+    }
 }

storm-core/src/jvm/org/apache/storm/command/UploadCredentials.java

@@ -24,6 +24,7 @@
 import org.apache.storm.StormSubmitter;
 import org.apache.storm.generated.Nimbus;
 import org.apache.storm.generated.TopologySummary;
+import org.apache.storm.utils.ConfigUtils;
 import org.apache.storm.utils.NimbusClient;
 import org.apache.storm.utils.Utils;
 import org.slf4j.Logger;
@@ -113,6 +114,9 @@ public static void main(String[] args) throws Exception {
          */
         topologyConf.remove("java.security.auth.login.config");
         topologyConf.remove(Config.NIMBUS_THRIFT_CLIENT_USE_TLS);
+        // Nimbus masks credentials before serving a conf, so these entries hold no usable value here.
+        // Dropping them lets the client's own configuration supply them, e.g. TLS store passwords.
+        topologyConf.keySet().removeIf(ConfigUtils::isCredentialKey);
 
         boolean throwExceptionForEmptyCreds = (boolean) cl.get("e");
         boolean hasCreds = StormSubmitter.pushCredentials(topologyName, topologyConf, credentialsMap, (String) cl.get("u"));

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -1103,7 +1103,7 @@
         cleanable.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(store.storedTopoIds(), EMPTY_STRING_SET));
         cleanable.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
         Set<String> delayedCleanable = getExpiredTopologyIds(cleanable, conf);
         delayedCleanable.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
@@ -1223,8 +1223,8 @@
         ret.put(Config.TOPOLOGY_WORKER_NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
         ret.put(Config.NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
 
         if (!mergedConf.containsKey(Config.TOPOLOGY_METRICS_REPORTERS) && mergedConf.containsKey(Config.STORM_METRICS_REPORTERS)) {
             ret.put(Config.TOPOLOGY_METRICS_REPORTERS, mergedConf.get(Config.STORM_METRICS_REPORTERS));
         }
 
         // add any system metrics reporters to the topology metrics reporters
@@ -2945,7 +2945,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyJarsInTopology(topoId);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3420,8 +3420,8 @@
                 waitForDesiredCodeReplication(totalConf, topoId);
                 state.setupHeatbeats(topoId, topoConf);
                 state.setupErrors(topoId, topoConf);
                 if (ObjectReader.getBoolean(totalConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false)) {
                     state.setupBackpressure(topoId, topoConf);
                 }
                 notifyTopologyActionListener(topoName, "submitTopology");
                 TopologyStatus status = null;
@@ -4249,7 +4249,7 @@
         try {
             getNimbusConfCalls.mark();
             checkAuthorization(null, null, "getNimbusConf");
-            return JSONValue.toJSONString(conf);
+            return JSONValue.toJSONString(ConfigUtils.maskCredentials(conf));
         } catch (Exception e) {
             LOG.warn("get nimbus conf exception.", e);
             if (e instanceof TException) {
@@ -4828,8 +4828,8 @@
             String topoName = (String) checkConf.get(Config.TOPOLOGY_NAME);
             checkAuthorization(topoName, checkConf, "getTopologyConf");
             Map<String, Object> maskedConf = new HashMap<>(ConfigUtils.maskPasswords(topoConf));
             if (maskedConf.get(BlowfishTupleSerializer.SECRET_KEY) instanceof String) {
                 maskedConf.put(BlowfishTupleSerializer.SECRET_KEY, "*****");
             }
             return JSONValue.toJSONString(maskedConf);
         } catch (Exception e) {

storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusGetNimbusConfTest.java

@@ -0,0 +1,72 @@
+/*
+ * 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.storm.daemon.nimbus;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import net.minidev.json.JSONValue;
+import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.LocalCluster;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class NimbusGetNimbusConfTest {
+
+    private static final String MASKED = "*****";
+
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> parse(String json) {
+        return (Map<String, Object>) JSONValue.parse(json);
+    }
+
+    @Test
+    public void getNimbusConfMasksCredentialsAndKeepsOtherValues() throws Exception {
+        Map<String, Object> daemonConf = new HashMap<>();
+        daemonConf.put(DaemonConfig.NIMBUS_AUTHORIZER, "org.apache.storm.security.auth.authorizer.NoopAuthorizer");
+        daemonConf.put(DaemonConfig.SUPERVISOR_AUTHORIZER, "org.apache.storm.security.auth.authorizer.NoopAuthorizer");
+        daemonConf.put(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD, "zk-user:zk-secret");
+        daemonConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD, "nimbus-keystore-secret");
+        daemonConf.put(DaemonConfig.UI_HTTPS_KEYSTORE_PASSWORD, "ui-keystore-secret");
+        daemonConf.put(Config.STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD, "zk-ssl-keystore-secret");
+        daemonConf.put("storm.daemon.metrics.reporter.plugin.prometheus.basic_auth_password", "plugin-secret");
+        daemonConf.put(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PATH, "/etc/storm/nimbus.jks");
+
+        try (LocalCluster cluster = new LocalCluster.Builder().withDaemonConf(daemonConf).build()) {
+            Map<String, Object> served = parse(cluster.getNimbus().getNimbusConf());
+
+            assertEquals(MASKED, served.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD),
+                "the cluster ZooKeeper auth payload should be masked");
+            assertEquals(MASKED, served.get(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PASSWORD),
+                "thrift TLS store passwords should be masked");
+            assertEquals(MASKED, served.get(DaemonConfig.UI_HTTPS_KEYSTORE_PASSWORD),
+                "UI keystore passwords should be masked");
+
+            assertEquals(MASKED, served.get(Config.STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD),
+                "ZooKeeper TLS store passwords should be masked");
+            assertEquals(MASKED, served.get("storm.daemon.metrics.reporter.plugin.prometheus.basic_auth_password"),
+                "credential keys that only a plugin declares should be masked too");
+
+            assertEquals("/etc/storm/nimbus.jks", served.get(Config.NIMBUS_THRIFT_TLS_SERVER_KEYSTORE_PATH),
+                "non-credential values should be served unchanged");
+        }
+    }
+}