Return only the log files the requesting user may access from the log file listing

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

@@ -114,7 +114,7 @@ public LogviewerLogPageHandler(String logRoot, String daemonLogRoot,
     }
 
     /**
-     * Enumerate worker log files for given criteria.
+     * Enumerate worker log files for given criteria. Only the files the user is allowed to access are returned.
      *
      * @param user username
      * @param port worker's port, null for all workers
@@ -175,8 +175,10 @@ public Response listLogFiles(String user, Integer port, String topologyId, Strin
 
         List<String> files;
         if (fileResults != null) {
+            Map<String, Boolean> authorizedPortDirs = new HashMap<>();
             files = fileResults.stream()
                     .map(WorkerLogs::getTopologyPortWorkerLog)
+                    .filter(fileStr -> isUserAllowedToAccessLog(user, fileStr, authorizedPortDirs))
                     .sorted().collect(toList());
         } else {
             files = new ArrayList<>();
@@ -185,6 +187,19 @@ public Response listLogFiles(String user, Integer port, String topologyId, Strin
         return LogviewerResponseBuilder.buildSuccessJsonResponse(files, callback, origin);
     }
 
+    /**
+     * Check whether the user may access the given "topologyId/port/fileName" worker log. The authorization only depends on the
+     * topology and the port, so the answer is cached per port directory to avoid re-reading the log metadata for every file.
+     */
+    private boolean isUserAllowedToAccessLog(String user, String fileStr, Map<String, Boolean> authorizedPortDirs) {
+        Path portDir = Paths.get(fileStr).getParent();
+        if (portDir == null) {
+            return resourceAuthorizer.isUserAllowedToAccessFile(user, fileStr);
+        }
+        return authorizedPortDirs.computeIfAbsent(portDir.toString(),
+            key -> resourceAuthorizer.isUserAllowedToAccessFile(user, fileStr));
+    }
+
     /**
      * Provides a worker log file to view, starting from the specified position
      * or default starting position of the most recent page.

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

@@ -22,6 +22,12 @@
 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.ArgumentMatchers.startsWith;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -69,19 +75,22 @@ public void testListLogFiles() throws IOException {
                 new WorkerLogs(stormConf, Paths.get(rootPath), metricsRegistry), new ResourceAuthorizer(stormConf), metricsRegistry);
 
         final Response expectedAll = LogviewerResponseBuilder.buildSuccessJsonResponse(
-                List.of("topoA/port1/worker.log", "topoA/port2/worker.log", "topoB/port1/worker.log"),
+                List.of(String.join(File.separator, "topoA", "1111", "worker.log"),
+                        String.join(File.separator, "topoA", "2222", "worker.log"),
+                        String.join(File.separator, "topoB", "1111", "worker.log")),
                 null,
                 origin
         );
 
         final Response expectedFilterPort = LogviewerResponseBuilder.buildSuccessJsonResponse(
-                List.of("topoA/port1/worker.log", "topoB/port1/worker.log"),
+                List.of(String.join(File.separator, "topoA", "1111", "worker.log"),
+                        String.join(File.separator, "topoB", "1111", "worker.log")),
                 null,
                 origin
         );
 
         final Response expectedFilterTopoId = LogviewerResponseBuilder.buildSuccessJsonResponse(
-                List.of("topoB/port1/worker.log"),
+                List.of(String.join(File.separator, "topoB", "1111", "worker.log")),
                 null,
                 origin
         );
@@ -97,11 +106,48 @@ public void testListLogFiles() throws IOException {
         assertEqualsJsonResponse(expectedFilterTopoId, returnedFilterTopoId, List.class);
     }
 
+    /**
+     * list-log-files only returns the log files the user is allowed to access.
+     */
+    @Test
+    public void testListLogFilesFiltersFilesTheUserMayNotAccess() throws IOException {
+        String rootPath = Files.createTempDirectory("workers-artifacts").toFile().getCanonicalPath();
+        File file1 = new File(String.join(File.separator, rootPath, "topoA", "1111"), "worker.log");
+        File file2 = new File(String.join(File.separator, rootPath, "topoA", "1111"), "worker.log.1");
+        File file3 = new File(String.join(File.separator, rootPath, "topoB", "1111"), "worker.log");
+
+        file1.getParentFile().mkdirs();
+        file3.getParentFile().mkdirs();
+        file1.createNewFile();
+        file2.createNewFile();
+        file3.createNewFile();
+
+        String origin = "www.origin.server.net";
+        String topoAPortDir = String.join(File.separator, "topoA", "1111");
+        Map<String, Object> stormConf = Utils.readStormConfig();
+        StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+        ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+        when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(), startsWith(topoAPortDir))).thenReturn(true);
+        LogviewerLogPageHandler handler = new LogviewerLogPageHandler(rootPath, rootPath,
+                new WorkerLogs(stormConf, Paths.get(rootPath), metricsRegistry), resourceAuthorizer, metricsRegistry);
+
+        final Response returned = handler.listLogFiles("user", null, null, null, origin);
+
+        List<?> files = new ObjectMapper().readValue((String) returned.getEntity(), List.class);
+
+        Utils.forceDelete(rootPath);
+
+        assertEquals(List.of(String.join(File.separator, topoAPortDir, "worker.log"),
+                String.join(File.separator, topoAPortDir, "worker.log.1")), files);
+        //The authorization only depends on the port directory, so it is checked once per port directory, not once per file.
+        verify(resourceAuthorizer, times(2)).isUserAllowedToAccessFile(anyString(), anyString());
+    }
+
     private <T> void assertEqualsJsonResponse(Response expected, Response actual, Class<T> entityClass) throws IOException {
         ObjectMapper objectMapper = new ObjectMapper();
         T entityFromExpected = objectMapper.readValue((String) expected.getEntity(), entityClass);
-        T actualFromExpected = objectMapper.readValue((String) expected.getEntity(), entityClass);
-        assertEquals(entityFromExpected, actualFromExpected);
+        T entityFromActual = objectMapper.readValue((String) actual.getEntity(), entityClass);
+        assertEquals(entityFromExpected, entityFromActual);
 
         assertEquals(expected.getStatus(), actual.getStatus());
         assertTrue(expected.getHeaders().equalsIgnoreValueOrder(actual.getHeaders()));