Apply the nimbus ACLs to listBlobs so callers only see blobs they may read

storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java

@@ -39,6 +39,7 @@ public class SimpleACLAuthorizer implements IAuthorizer {
         "submitTopology",
         "fileUpload",
         "getNimbusConf",
+        "listBlobs",
         "getClusterInfo",
         "getLeader",
         "isTopologyNameAllowed",

storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java

@@ -68,6 +68,11 @@ public void SimpleACLUserAuthTest() {
         assertTrue(authorizer.permit(new ReqContext(userA), "getNimbusConf", new HashMap<>()));
         assertTrue(authorizer.permit(new ReqContext(userB), "getNimbusConf", new HashMap<>()));
 
+        assertTrue(authorizer.permit(new ReqContext(adminUser), "listBlobs", new HashMap<>()));
+        assertFalse(authorizer.permit(new ReqContext(supervisorUser), "listBlobs", new HashMap<>()));
+        assertTrue(authorizer.permit(new ReqContext(userA), "listBlobs", new HashMap<>()));
+        assertTrue(authorizer.permit(new ReqContext(userB), "listBlobs", new HashMap<>()));
+
         assertTrue(authorizer.permit(new ReqContext(adminUser), "getClusterInfo", new HashMap<>()));
         assertFalse(authorizer.permit(new ReqContext(supervisorUser), "getClusterInfo", new HashMap<>()));
         assertTrue(authorizer.permit(new ReqContext(userA), "getClusterInfo", new HashMap<>()));

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;
@@ -4070,6 +4070,7 @@
     @Override
     public ListBlobsResult listBlobs(String session) throws TException {
         try {
+            checkAuthorization(null, null, "listBlobs");
             Iterator<String> keyIt;
             //Create a new session id if the user gave an empty session string.
             // This is the use case when the user wishes to list blobs
@@ -4092,9 +4093,17 @@
                 return new ListBlobsResult(Collections.emptyList(), session);
             }
 
+            Subject who = getSubject();
             ArrayList<String> listChunk = new ArrayList<>();
-            for (int i = 0; i < 100 && keyIt.hasNext(); i++) {
-                listChunk.add(keyIt.next());
+            while (listChunk.size() < 100 && keyIt.hasNext()) {
+                String key = keyIt.next();
+                //Only list the blobs whose metadata the caller may read, the same check getBlobMeta does.
+                try {
+                    blobStore.getBlobMeta(key, who);
+                    listChunk.add(key);
+                } catch (AuthorizationException | KeyNotFoundException e) {
+                    LOG.debug("Not listing blob {} for {}", key, who);
+                }
             }
             blobListers.put(session, keyIt);
             LOG.info("Downloading {} entries", listChunk.size());
@@ -4831,8 +4840,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/NimbusTest.java

@@ -23,6 +23,7 @@
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -39,6 +40,7 @@
 import org.apache.storm.generated.AuthorizationException;
 import org.apache.storm.generated.InvalidTopologyException;
 import org.apache.storm.generated.KeyNotFoundException;
+import org.apache.storm.generated.ListBlobsResult;
 import org.apache.storm.generated.RebalanceOptions;
 import org.apache.storm.generated.StormTopology;
 import org.apache.storm.metric.StormMetricsRegistry;
@@ -52,11 +54,13 @@
 import org.apache.storm.security.auth.IGroupMappingServiceProvider;
 import org.apache.storm.security.auth.ReqContext;
 import org.apache.storm.security.auth.SingleUserPrincipal;
+import org.apache.storm.security.auth.authorizer.DenyAuthorizer;
 import org.apache.storm.testing.TestWordSpout;
 import org.apache.storm.thrift.TException;
 import org.apache.storm.topology.TopologyBuilder;
 import org.apache.storm.utils.ServerUtils;
 import org.apache.storm.utils.Time;
+import org.apache.storm.utils.WrappedAuthorizationException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
@@ -214,6 +218,28 @@ void testCreateStateInZookeeperWhenKeyNotFoundHandlesException() throws Exceptio
         }
     }
 
+    @Test
+    void testListBlobsOnlyReturnsKeysTheCallerMayReadTheMetadataOf() throws Exception {
+        when(localBlobStore.listKeys()).thenReturn(List.of("readable-key", "other-users-key").iterator());
+        when(localBlobStore.getBlobMeta(eq("other-users-key"), any()))
+            .thenThrow(new WrappedAuthorizationException("not allowed"));
+
+        ListBlobsResult result = nimbus.listBlobs("");
+
+        assertEquals(List.of("readable-key"), result.get_keys());
+    }
+
+    @Test
+    void testListBlobsIsAuthorized() throws Exception {
+        Map<String, Object> conf = Map.of(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10,
+                                          DaemonConfig.NIMBUS_AUTHORIZER, DenyAuthorizer.class.getName());
+        nimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo, localBlobStore, leaderElector, groupMapper, metricRegistry);
+        when(localBlobStore.listKeys()).thenReturn(List.of("readable-key").iterator());
+
+        assertThrows(AuthorizationException.class, () -> nimbus.listBlobs(""));
+        verify(localBlobStore, never()).listKeys();
+    }
+
     @Test
     void testValidateUploadedJarLocationRejectsLocationsOutsideTheInbox() throws Exception {
         Path inbox = Files.createTempDirectory("nimbus-inbox");