Reject a topology submission that lists a blob key which is not a dependency blob key

storm-client/src/jvm/org/apache/storm/dependency/DependencyBlobStoreUtils.java

@@ -24,12 +24,28 @@
 
 public class DependencyBlobStoreUtils {
 
-    private static final String BLOB_DEPENDENCIES_PREFIX = "dep-";
+    /**
+     * The prefix every blob key holding a topology dependency starts with.
+     */
+    public static final String BLOB_DEPENDENCIES_PREFIX = "dep-";
 
     public static String generateDependencyBlobKey(String key) {
         return BLOB_DEPENDENCIES_PREFIX + key;
     }
 
+    /**
+     * Tell whether a blob key names a topology dependency, i.e. whether it could have been produced by
+     * {@link #generateDependencyBlobKey(String)}. Keys that a topology only refers to, rather than owns, must be
+     * checked with this before they are acted upon, because the dependency lists of a submitted topology are filled
+     * in by the client and can name any blob at all.
+     *
+     * @param key the blob key to check, may be null
+     * @return true if the key is a dependency blob key
+     */
+    public static boolean isDependencyBlobKey(String key) {
+        return key != null && key.startsWith(BLOB_DEPENDENCIES_PREFIX);
+    }
+
     @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
     public static String applyUUIDToFileName(String fileName) {
         String fileNameWithExt = Files.getNameWithoutExtension(fileName);

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

@@ -87,6 +87,7 @@
 import org.apache.storm.daemon.Shutdownable;
 import org.apache.storm.daemon.StormCommon;
 import org.apache.storm.daemon.common.FileWatcher;
+import org.apache.storm.dependency.DependencyBlobStoreUtils;
 import org.apache.storm.generated.AlreadyAliveException;
 import org.apache.storm.generated.Assignment;
 import org.apache.storm.generated.AuthorizationException;
@@ -1103,7 +1104,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 +1224,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
@@ -1304,6 +1305,61 @@
         }
     }
 
+    /**
+     * Check that a submitted topology only claims blobs that are topology dependencies, and that every one of them
+     * exists. The dependency lists of a submitted topology are filled in by the client, so without this a submission
+     * could name any blob at all, for example another topology's code or configuration blob, and nimbus would delete
+     * it as its own dependency once the submitted topology is cleaned up. A key that exists nowhere is just as
+     * damaging: on gaining leadership a nimbus compares the dependencies of every active topology against its
+     * blobstore and gives up leadership when one is missing, so a single unresolvable key on a single active topology
+     * leaves the cluster without a leader for as long as that topology is active.
+     *
+     * <p>Existence is probed with {@code getBlobMeta} as the submitter, exactly as
+     * {@link Utils#validateTopologyBlobStoreMap(Map, BlobStore)} probes the blobs of
+     * {@link Config#TOPOLOGY_BLOBSTORE_MAP}, so a submitter can only claim a dependency it is allowed to read. The
+     * client uploads the dependency blobs before it submits, so they are present by the time this runs.
+     *
+     * @param topology  the submitted topology
+     * @param blobStore the blobstore to look the keys up in
+     * @param subject   the subject to look the keys up as, i.e. the submitter
+     * @throws InvalidTopologyException if a dependency list holds something that is not a dependency blob key, or
+     *                                  names a dependency blob that does not exist
+     * @throws AuthorizationException   if the submitter may not read one of the dependency blobs it named
+     */
+    @VisibleForTesting
+    static void validateDependencyBlobKeys(StormTopology topology, BlobStore blobStore, Subject subject)
+        throws InvalidTopologyException, AuthorizationException {
+        Set<String> checked = new HashSet<>();
+        validateDependencyBlobKeys(topology.get_dependency_jars(), "dependency_jars", blobStore, subject, checked);
+        validateDependencyBlobKeys(topology.get_dependency_artifacts(), "dependency_artifacts", blobStore, subject, checked);
+    }
+
+    private static void validateDependencyBlobKeys(List<String> keys, String fieldName, BlobStore blobStore, Subject subject,
+                                                   Set<String> checked) throws InvalidTopologyException, AuthorizationException {
+        if (keys == null) {
+            return;
+        }
+        for (String key : keys) {
+            if (!DependencyBlobStoreUtils.isDependencyBlobKey(key)) {
+                throw new WrappedInvalidTopologyException("Topology " + fieldName + " lists [" + key
+                    + "], which is not a dependency blob key; every entry must start with \""
+                    + DependencyBlobStoreUtils.BLOB_DEPENDENCIES_PREFIX + "\"");
+            }
+            if (!checked.add(key)) {
+                // the same dependency may be listed twice, one lookup for it is enough
+                continue;
+            }
+            try {
+                blobStore.getBlobMeta(key, subject);
+            } catch (KeyNotFoundException keyNotFound) {
+                throw new WrappedInvalidTopologyException("Topology " + fieldName + " lists [" + key
+                    + "], which is not in the blobstore; upload the dependency before submitting the topology, and if it "
+                    + "was uploaded earlier note that a dependency blob is deleted once no topology uses it any more, so "
+                    + "it has to be uploaded again");
+            }
+        }
+    }
+
     private static StormTopology tryReadTopology(String topoId, TopoCache tc)
         throws NotAliveException, AuthorizationException, IOException {
         try {
@@ -2945,7 +3001,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyJarsInTopology(topoId);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3333,6 +3389,7 @@
                 throw new WrappedInvalidTopologyException(ex.getMessage());
             }
             validator.validate(topoName, topoConf, topology);
+            validateDependencyBlobKeys(topology, blobStore, getSubject());
             if ((boolean) conf.getOrDefault(Config.DISABLE_SYMLINKS, false)) {
                 @SuppressWarnings("unchecked")
                 Map<String, Object> blobMap = (Map<String, Object>) topoConf.get(Config.TOPOLOGY_BLOBSTORE_MAP);
@@ -3458,8 +3515,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;
@@ -4883,8 +4940,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

@@ -40,6 +40,7 @@
 import org.apache.storm.blobstore.KeySequenceNumber;
 import org.apache.storm.blobstore.LocalFsBlobStore;
 import org.apache.storm.cluster.IStormClusterState;
+import org.apache.storm.dependency.DependencyBlobStoreUtils;
 import org.apache.storm.generated.AuthorizationException;
 import org.apache.storm.generated.Credentials;
 import org.apache.storm.generated.InvalidTopologyException;
@@ -49,6 +50,8 @@
 import org.apache.storm.generated.ReadableBlobMeta;
 import org.apache.storm.generated.SettableBlobMeta;
 import org.apache.storm.generated.StormTopology;
+import org.apache.storm.generated.SubmitOptions;
+import org.apache.storm.generated.TopologyInitialStatus;
 import org.apache.storm.metric.StormMetricsRegistry;
 import org.apache.storm.nimbus.ILeaderElector;
 import org.apache.storm.nimbus.NimbusInfo;
@@ -83,6 +86,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.mockito.ArgumentMatchers.any;
@@ -92,6 +96,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -464,4 +469,146 @@ private static void setCaller(String user) {
         subject.getPrincipals().add(new SingleUserPrincipal(user));
         ReqContext.context().setSubject(subject);
     }
+
+    @Test
+    void testValidateDependencyBlobKeysRejectsKeysThatAreNotDependencies() throws Exception {
+        // a topology fills its own dependency lists in on the client side, so nimbus has to check that they only
+        // name dependency blobs before it takes ownership of them and deletes them during cleanup
+        String victimJarKey = ConfigUtils.masterStormJarKey("victim-1-1234567890");
+        String victimConfKey = ConfigUtils.masterStormConfKey("victim-1-1234567890");
+        Subject submitter = new Subject();
+
+        StormTopology jarField = new StormTopology();
+        jarField.set_dependency_jars(List.of(victimJarKey));
+        InvalidTopologyException jarException = assertThrows(InvalidTopologyException.class,
+            () -> Nimbus.validateDependencyBlobKeys(jarField, localBlobStore, submitter));
+        assertTrue(jarException.get_msg().contains(victimJarKey), jarException.get_msg());
+        assertTrue(jarException.get_msg().contains("dependency_jars"), jarException.get_msg());
+
+        StormTopology artifactField = new StormTopology();
+        artifactField.set_dependency_artifacts(List.of(victimConfKey));
+        InvalidTopologyException artifactException = assertThrows(InvalidTopologyException.class,
+            () -> Nimbus.validateDependencyBlobKeys(artifactField, localBlobStore, submitter));
+        assertTrue(artifactException.get_msg().contains(victimConfKey), artifactException.get_msg());
+        assertTrue(artifactException.get_msg().contains("dependency_artifacts"), artifactException.get_msg());
+
+        // a good key followed by a bad one is caught too, and the message names the bad one
+        StormTopology mixed = new StormTopology();
+        mixed.set_dependency_jars(List.of(dependencyKey("a.jar"), victimJarKey));
+        InvalidTopologyException mixedException = assertThrows(InvalidTopologyException.class,
+            () -> Nimbus.validateDependencyBlobKeys(mixed, localBlobStore, submitter));
+        assertTrue(mixedException.get_msg().contains(victimJarKey), mixedException.get_msg());
+
+        // a key that is not a dependency key at all is rejected on its name, without asking the blobstore about it
+        verify(localBlobStore, never()).getBlobMeta(eq(victimJarKey), any());
+        verify(localBlobStore, never()).getBlobMeta(eq(victimConfKey), any());
+    }
+
+    @Test
+    void testValidateDependencyBlobKeysRejectsKeyThatIsNotInTheBlobStore() throws Exception {
+        // a key that merely looks like a dependency key is just as damaging: on gaining leadership a nimbus gives up
+        // leadership again when an active topology names a dependency it cannot find, so an unresolvable key leaves
+        // the cluster without a leader
+        String presentKey = dependencyKey("present.jar");
+        String missingKey = dependencyKey("missing.jar");
+        Subject submitter = new Subject();
+        when(localBlobStore.getBlobMeta(eq(missingKey), any())).thenThrow(new KeyNotFoundException(missingKey));
+
+        StormTopology jarField = new StormTopology();
+        jarField.set_dependency_jars(List.of(presentKey, missingKey));
+        InvalidTopologyException jarException = assertThrows(InvalidTopologyException.class,
+            () -> Nimbus.validateDependencyBlobKeys(jarField, localBlobStore, submitter));
+        assertTrue(jarException.get_msg().contains(missingKey), jarException.get_msg());
+        assertTrue(jarException.get_msg().contains("dependency_jars"), jarException.get_msg());
+        assertTrue(jarException.get_msg().contains("not in the blobstore"), jarException.get_msg());
+
+        StormTopology artifactField = new StormTopology();
+        artifactField.set_dependency_artifacts(List.of(missingKey));
+        InvalidTopologyException artifactException = assertThrows(InvalidTopologyException.class,
+            () -> Nimbus.validateDependencyBlobKeys(artifactField, localBlobStore, submitter));
+        assertTrue(artifactException.get_msg().contains(missingKey), artifactException.get_msg());
+        assertTrue(artifactException.get_msg().contains("dependency_artifacts"), artifactException.get_msg());
+    }
+
+    @Test
+    void testValidateDependencyBlobKeysLooksBlobsUpAsTheSubmitter() throws Exception {
+        // looking the blob up as the submitter, the way the TOPOLOGY_BLOBSTORE_MAP entries are looked up, also
+        // answers whether the submitter is allowed to read the dependency it claims; a dependency blob is uploaded
+        // with OTHER READ, so a legitimate submission passes
+        String key = dependencyKey("some-jar.jar");
+        Subject submitter = new Subject();
+
+        StormTopology topology = new StormTopology();
+        // listed under both fields to show that a key is looked up once no matter how often it is named
+        topology.set_dependency_jars(List.of(key, key));
+        topology.set_dependency_artifacts(List.of(key));
+        assertDoesNotThrow(() -> Nimbus.validateDependencyBlobKeys(topology, localBlobStore, submitter));
+
+        verify(localBlobStore, times(1)).getBlobMeta(key, submitter);
+    }
+
+    @Test
+    void testValidateDependencyBlobKeysAcceptsGeneratedKeysThatExist() throws Exception {
+        Subject submitter = new Subject();
+        StormTopology topology = new StormTopology();
+        topology.set_dependency_jars(List.of(dependencyKey("some-jar.jar"), dependencyKey("no-extension")));
+        topology.set_dependency_artifacts(List.of(dependencyKey("group-artifact-1.0.jar")));
+        assertDoesNotThrow(() -> Nimbus.validateDependencyBlobKeys(topology, localBlobStore, submitter));
+
+        // unset lists are how a topology submitted without dependencies looks
+        assertDoesNotThrow(() -> Nimbus.validateDependencyBlobKeys(new StormTopology(), localBlobStore, submitter));
+        verify(localBlobStore, never()).getBlobMeta(eq(null), any());
+    }
+
+    @Test
+    void testSubmitTopologyRejectsDependencyBlobKeyOfAnotherTopology() throws Exception {
+        Map<String, Object> conf = Map.of(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10);
+        Nimbus submitNimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo, localBlobStore, leaderElector,
+            groupMapper, new StormMetricsRegistry());
+        when(leaderElector.isLeader()).thenReturn(true);
+        when(stormClusterState.getTopoId(any())).thenReturn(Optional.empty());
+
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("wordSpout", new TestWordSpout(), 1);
+        StormTopology topology = builder.createTopology();
+        String victimJarKey = ConfigUtils.masterStormJarKey("victim-1-1234567890");
+        topology.set_dependency_artifacts(List.of(victimJarKey));
+
+        InvalidTopologyException exception = assertThrows(InvalidTopologyException.class,
+            () -> submitNimbus.submitTopologyWithOpts("thief", "/dev/null", "{}", topology,
+                new SubmitOptions(TopologyInitialStatus.ACTIVE)));
+        assertTrue(exception.get_msg().contains(victimJarKey), exception.get_msg());
+
+        // the submission was rejected before anything was stored for it
+        verify(stormClusterState, never()).setupHeatbeats(any(), any());
+    }
+
+    @Test
+    void testSubmitTopologyRejectsDependencyBlobKeyThatDoesNotExist() throws Exception {
+        Map<String, Object> conf = Map.of(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10);
+        Nimbus submitNimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo, localBlobStore, leaderElector,
+            groupMapper, new StormMetricsRegistry());
+        when(leaderElector.isLeader()).thenReturn(true);
+        when(stormClusterState.getTopoId(any())).thenReturn(Optional.empty());
+        String missingKey = dependencyKey("missing.jar");
+        when(localBlobStore.getBlobMeta(eq(missingKey), any())).thenThrow(new KeyNotFoundException(missingKey));
+
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("wordSpout", new TestWordSpout(), 1);
+        StormTopology topology = builder.createTopology();
+        topology.set_dependency_jars(List.of(missingKey));
+
+        InvalidTopologyException exception = assertThrows(InvalidTopologyException.class,
+            () -> submitNimbus.submitTopologyWithOpts("ghost", "/dev/null", "{}", topology,
+                new SubmitOptions(TopologyInitialStatus.ACTIVE)));
+        assertTrue(exception.get_msg().contains(missingKey), exception.get_msg());
+        assertTrue(exception.get_msg().contains("not in the blobstore"), exception.get_msg());
+
+        // the submission was rejected before anything was stored for it
+        verify(stormClusterState, never()).setupHeatbeats(any(), any());
+    }
+
+    private static String dependencyKey(String fileName) {
+        return DependencyBlobStoreUtils.generateDependencyBlobKey(DependencyBlobStoreUtils.applyUUIDToFileName(fileName));
+    }
 }