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 9a6490a0f Filter the topology history by the authenticated caller
rather than the requested user name (#9003)
9a6490a0f is described below
commit 9a6490a0f22300ce268efacbe6a9b02d9b15dc0e
Author: Richard Zowalla <[email protected]>
AuthorDate: Mon Aug 24 10:06:23 2026 +0200
Filter the topology history by the authenticated caller rather than the
requested user name (#9003)
---
.../auth/authorizer/SimpleACLAuthorizer.java | 1 +
.../auth/authorizer/SimpleACLAuthorizerTest.java | 5 ++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 52 ++++++++++++++--
.../org/apache/storm/daemon/nimbus/NimbusTest.java | 69 ++++++++++++++++++++++
4 files changed, 121 insertions(+), 6 deletions(-)
diff --git
a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
index d95965eaa..baaba221d 100644
---
a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
+++
b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
@@ -43,6 +43,7 @@ public class SimpleACLAuthorizer implements IAuthorizer {
"getNimbusConf",
"listBlobs",
"getClusterInfo",
+ "getTopologyHistory",
"getLeader",
"isTopologyNameAllowed",
"getTopologySummaries",
diff --git
a/storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java
b/storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java
index b2c05c385..7d67b034c 100644
---
a/storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java
+++
b/storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java
@@ -83,6 +83,11 @@ public class SimpleACLAuthorizerTest {
assertTrue(authorizer.permit(new ReqContext(userA), "getClusterInfo",
new HashMap<>()));
assertTrue(authorizer.permit(new ReqContext(userB), "getClusterInfo",
new HashMap<>()));
+ assertTrue(authorizer.permit(new ReqContext(adminUser),
"getTopologyHistory", new HashMap<>()));
+ assertFalse(authorizer.permit(new ReqContext(supervisorUser),
"getTopologyHistory", new HashMap<>()));
+ assertTrue(authorizer.permit(new ReqContext(userA),
"getTopologyHistory", new HashMap<>()));
+ assertTrue(authorizer.permit(new ReqContext(userB),
"getTopologyHistory", new HashMap<>()));
+
assertTrue(authorizer.permit(new ReqContext(adminUser),
"getSupervisorPageInfo", new HashMap<>()));
assertFalse(authorizer.permit(new ReqContext(supervisorUser),
"getSupervisorPageInfo", new HashMap<>()));
assertTrue(authorizer.permit(new ReqContext(userA),
"getSupervisorPageInfo", new HashMap<>()));
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 3064c6ebe..db2957804 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
@@ -3008,6 +3008,44 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
return !userGroups.isEmpty();
}
+ /**
+ * Get the user whose topology history is to be returned.
+ *
+ * <p>The history is filtered for the caller authenticated on this
request, not for the user named in the RPC
+ * argument. Only an admin (the ui daemon is expected to be one, see
SECURITY.md) may ask for the history of
+ * somebody else, because it serves the endpoint on behalf of its own
authenticated web users.
+ *
+ * @param user the user asked for by the caller
+ * @param adminUsers the configured admin users
+ * @param adminGroups the configured admin groups
+ * @return the user to filter the history with, the argument unchanged if
security is off
+ *
+ * @throws AuthorizationException if a non admin caller asked for somebody
else's history
+ * @throws IOException on any error while looking up the caller's groups
+ */
+ private String topologyHistoryUser(String user, Collection<String>
adminUsers,
+ Collection<String> adminGroups) throws
AuthorizationException, IOException {
+ Principal principal = ReqContext.context().principal();
+ if (principal == null) {
+ //security is off, there is no caller to filter by
+ return user;
+ }
+ String callerPrincipal = principal.getName();
+ String callerUser = principalToLocal.toLocal(principal);
+ if (adminUsers.contains(callerPrincipal) ||
adminUsers.contains(callerUser) || isUserPartOf(callerUser, adminGroups)) {
+ return user;
+ }
+ if (user != null && !user.equals(callerPrincipal) &&
!user.equals(callerUser)) {
+ //Only an admin may ask for somebody else's history. Fall back to
the caller's own
+ //rather than failing the call: a UI that is not in nimbus.admins
asks on behalf of
+ //its web users, and answering with the caller's history keeps
that page working.
+ LOG.warn("{} is not an admin and asked for the topology history of
{}, returning its own history instead. "
+ + "Add {} to {} if it should be able to read the history of
other users.",
+ callerUser, user, callerPrincipal, Config.NIMBUS_ADMINS);
+ }
+ return callerUser;
+ }
+
private List<String> readTopologyHistory(String user, Collection<String>
adminUsers) throws IOException {
LocalState state = topologyHistoryState;
List<LSTopoHistory> topoHistoryList = state.getTopoHistoryList();
@@ -4894,25 +4932,27 @@ public class Nimbus implements Iface, Shutdownable,
DaemonCommon {
@Override
public TopologyHistoryInfo getTopologyHistory(String user) throws
AuthorizationException, TException {
try {
+ checkAuthorization(null, null, "getTopologyHistory");
List<String> adminUsers = (List<String>)
conf.getOrDefault(Config.NIMBUS_ADMINS, Collections.emptyList());
List<String> adminGroups = (List<String>)
conf.getOrDefault(Config.NIMBUS_ADMINS_GROUPS, Collections.emptyList());
+ String historyUser = topologyHistoryUser(user, adminUsers,
adminGroups);
IStormClusterState state = stormClusterState;
List<String> assignedIds = state.assignments(null);
Set<String> ret = new HashSet<>();
- boolean isAdmin = adminUsers.contains(user);
+ boolean isAdmin = adminUsers.contains(historyUser);
for (String topoId : assignedIds) {
Map<String, Object> topoConf = tryReadTopoConf(topoId,
topoCache);
topoConf = Utils.merge(conf, topoConf);
List<String> groups =
ServerConfigUtils.getTopoLogsGroups(topoConf);
List<String> topoLogUsers =
ServerConfigUtils.getTopoLogsUsers(topoConf);
- if (user == null || isAdmin
- || isUserPartOf(user, groups)
- || isUserPartOf(user, adminGroups)
- || topoLogUsers.contains(user)) {
+ if (historyUser == null || isAdmin
+ || isUserPartOf(historyUser, groups)
+ || isUserPartOf(historyUser, adminGroups)
+ || topoLogUsers.contains(historyUser)) {
ret.add(topoId);
}
}
- ret.addAll(readTopologyHistory(user, adminUsers));
+ ret.addAll(readTopologyHistory(historyUser, adminUsers));
return new TopologyHistoryInfo(new ArrayList<>(ret));
} catch (Exception e) {
LOG.warn("Get topology history. (user='{}')", user, e);
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 a6c0771f1..49baebbd2 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
@@ -21,6 +21,8 @@ package org.apache.storm.daemon.nimbus;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -29,6 +31,8 @@ import java.util.Optional;
import java.util.Set;
import javax.security.auth.Subject;
+import javax.security.auth.Subject;
+
import net.minidev.json.JSONValue;
import org.apache.commons.io.FileUtils;
import org.apache.storm.Config;
@@ -42,6 +46,8 @@ import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.KeyNotFoundException;
import org.apache.storm.generated.ListBlobsResult;
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.metric.StormMetricsRegistry;
import org.apache.storm.nimbus.ILeaderElector;
@@ -51,6 +57,7 @@ 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.IAuthorizer;
import org.apache.storm.security.auth.IGroupMappingServiceProvider;
import org.apache.storm.security.auth.ReqContext;
@@ -59,8 +66,10 @@ import
org.apache.storm.security.auth.authorizer.DenyAuthorizer;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.thrift.TException;
import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.utils.ConfigUtils;
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.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -86,6 +95,7 @@ import static org.mockito.Mockito.when;
class NimbusTest {
private static final String BLOB_FILE_KEY = "file-key";
+ private static final String TOPO_ID = "topology1-1-1";
@Mock
private StormMetricsRegistry metricRegistry;
@@ -333,4 +343,63 @@ class NimbusTest {
ReqContext.reset();
}
}
+
+ @Test
+ void testGetTopologyHistoryFiltersByTheAuthenticatedCaller() throws
Exception {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10);
+ conf.put(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN,
DefaultPrincipalToLocal.class.getName());
+ conf.put(Config.NIMBUS_ADMINS, Collections.singletonList("admin"));
+ nimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo,
localBlobStore, leaderElector, groupMapper, metricRegistry);
+
+ Map<String, Object> topoConf = new HashMap<>();
+ topoConf.put(Config.TOPOLOGY_NAME, "topology1");
+ topoConf.put(Config.TOPOLOGY_USERS,
Collections.singletonList("alice"));
+
when(stormClusterState.assignments(null)).thenReturn(Collections.singletonList(TOPO_ID));
+
when(localBlobStore.readBlob(eq(ConfigUtils.masterStormConfKey(TOPO_ID)),
any()))
+ .thenReturn(Utils.toCompressedJsonConf(topoConf));
+
when(localBlobStore.getBlobMeta(eq(ConfigUtils.masterStormConfKey(TOPO_ID)),
any()))
+ .thenReturn(new ReadableBlobMeta(new SettableBlobMeta(new
ArrayList<>()), 0));
+
+ try {
+ setCaller("bob");
+ // asking for somebody else's history is only for admins, the ui
daemon is expected to be one.
+ // a caller that is not an admin gets its own history back rather
than an error, so a ui that
+ // was left out of nimbus.admins keeps serving the page instead of
failing it
+
assertTrue(nimbus.getTopologyHistory("alice").get_topo_ids().isEmpty());
+ // and no user argument at all is the caller's own history, not
everybody's
+
assertTrue(nimbus.getTopologyHistory(null).get_topo_ids().isEmpty());
+
+ setCaller("alice");
+ assertEquals(Collections.singletonList(TOPO_ID),
nimbus.getTopologyHistory(null).get_topo_ids());
+
+ setCaller("admin");
+ assertEquals(Collections.singletonList(TOPO_ID),
nimbus.getTopologyHistory("alice").get_topo_ids());
+
assertTrue(nimbus.getTopologyHistory("bob").get_topo_ids().isEmpty());
+ } finally {
+ ReqContext.reset();
+ }
+ }
+
+ @Test
+ void testGetTopologyHistoryIsAuthorized() throws Exception {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 10);
+ conf.put(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN,
DefaultPrincipalToLocal.class.getName());
+ conf.put(DaemonConfig.NIMBUS_AUTHORIZER,
DenyAuthorizer.class.getName());
+ nimbus = new Nimbus(conf, iNimbus, stormClusterState, nimbusInfo,
localBlobStore, leaderElector, groupMapper, metricRegistry);
+
+ try {
+ setCaller("bob");
+ assertThrows(AuthorizationException.class, () ->
nimbus.getTopologyHistory("bob"));
+ } finally {
+ ReqContext.reset();
+ }
+ }
+
+ private static void setCaller(String user) {
+ Subject subject = new Subject();
+ subject.getPrincipals().add(new SingleUserPrincipal(user));
+ ReqContext.context().setSubject(subject);
+ }
}
