Change the worker log file permission only for download requests that are served (#8978)

Author: rzo1

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

@@ -29,7 +29,6 @@
 
 public class LogviewerLogDownloadHandler {
 
-    private WorkerLogs workerLogs;
     private final LogFileDownloader logFileDownloadHelper;
 
     /**
@@ -43,8 +42,7 @@ public class LogviewerLogDownloadHandler {
      */
     public LogviewerLogDownloadHandler(String logRoot, String daemonLogRoot, WorkerLogs workerLogs,
         ResourceAuthorizer resourceAuthorizer, StormMetricsRegistry metricsRegistry) {
-        this.workerLogs = workerLogs;
-        this.logFileDownloadHelper = new LogFileDownloader(logRoot, daemonLogRoot, resourceAuthorizer, metricsRegistry);
+        this.logFileDownloadHelper = new LogFileDownloader(logRoot, daemonLogRoot, workerLogs, resourceAuthorizer, metricsRegistry);
     }
 
     /**
@@ -57,7 +55,6 @@ public LogviewerLogDownloadHandler(String logRoot, String daemonLogRoot, WorkerL
      *
      */
     public Response downloadLogFile(String host, String fileName, String user) throws IOException {
-        workerLogs.setLogFilePermission(fileName);
         return logFileDownloadHelper.downloadFile(host, fileName, user, false);
     }
 

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

@@ -36,20 +36,23 @@ public class LogFileDownloader {
     private final Meter numFileDownloadExceptions;
     private final Path logRoot;
     private final Path daemonLogRoot;
+    private final WorkerLogs workerLogs;
     private final ResourceAuthorizer resourceAuthorizer;
 
     /**
      * Constructor.
      *
      * @param logRoot root worker log directory
      * @param daemonLogRoot root daemon log directory
+     * @param workerLogs {@link WorkerLogs}
      * @param resourceAuthorizer {@link ResourceAuthorizer}
      * @param metricsRegistry The logviewer metrics registry
      */
-    public LogFileDownloader(String logRoot, String daemonLogRoot, ResourceAuthorizer resourceAuthorizer,
-        StormMetricsRegistry metricsRegistry) {
+    public LogFileDownloader(String logRoot, String daemonLogRoot, WorkerLogs workerLogs,
+        ResourceAuthorizer resourceAuthorizer, StormMetricsRegistry metricsRegistry) {
         this.logRoot = Paths.get(logRoot).toAbsolutePath().normalize();
         this.daemonLogRoot = Paths.get(daemonLogRoot).toAbsolutePath().normalize();
+        this.workerLogs = workerLogs;
         this.resourceAuthorizer = resourceAuthorizer;
         this.fileDownloadSizeDistMb = metricsRegistry.registerHistogram("logviewer:download-file-size-rounded-MB");
         this.numFileDownloadExceptions = metricsRegistry.registerMeter(ExceptionMeterNames.NUM_FILE_DOWNLOAD_EXCEPTIONS);
@@ -79,6 +82,10 @@ public Response downloadFile(String host, String fileName, String user, boolean
 
         if (file.toFile().exists()) {
             if (isDaemon || resourceAuthorizer.isUserAllowedToAccessFile(user, fileName)) {
+                if (!isDaemon) {
+                    //Only widen the permission of a worker log once the request is known to be served
+                    workerLogs.setLogFilePermission(fileName);
+                }
                 fileDownloadSizeDistMb.update(Math.round((double) file.toFile().length() / FileUtils.ONE_MB));
                 String downloadedFileName;
                 Path pathRelativeToRootDir = rootDir.relativize(file);

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

@@ -24,6 +24,11 @@
 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.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import com.google.common.net.HttpHeaders;
 import java.io.IOException;
@@ -121,6 +126,79 @@ public void testDownloadDaemonLogFilePathOutsideLogRoot() throws IOException {
         }
     }
 
+    @Test
+    public void testDownloadLogFileUnauthorizedUserDoesNotChangeLogFilePermission() throws IOException {
+        try (TmpPath rootPath = new TmpPath()) {
+            Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+            Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
+            Path file = workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log");
+            Files.createDirectories(file.getParent());
+            Files.createFile(file);
+
+            ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+            when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(), anyString())).thenReturn(false);
+            WorkerLogs workerLogs = mock(WorkerLogs.class);
+
+            LogviewerLogDownloadHandler handler = new LogviewerLogDownloadHandler(workerLogRoot.toString(),
+                daemonLogRoot.toString(), workerLogs, resourceAuthorizer, new StormMetricsRegistry());
+
+            Response response = handler.downloadLogFile("host", "topoA/1111/worker.log", "user");
+
+            Utils.forceDelete(rootPath.toString());
+
+            assertThat(response.getStatus(), is(Response.Status.FORBIDDEN.getStatusCode()));
+            verify(workerLogs, never()).setLogFilePermission(anyString());
+        }
+    }
+
+    @Test
+    public void testDownloadLogFileAuthorizedUserSetsLogFilePermission() throws IOException {
+        try (TmpPath rootPath = new TmpPath()) {
+            Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+            Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
+            Path file = workerLogRoot.resolve("topoA").resolve("1111").resolve("worker.log");
+            Files.createDirectories(file.getParent());
+            Files.createFile(file);
+
+            ResourceAuthorizer resourceAuthorizer = mock(ResourceAuthorizer.class);
+            when(resourceAuthorizer.isUserAllowedToAccessFile(anyString(), anyString())).thenReturn(true);
+            WorkerLogs workerLogs = mock(WorkerLogs.class);
+
+            LogviewerLogDownloadHandler handler = new LogviewerLogDownloadHandler(workerLogRoot.toString(),
+                daemonLogRoot.toString(), workerLogs, resourceAuthorizer, new StormMetricsRegistry());
+
+            Response response = handler.downloadLogFile("host", "topoA/1111/worker.log", "user");
+
+            Utils.forceDelete(rootPath.toString());
+
+            assertThat(response.getStatus(), is(Response.Status.OK.getStatusCode()));
+            verify(workerLogs).setLogFilePermission("topoA/1111/worker.log");
+        }
+    }
+
+    @Test
+    public void testDownloadDaemonLogFileDoesNotChangeLogFilePermission() throws IOException {
+        try (TmpPath rootPath = new TmpPath()) {
+            Path daemonLogRoot = rootPath.getFile().toPath().resolve("logs");
+            Path workerLogRoot = daemonLogRoot.resolve("workers-artifacts");
+            Path daemonFile = daemonLogRoot.resolve("nimbus.log");
+            Files.createDirectories(workerLogRoot);
+            Files.createFile(daemonFile);
+
+            WorkerLogs workerLogs = mock(WorkerLogs.class);
+
+            LogviewerLogDownloadHandler handler = new LogviewerLogDownloadHandler(workerLogRoot.toString(),
+                daemonLogRoot.toString(), workerLogs, new ResourceAuthorizer(Utils.readStormConfig()), new StormMetricsRegistry());
+
+            Response response = handler.downloadDaemonLogFile("host", "nimbus.log", "user");
+
+            Utils.forceDelete(rootPath.toString());
+
+            assertThat(response.getStatus(), is(Response.Status.OK.getStatusCode()));
+            verify(workerLogs, never()).setLogFilePermission(anyString());
+        }
+    }
+
     private LogviewerLogDownloadHandler createHandlerTraversalTests(Path rootPath) throws IOException {
         Path daemonLogRoot = rootPath.resolve("logs");
         Path fileOutsideDaemonRoot = rootPath.resolve("evil.sh");