Reject submitted jar locations that are not inside the Nimbus inbox

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

@@ -40,6 +40,7 @@
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
 import java.security.Principal;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -1101,7 +1102,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));
@@ -1221,8 +1222,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
@@ -1835,6 +1836,28 @@
         return ServerConfigUtils.masterInbox(conf);
     }
 
+    /**
+     * Check that a client supplied jar location names a file inside the nimbus inbox, i.e. one that was handed out by
+     * beginFileUpload and written through uploadChunk/finishFileUpload. Both paths are canonicalized first so that
+     * ".." segments and symlinks cannot point outside of the inbox.
+     *
+     * @param inboxLocation the nimbus inbox directory
+     * @param uploadedJarLocation the client supplied jar location
+     * @throws AuthorizationException if uploadedJarLocation is not inside the inbox
+     * @throws IOException if the paths could not be resolved
+     */
+    @VisibleForTesting
+    static void validateUploadedJarLocation(String inboxLocation, String uploadedJarLocation)
+        throws AuthorizationException, IOException {
+        Path inboxDir = new File(inboxLocation).getCanonicalFile().toPath();
+        Path uploadedJar = new File(uploadedJarLocation).getCanonicalFile().toPath();
+        if (uploadedJar.equals(inboxDir) || !uploadedJar.startsWith(inboxDir)) {
+            throw new WrappedAuthorizationException("uploadedJarLocation " + uploadedJarLocation
+                + " is not inside the nimbus inbox. Topology jars must be uploaded through beginFileUpload/uploadChunk"
+                + "/finishFileUpload before the topology is submitted.");
+        }
+    }
+
     /**
      * Used for local cluster.
      *
@@ -1927,6 +1950,7 @@
         String jarKey = ConfigUtils.masterStormJarKey(topoId);
         if (tmpJarLocation != null) {
             //in local mode there is no jar
+            validateUploadedJarLocation(getInbox(), tmpJarLocation);
             try (FileInputStream fin = new FileInputStream(tmpJarLocation)) {
                 store.createBlob(jarKey, fin, new SettableBlobMeta(BlobStoreAclHandler.DEFAULT), subject);
             }
@@ -2920,7 +2944,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyJarsInTopology(topoId);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3395,8 +3419,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;

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

@@ -18,17 +18,22 @@
 
 package org.apache.storm.daemon.nimbus;
 
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.commons.io.FileUtils;
 import org.apache.storm.Config;
 import org.apache.storm.DaemonConfig;
 import org.apache.storm.blobstore.BlobStore;
 import org.apache.storm.blobstore.KeySequenceNumber;
 import org.apache.storm.blobstore.LocalFsBlobStore;
 import org.apache.storm.cluster.IStormClusterState;
+import org.apache.storm.generated.AuthorizationException;
 import org.apache.storm.generated.InvalidTopologyException;
 import org.apache.storm.generated.KeyNotFoundException;
 import org.apache.storm.generated.StormTopology;
@@ -200,4 +205,38 @@ void testCreateStateInZookeeperWhenKeyNotFoundHandlesException() throws Exceptio
             verify(stormClusterState, never()).setupBlob(eq(BLOB_FILE_KEY), eq(nimbusInfo), any());
         }
     }
+
+    @Test
+    void testValidateUploadedJarLocationRejectsLocationsOutsideTheInbox() throws Exception {
+        Path inbox = Files.createTempDirectory("nimbus-inbox");
+        Path sibling = Paths.get(inbox + "evil");
+        try {
+            Path jar = Files.write(inbox.resolve("stormjar-cafebabe.jar"), new byte[]{ 1 });
+            Path outside = Files.write(Files.createDirectory(sibling).resolve("stormjar-cafebabe.jar"), new byte[]{ 1 });
+
+            // a location handed out by beginFileUpload is accepted, and so is one that only walks inside the inbox
+            Nimbus.validateUploadedJarLocation(inbox.toString(), jar.toString());
+            Files.createDirectory(inbox.resolve("nested"));
+            Nimbus.validateUploadedJarLocation(inbox.toString(), inbox + "/nested/../stormjar-cafebabe.jar");
+
+            // an absolute path elsewhere, a ".." walk out of the inbox, the inbox itself and a sibling directory
+            // whose name merely starts with the inbox path are all rejected
+            assertThrows(AuthorizationException.class,
+                () -> Nimbus.validateUploadedJarLocation(inbox.toString(), "/etc/passwd"));
+            assertThrows(AuthorizationException.class,
+                () -> Nimbus.validateUploadedJarLocation(inbox.toString(), inbox + "/../../etc/passwd"));
+            assertThrows(AuthorizationException.class,
+                () -> Nimbus.validateUploadedJarLocation(inbox.toString(), inbox.toString()));
+            assertThrows(AuthorizationException.class,
+                () -> Nimbus.validateUploadedJarLocation(inbox.toString(), outside.toString()));
+
+            // a symlink inside the inbox pointing back out of it is rejected too
+            Path link = Files.createSymbolicLink(inbox.resolve("stormjar-link.jar"), outside);
+            assertThrows(AuthorizationException.class,
+                () -> Nimbus.validateUploadedJarLocation(inbox.toString(), link.toString()));
+        } finally {
+            FileUtils.deleteQuietly(inbox.toFile());
+            FileUtils.deleteQuietly(sibling.toFile());
+        }
+    }
 }