Apply the configured log user and group lists to daemon log requests

storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandler.java

@@ -298,6 +298,10 @@ public Response daemonLogPage(String fileName, Integer start, Integer length, St
             return LogviewerResponseBuilder.buildResponsePageNotFound();
         }
 
+        if (!resourceAuthorizer.isUserAllowedToAccessDaemonFile(user)) {
+            return LogviewerResponseBuilder.buildResponseUnauthorizedUser(user);
+        }
+
         if (file.toFile().exists()) {
             // all types of files included
             List<File> logFiles = Arrays.stream(daemonLogRoot.toFile().listFiles())

storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandler.java

@@ -165,7 +165,8 @@ public Response searchLogFile(String fileName, String user, boolean isDaemon, St
         }
         Response response;
         if (absFile.toFile().exists()) {
-            if (isDaemon || resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
+            if (isDaemon ? resourceAuthorizer.isUserAllowedToAccessDaemonFile(user)
+                    : resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
                 Integer numMatchesInt = numMatchesStr != null ? tryParseIntParam("num-matches", numMatchesStr) : null;
                 Integer offsetInt = offsetStr != null ? tryParseIntParam("start-byte-offset", offsetStr) : null;
 

storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/LogFileDownloader.java

@@ -78,7 +78,8 @@ public Response downloadFile(String host, String fileName, String user, boolean
         }
 
         if (file.toFile().exists()) {
-            if (isDaemon || resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
+            if (isDaemon ? resourceAuthorizer.isUserAllowedToAccessDaemonFile(user)
+                    : resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
                 fileDownloadSizeDistMb.update(Math.round((double) file.toFile().length() / FileUtils.ONE_MB));
                 String downloadedFileName;
                 Path pathRelativeToRootDir = rootDir.relativize(file);

storm-webapp/src/main/java/org/apache/storm/daemon/logviewer/utils/ResourceAuthorizer.java

@@ -72,6 +72,41 @@ public boolean isUserAllowedToAccessFile(String user, String fileName) {
         return !isLogviewerFilterConfigured() || isAuthorizedLogUser(user, fileName);
     }
 
+    /**
+     * Checks whether user is allowed to access a daemon log file via UI. Daemon logs have no owning topology, so only the
+     * cluster level lists are consulted. Always true when the Logviewer filter is not configured.
+     *
+     * @param user username
+     */
+    public boolean isUserAllowedToAccessDaemonFile(String user) {
+        return !isLogviewerFilterConfigured() || isAuthorizedDaemonLogUser(user);
+    }
+
+    /**
+     * Checks whether user is authorized to access daemon log files. Checks regardless of UI filter.
+     *
+     * @param user username
+     */
+    public boolean isAuthorizedDaemonLogUser(String user) {
+        if (StringUtils.isEmpty(user)) {
+            return false;
+        }
+
+        List<String> logsUsers = new ArrayList<>();
+        logsUsers.addAll(ObjectReader.getStrings(stormConf.get(DaemonConfig.LOGS_USERS)));
+        logsUsers.addAll(ObjectReader.getStrings(stormConf.get(Config.NIMBUS_ADMINS)));
+
+        List<String> logsGroups = new ArrayList<>();
+        logsGroups.addAll(ObjectReader.getStrings(stormConf.get(DaemonConfig.LOGS_GROUPS)));
+        logsGroups.addAll(ObjectReader.getStrings(stormConf.get(Config.NIMBUS_ADMINS_GROUPS)));
+
+        String userName = principalToLocal.toLocal(user);
+        Set<String> groups = getUserGroups(userName);
+
+        return logsUsers.stream().anyMatch(u -> u.equals(userName))
+            || Sets.intersection(groups, new HashSet<>(logsGroups)).size() > 0;
+    }
+
     /**
      * Checks whether user is authorized to access file. Checks regardless of UI filter.
      *

storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogDownloadHandlerTest.java

@@ -24,6 +24,9 @@
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import com.google.common.net.HttpHeaders;
 import java.io.IOException;
@@ -121,7 +124,28 @@ public void testDownloadDaemonLogFilePathOutsideLogRoot() throws IOException {
         }
     }
 
+    @Test
+    public void testDownloadDaemonLogFileUnauthorizedUser() throws IOException {
+        try (TmpPath rootPath = new TmpPath()) {
+
+            ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+            when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(false);
+            LogviewerLogDownloadHandler handler = createHandlerTraversalTests(rootPath.getFile().toPath(), resourceAuthorizer);
+
+            Response response = handler.downloadDaemonLogFile("host", "nimbus.log", "user");
+
+            Utils.forceDelete(rootPath.toString());
+
+            assertThat(response.getStatus(), is(Response.Status.FORBIDDEN.getStatusCode()));
+        }
+    }
+
     private LogviewerLogDownloadHandler createHandlerTraversalTests(Path rootPath) throws IOException {
+        return createHandlerTraversalTests(rootPath, new ResourceAuthorizer(Utils.readStormConfig()));
+    }
+
+    private LogviewerLogDownloadHandler createHandlerTraversalTests(Path rootPath, ResourceAuthorizer resourceAuthorizer)
+            throws IOException {
         Path daemonLogRoot = rootPath.resolve("logs");
         Path fileOutsideDaemonRoot = rootPath.resolve("evil.sh");
         Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
@@ -143,7 +167,7 @@ private LogviewerLogDownloadHandler createHandlerTraversalTests(Path rootPath) t
         Map<String, Object> stormConf = Utils.readStormConfig();
         StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
         return new LogviewerLogDownloadHandler(workerLogRoot.toString(), daemonLogRoot.toString(),
-            new WorkerLogs(stormConf, workerLogRoot, metricsRegistry), new ResourceAuthorizer(stormConf), metricsRegistry);
+            new WorkerLogs(stormConf, workerLogRoot, metricsRegistry), resourceAuthorizer, metricsRegistry);
     }
 
 }

storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogPageHandlerTest.java

@@ -22,6 +22,9 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -168,7 +171,29 @@ public void testDaemonLogPagePathIntoWorkerLogs() throws Exception {
         }
     }
 
+    @Test
+    public void testDaemonLogPageUnauthorizedUser() throws Exception {
+        try (TmpPath rootPath = new TmpPath()) {
+            ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+            when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(false);
+            LogviewerLogPageHandler handler = createHandlerForTraversalTests(rootPath.getFile().toPath(), resourceAuthorizer);
+            //Give the daemon log some content, so that an unauthorized request is the only reason not to render the page.
+            Files.writeString(rootPath.getFile().toPath().resolve("logs").resolve("nimbus.log"), "nimbus log content");
+
+            final Response returned = handler.daemonLogPage("nimbus.log", 0, 100, null, "user");
+
+            Utils.forceDelete(rootPath.toString());
+
+            assertThat(returned.getStatus(), is(Response.Status.FORBIDDEN.getStatusCode()));
+        }
+    }
+
     private LogviewerLogPageHandler createHandlerForTraversalTests(Path rootPath) throws IOException {
+        return createHandlerForTraversalTests(rootPath, new ResourceAuthorizer(Utils.readStormConfig()));
+    }
+
+    private LogviewerLogPageHandler createHandlerForTraversalTests(Path rootPath, ResourceAuthorizer resourceAuthorizer)
+            throws IOException {
         Path daemonLogRoot = rootPath.resolve("logs");
         Path fileOutsideDaemonRoot = rootPath.resolve("evil.sh");
         Path daemonFile = daemonLogRoot.resolve("nimbus.log");
@@ -190,6 +215,6 @@ private LogviewerLogPageHandler createHandlerForTraversalTests(Path rootPath) th
         Map<String, Object> stormConf = Utils.readStormConfig();
         StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
         return new LogviewerLogPageHandler(workerLogRoot.toString(), daemonLogRoot.toString(),
-            new WorkerLogs(stormConf, workerLogRoot, metricsRegistry), new ResourceAuthorizer(stormConf), metricsRegistry);
+            new WorkerLogs(stormConf, workerLogRoot, metricsRegistry), resourceAuthorizer, metricsRegistry);
     }
 }

storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/handler/LogviewerLogSearchHandlerTest.java

@@ -47,12 +47,15 @@
 import java.util.Map;
 import java.util.function.Function;
 
+import jakarta.ws.rs.core.Response;
+
 import org.apache.storm.DaemonConfig;
 import org.apache.storm.daemon.logviewer.LogviewerConstant;
 import org.apache.storm.daemon.logviewer.utils.ResourceAuthorizer;
 import org.apache.storm.daemon.ui.InvalidRequestException;
 import org.apache.storm.metric.StormMetricsRegistry;
 import org.apache.storm.streams.tuple.Tuple3;
+import org.apache.storm.testing.TmpPath;
 import org.apache.storm.utils.Utils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -863,6 +866,25 @@ private LogviewerLogSearchHandler getStubbedSearchHandler() {
         }
     }
 
+    @Test
+    public void testSearchDaemonLogFileUnauthorizedUser() throws Exception {
+        try (TmpPath rootPath = new TmpPath()) {
+            Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+            Files.createDirectories(daemonLogRoot);
+            Files.createFile(daemonLogRoot.resolve("nimbus.log"));
+
+            Map<String, Object> stormConf = Utils.readStormConfig();
+            ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+            when(resourceAuthorizer.isUserAllowedToAccessDaemonFile(anyString())).thenReturn(false);
+            LogviewerLogSearchHandler handler = new LogviewerLogSearchHandler(stormConf, Paths.get(""), daemonLogRoot,
+                resourceAuthorizer, new StormMetricsRegistry());
+
+            Response response = handler.searchLogFile("nimbus.log", "user", true, "needle", null, null, null, null);
+
+            assertEquals(403, response.getStatus());
+        }
+    }
+
     private static LogviewerLogSearchHandler getSearchHandler() {
         Map<String, Object> stormConf = Utils.readStormConfig();
         return new LogviewerLogSearchHandler(stormConf, Paths.get(""), Paths.get(""),

storm-webapp/src/test/java/org/apache/storm/daemon/logviewer/utils/ResourceAuthorizerTest.java

@@ -211,4 +211,44 @@ public void authorizationFailsWhenFilterConfigured() {
         authorized = authorizer.isUserAllowedToAccessFile("bob", "anyfile");
         assertFalse(authorized); // filter configured, should fail all users
     }
+
+    /**
+     * daemon logs are allowed for cluster logs users and cluster admins only.
+     */
+    @Test
+    public void testAuthorizedDaemonLogUserAllowsClusterLogsUserAndClusterAdmin() {
+        Map<String, Object> stormConf = Utils.readStormConfig();
+
+        Map<String, Object> conf = new HashMap<>(stormConf);
+        conf.put(LOGS_USERS, Collections.singletonList("alice"));
+        conf.put(NIMBUS_ADMINS, Collections.singletonList("bob"));
+
+        ResourceAuthorizer authorizer = spy(new ResourceAuthorizer(conf));
+
+        doReturn(Collections.emptySet()).when(authorizer).getUserGroups(anyString());
+
+        assertTrue(authorizer.isAuthorizedDaemonLogUser("alice"));
+        assertTrue(authorizer.isAuthorizedDaemonLogUser("bob"));
+        assertFalse(authorizer.isAuthorizedDaemonLogUser("mallory"));
+    }
+
+    /**
+     * daemon log access consults the cluster level lists once a filter is configured.
+     */
+    @Test
+    public void daemonLogAuthorizationFailsWhenFilterConfigured() {
+        Map<String, Object> stormConf = Utils.readStormConfig();
+        Map<String, Object> conf = new HashMap<>(stormConf);
+        conf.put(LOGS_USERS, Collections.singletonList("alice"));
+
+        ResourceAuthorizer authorizer = spy(new ResourceAuthorizer(conf));
+
+        doReturn(Collections.emptySet()).when(authorizer).getUserGroups(anyString());
+
+        assertTrue(authorizer.isUserAllowedToAccessDaemonFile("bob")); // no filter configured, allow anyone
+
+        conf.put(DaemonConfig.LOGVIEWER_FILTER, "someFilter");
+        assertTrue(authorizer.isUserAllowedToAccessDaemonFile("alice"));
+        assertFalse(authorizer.isUserAllowedToAccessDaemonFile("bob"));
+    }
 }