Filter the topology history by the authenticated caller rather than the requested user name

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",

storm-client/test/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizerTest.java

@@ -83,6 +83,11 @@ public void SimpleACLUserAuthTest() {
         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<>()));

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -1103,7 +1103,7 @@
         cleanable.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(store.storedTopoIds(), EMPTY_STRING_SET));
         cleanable.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
         Set<String> delayedCleanable = getExpiredTopologyIds(cleanable, conf);
         delayedCleanable.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
@@ -1223,8 +1223,8 @@
         ret.put(Config.TOPOLOGY_WORKER_NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
         ret.put(Config.NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
 
         if (!mergedConf.containsKey(Config.TOPOLOGY_METRICS_REPORTERS) && mergedConf.containsKey(Config.STORM_METRICS_REPORTERS)) {
             ret.put(Config.TOPOLOGY_METRICS_REPORTERS, mergedConf.get(Config.STORM_METRICS_REPORTERS));
         }
 
         // add any system metrics reporters to the topology metrics reporters
@@ -2945,7 +2945,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyJarsInTopology(topoId);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3008,6 +3008,44 @@
         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();
@@ -3420,8 +3458,8 @@
                 waitForDesiredCodeReplication(totalConf, topoId);
                 state.setupHeatbeats(topoId, topoConf);
                 state.setupErrors(topoId, topoConf);
                 if (ObjectReader.getBoolean(totalConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false)) {
                     state.setupBackpressure(topoId, topoConf);
                 }
                 notifyTopologyActionListener(topoName, "submitTopology");
                 TopologyStatus status = null;
@@ -4841,8 +4879,8 @@
             String topoName = (String) checkConf.get(Config.TOPOLOGY_NAME);
             checkAuthorization(topoName, checkConf, "getTopologyConf");
             Map<String, Object> maskedConf = new HashMap<>(ConfigUtils.maskPasswords(topoConf));
             if (maskedConf.get(BlowfishTupleSerializer.SECRET_KEY) instanceof String) {
                 maskedConf.put(BlowfishTupleSerializer.SECRET_KEY, "*****");
             }
             return JSONValue.toJSONString(maskedConf);
         } catch (Exception e) {
@@ -4894,25 +4932,27 @@
     @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);

storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java

@@ -21,6 +21,8 @@
 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.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.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.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.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 @@
 
 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 @@ void testRebalanceRejectsConfOverridesWithBlobsTheCallerCannotRead() throws Exce
             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);
+    }
 }