This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch fix/nimbus-credentials-owner-check in repository https://gitbox.apache.org/repos/asf/storm.git
commit 23fdf7e1f336dbae86e722c7b586830c372b8964 Author: Richard Zowalla <[email protected]> AuthorDate: Sat Aug 22 20:06:53 2026 +0200 Check the calling principal against the topology owner in uploadNewCredentials --- .../org/apache/storm/daemon/nimbus/Nimbus.java | 28 ++++---- .../org/apache/storm/daemon/nimbus/NimbusTest.java | 76 ++++++++++++++++++++++ 2 files changed, 92 insertions(+), 12 deletions(-) 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..46e63cebf 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 @@ -3821,22 +3821,26 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon { checkAuthorization(topoName, topoConf, "uploadNewCredentials"); String realPrincipal = (String) topoConf.get(Config.TOPOLOGY_SUBMITTER_PRINCIPAL); String realUser = (String) topoConf.get(Config.TOPOLOGY_SUBMITTER_USER); - String expectedOwner = null; - if (credentials.is_set_topoOwner()) { - expectedOwner = credentials.get_topoOwner(); - } else { - Principal p = ReqContext.context().principal(); - if (p != null) { - expectedOwner = principalToLocal.toLocal(p); - } + String caller = null; + Principal p = ReqContext.context().principal(); + if (p != null) { + caller = principalToLocal.toLocal(p); } - // expectedOwner being null means that security is disabled (which why are we uploading credentials with security disabled??? - if (expectedOwner == null) { + // caller being null means that security is disabled (which why are we uploading credentials with security disabled??? + if (caller == null) { LOG.warn("Please check you settings. Credentials are being uploaded to {} with security disabled.", topoId); - } else if (!realPrincipal.equals(expectedOwner) && !realUser.equals(expectedOwner)) { - throw new AuthorizationException(topoId + " is expected to be owned by " + expectedOwner + } else if (!realPrincipal.equals(caller) && !realUser.equals(caller)) { + throw new AuthorizationException(topoId + " is expected to be owned by " + caller + " but is actually owned by " + realPrincipal); } + // topoOwner is just the owner the client expects, so it can only reject a mismatch, never stand in for the caller. + if (credentials.is_set_topoOwner()) { + String expectedOwner = credentials.get_topoOwner(); + if (!expectedOwner.equals(realPrincipal) && !expectedOwner.equals(realUser)) { + throw new AuthorizationException(topoId + " is expected to be owned by " + expectedOwner + + " but is actually owned by " + realPrincipal); + } + } synchronized (credUpdateLock) { //Merge the old credentials so creds nimbus created are not lost. 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..7445b5d81 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 com.codahale.metrics.Meter; import org.apache.commons.io.FileUtils; import org.apache.storm.Config; import org.apache.storm.DaemonConfig; @@ -34,6 +37,7 @@ import org.apache.storm.blobstore.KeySequenceNumber; import org.apache.storm.blobstore.LocalFsBlobStore; import org.apache.storm.cluster.IStormClusterState; import org.apache.storm.generated.AuthorizationException; +import org.apache.storm.generated.Credentials; import org.apache.storm.generated.InvalidTopologyException; import org.apache.storm.generated.KeyNotFoundException; import org.apache.storm.generated.StormTopology; @@ -45,12 +49,16 @@ import org.apache.storm.scheduler.resource.strategies.priority.DefaultScheduling import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy; import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategyOld; import org.apache.storm.scheduler.resource.strategies.scheduling.RoundRobinResourceAwareStrategy; +import org.apache.storm.security.auth.DefaultPrincipalToLocal; 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; import org.apache.storm.utils.ServerUtils; import org.apache.storm.utils.Time; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mock; @@ -63,6 +71,7 @@ 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.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -73,6 +82,8 @@ import static org.mockito.Mockito.when; class NimbusTest { private static final String BLOB_FILE_KEY = "file-key"; + private static final String TOPO_NAME = "topo"; + private static final String TOPO_ID = "topo-id"; @Mock private StormMetricsRegistry metricRegistry; @@ -88,6 +99,8 @@ class NimbusTest { private ILeaderElector leaderElector; @Mock private IGroupMappingServiceProvider groupMapper; + @Mock + private TopoCache topoCache; private Nimbus nimbus; @@ -99,6 +112,11 @@ class NimbusTest { nimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo, localBlobStore, leaderElector, groupMapper, metricRegistry); } + @AfterEach + public void tearDown() { + ReqContext.reset(); + } + @Test public void testMemoryLoadLargerThanMaxHeapSize() { // Topology will not be able to be successfully scheduled: Config TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB=128.0 < 129.0, @@ -206,6 +224,64 @@ class NimbusTest { } } + @Test + void testUploadNewCredentialsRejectsACallerWhoIsNotTheOwner() throws Exception { + Nimbus nimbus = makeNimbusOwningTopology(TOPO_NAME, TOPO_ID, "alice"); + setCaller("bob"); + + // bob claims the topology is owned by alice, which it is, but bob is not alice + Credentials creds = new Credentials(Map.of("key", "value")); + creds.set_topoOwner("alice"); + + assertThrows(AuthorizationException.class, () -> nimbus.uploadNewCredentials(TOPO_NAME, creds)); + verify(stormClusterState, never()).setCredentials(eq(TOPO_ID), any(), any()); + } + + @Test + void testUploadNewCredentialsAcceptsTheOwner() throws Exception { + Nimbus nimbus = makeNimbusOwningTopology(TOPO_NAME, TOPO_ID, "alice"); + setCaller("alice"); + + Credentials creds = new Credentials(Map.of("key", "value")); + creds.set_topoOwner("alice"); + nimbus.uploadNewCredentials(TOPO_NAME, creds); + + verify(stormClusterState).setCredentials(eq(TOPO_ID), eq(creds), any()); + } + + @Test + void testUploadNewCredentialsRejectsAnOwnerMismatchClaimedByTheOwner() throws Exception { + Nimbus nimbus = makeNimbusOwningTopology(TOPO_NAME, TOPO_ID, "alice"); + setCaller("alice"); + + // alice expects the topology to be owned by bob, so the push must not happen + Credentials creds = new Credentials(Map.of("key", "value")); + creds.set_topoOwner("bob"); + + assertThrows(AuthorizationException.class, () -> nimbus.uploadNewCredentials(TOPO_NAME, creds)); + verify(stormClusterState, never()).setCredentials(eq(TOPO_ID), any(), any()); + } + + private Nimbus makeNimbusOwningTopology(String topoName, String topoId, String owner) throws Exception { + Map<String, Object> topoConf = new HashMap<>(); + topoConf.put(Config.TOPOLOGY_SUBMITTER_PRINCIPAL, owner); + topoConf.put(Config.TOPOLOGY_SUBMITTER_USER, owner); + when(stormClusterState.getTopoId(topoName)).thenReturn(Optional.of(topoId)); + when(topoCache.readTopoConf(eq(topoId), any())).thenReturn(topoConf); + when(metricRegistry.registerMeter(anyString())).thenReturn(new Meter()); + + Map<String, Object> conf = Map.of(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10, + Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN, DefaultPrincipalToLocal.class.getName()); + return new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo, localBlobStore, topoCache, leaderElector, groupMapper, + metricRegistry); + } + + private void setCaller(String user) { + Subject subject = new Subject(); + subject.getPrincipals().add(new SingleUserPrincipal(user)); + ReqContext.context().setSubject(subject); + } + @Test void testValidateUploadedJarLocationRejectsLocationsOutsideTheInbox() throws Exception { Path inbox = Files.createTempDirectory("nimbus-inbox");
