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 5ebe6e717 Check the calling principal against the topology owner in
uploadNewCredentials (#9001)
5ebe6e717 is described below
commit 5ebe6e717698472c703235853767b179817aafe6
Author: Richard Zowalla <[email protected]>
AuthorDate: Mon Aug 24 10:52:36 2026 +0200
Check the calling principal against the topology owner in
uploadNewCredentials (#9001)
---
.../org/apache/storm/daemon/nimbus/Nimbus.java | 28 +++++----
.../org/apache/storm/daemon/nimbus/NimbusTest.java | 66 +++++++++++++++++++++-
2 files changed, 80 insertions(+), 14 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 db2957804..682db2338 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
@@ -3862,22 +3862,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 49baebbd2..f24468d4c 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
@@ -31,8 +31,7 @@ import java.util.Optional;
import java.util.Set;
import javax.security.auth.Subject;
-import javax.security.auth.Subject;
-
+import com.codahale.metrics.Meter;
import net.minidev.json.JSONValue;
import org.apache.commons.io.FileUtils;
import org.apache.storm.Config;
@@ -42,6 +41,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.ListBlobsResult;
@@ -71,6 +71,7 @@ import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.apache.storm.utils.WrappedAuthorizationException;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -85,6 +86,7 @@ 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.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -95,6 +97,7 @@ 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 = "topology1-1-1";
@Mock
@@ -111,6 +114,8 @@ class NimbusTest {
private ILeaderElector leaderElector;
@Mock
private IGroupMappingServiceProvider groupMapper;
+ @Mock
+ private TopoCache topoCache;
private Nimbus nimbus;
@@ -122,6 +127,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,
@@ -272,6 +282,58 @@ class NimbusTest {
verify(localBlobStore, never()).listKeys();
}
+ @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);
+ }
+
@Test
void testValidateUploadedJarLocationRejectsLocationsOutsideTheInbox()
throws Exception {
Path inbox = Files.createTempDirectory("nimbus-inbox");
