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 053570037 Validate topology.blobstore.map in rebalance conf overrides
against the calling user (#8999)
053570037 is described below
commit 0535700370609cc5e38c50810dda20a6d4ac7eb2
Author: Richard Zowalla <[email protected]>
AuthorDate: Sun Aug 23 19:03:17 2026 +0200
Validate topology.blobstore.map in rebalance conf overrides against the
calling user (#8999)
---
.../org/apache/storm/daemon/nimbus/Nimbus.java | 3 ++
.../org/apache/storm/daemon/nimbus/NimbusTest.java | 46 ++++++++++++++++++++++
2 files changed, 49 insertions(+)
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 ecf7dd2a2..2d7704aaa 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
@@ -3592,6 +3592,9 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
topoConfigOverrides.remove(Config.TOPOLOGY_CLASSPATH_BEGINNING);
}
topoConfigOverrides.remove(Config.STORM_LOCAL_HOSTNAME);
+ //Blobs referenced by the overrides have to be readable by the
one asking for the rebalance,
+ // just like at submit time.
+ Utils.validateTopologyBlobStoreMap(topoConfigOverrides,
blobStore);
options.set_topology_conf_overrides(JSONValue.toJSONString(topoConfigOverrides));
}
Subject subject = getSubject();
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 2380d49a8..96336f7cc 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
@@ -24,8 +24,11 @@ import java.nio.file.Paths;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import javax.security.auth.Subject;
+import net.minidev.json.JSONValue;
import org.apache.commons.io.FileUtils;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
@@ -36,6 +39,7 @@ import org.apache.storm.cluster.IStormClusterState;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.KeyNotFoundException;
+import org.apache.storm.generated.RebalanceOptions;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.nimbus.ILeaderElector;
@@ -46,6 +50,8 @@ import
org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResource
import
org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategyOld;
import
org.apache.storm.scheduler.resource.strategies.scheduling.RoundRobinResourceAwareStrategy;
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.testing.TestWordSpout;
import org.apache.storm.thrift.TException;
import org.apache.storm.topology.TopologyBuilder;
@@ -53,6 +59,7 @@ import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.Time;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
import org.mockito.MockitoAnnotations;
@@ -62,6 +69,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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
@@ -239,4 +247,42 @@ class NimbusTest {
FileUtils.deleteQuietly(sibling.toFile());
}
}
+
+ @Test
+ void testRebalanceRejectsConfOverridesWithBlobsTheCallerCannotRead()
throws Exception {
+ final String topoName = "topo-with-blobs";
+ final String topoId = "topo-with-blobs-1-1234";
+ final String blobKey = "someone-elses-blob";
+
+ TopoCache topoCache = mock(TopoCache.class);
+ Map<String, Object> conf =
Map.of(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10);
+ nimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo,
localBlobStore, topoCache, leaderElector, groupMapper,
+ new StormMetricsRegistry());
+
+ StormTopology topology = new StormTopology();
+ topology.set_spouts(new HashMap<>());
+ topology.set_bolts(new HashMap<>());
+ topology.set_state_spouts(new HashMap<>());
+
when(stormClusterState.getTopoId(topoName)).thenReturn(Optional.of(topoId));
+ when(topoCache.readTopoConf(eq(topoId), any())).thenReturn(new
HashMap<>(Map.of(Config.TOPOLOGY_NAME, topoName)));
+ when(topoCache.readTopology(eq(topoId), any())).thenReturn(topology);
+ doThrow(new AuthorizationException("does not have READ access to " +
blobKey))
+ .when(localBlobStore).getBlobMeta(eq(blobKey), any());
+
+ RebalanceOptions options = new RebalanceOptions();
+ options.set_topology_conf_overrides(
+ JSONValue.toJSONString(Map.of(Config.TOPOLOGY_BLOBSTORE_MAP,
Map.of(blobKey, new HashMap<>()))));
+
+ Subject caller = new Subject(false, Set.of(new
SingleUserPrincipal("alice")), Set.of(), Set.of());
+ ReqContext.context().setSubject(caller);
+ try {
+ ArgumentCaptor<Subject> subjectCaptor =
ArgumentCaptor.forClass(Subject.class);
+ assertThrows(AuthorizationException.class, () ->
nimbus.rebalance(topoName, options));
+ verify(localBlobStore).getBlobMeta(eq(blobKey),
subjectCaptor.capture());
+ //the blobs are looked up as the one asking for the rebalance, not
as nimbus
+ assertSame(caller, subjectCaptor.getValue());
+ } finally {
+ ReqContext.reset();
+ }
+ }
}
