This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new c8d41cf60 Reject a topology submission that lists a blob key which is
not a dependency blob key (#9012)
c8d41cf60 is described below
commit c8d41cf607d256bb39eaa636b24801232c58fb67
Author: Richard Zowalla <[email protected]>
AuthorDate: Mon Aug 24 12:57:21 2026 +0200
Reject a topology submission that lists a blob key which is not a
dependency blob key (#9012)
---
.../storm/dependency/DependencyBlobStoreUtils.java | 18 ++-
.../org/apache/storm/daemon/nimbus/Nimbus.java | 57 ++++++++
.../org/apache/storm/daemon/nimbus/NimbusTest.java | 146 +++++++++++++++++++++
3 files changed, 220 insertions(+), 1 deletion(-)
diff --git
a/storm-client/src/jvm/org/apache/storm/dependency/DependencyBlobStoreUtils.java
b/storm-client/src/jvm/org/apache/storm/dependency/DependencyBlobStoreUtils.java
index 5211e5570..f692323b9 100644
---
a/storm-client/src/jvm/org/apache/storm/dependency/DependencyBlobStoreUtils.java
+++
b/storm-client/src/jvm/org/apache/storm/dependency/DependencyBlobStoreUtils.java
@@ -24,12 +24,28 @@ import
org.apache.storm.shade.org.apache.commons.lang3.StringUtils;
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);
diff --git
a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index ca45abc24..4469f8dfa 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -88,6 +88,7 @@ import org.apache.storm.daemon.DaemonCommon;
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;
@@ -1308,6 +1309,61 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
}
}
+ /**
+ * 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 {
@@ -3440,6 +3496,7 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
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);
diff --git
a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
index 92d1a9ecf..e27b5b4ec 100644
--- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
+++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
@@ -52,6 +52,8 @@ import org.apache.storm.generated.RebalanceOptions;
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;
@@ -87,6 +89,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
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;
@@ -97,6 +100,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -620,4 +624,146 @@ class NimbusTest {
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));
+ }
}
