This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch fix/rebalance-blobstore-map-validation in repository https://gitbox.apache.org/repos/asf/storm.git
commit e597fd90ceff3e41e1d4191fbd9e1461abf9403b Author: Richard Zowalla <[email protected]> AuthorDate: Sat Aug 22 20:05:11 2026 +0200 Validate topology.blobstore.map in rebalance conf overrides against the calling user --- .../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(); + } + } }
