Reject blobstore map local names that resolve outside the topology and worker directories

storm-server/src/main/java/org/apache/storm/daemon/supervisor/Container.java

@@ -50,6 +50,7 @@
 import org.apache.storm.utils.LocalState;
 import org.apache.storm.utils.ObjectReader;
 import org.apache.storm.utils.ServerConfigUtils;
+import org.apache.storm.utils.ServerUtils;
 import org.apache.storm.utils.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -410,8 +411,9 @@ protected void createBlobstoreLinks() throws IOException {
                     targetResourcesDir.toString());
             }
             for (String fileName : blobFileNames) {
-                ops.createSymlink(new File(workerRoot, fileName),
-                    new File(stormRoot, fileName));
+                // the localname may come from the topology conf, it must not point outside of the worker/dist dirs
+                ops.createSymlink(ServerUtils.resolveTopologyConfSuppliedName(new File(workerRoot), fileName),
+                    ServerUtils.resolveTopologyConfSuppliedName(new File(stormRoot), fileName));
             }
         } else if (blobFileNames.size() > 0) {
             LOG.warn("Symlinks are disabled, no symlinks created for blobs {}", blobFileNames);

storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java

@@ -743,7 +743,9 @@ public Void get() {
                                 // all things are from dependencies
                                 symlinkName = keyName;
                             }
-                            fsOps.createSymlink(new File(stormroot, symlinkName), rsrcFilePath);
+                            // the localname may come from the topology conf, it must not point outside of stormroot
+                            fsOps.createSymlink(ServerUtils.resolveTopologyConfSuppliedName(new File(stormroot), symlinkName),
+                                rsrcFilePath);
                         }
                     }
                 }

storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java

@@ -36,6 +36,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.FileSystems;
 import java.nio.file.Files;
+import java.nio.file.InvalidPathException;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -757,6 +758,44 @@ public static Subject principalNameToSubject(String name) {
         return sub;
     }
 
+    /**
+     * Resolve a name that came from the topology conf (for example the "localname" of a topology.blobstore.map entry)
+     * against a base directory. The name is only allowed to point at something strictly inside the base directory, so
+     * that a topology cannot make the supervisor create a symlink, and force delete whatever was there before, outside
+     * of the directories the supervisor manages for that topology.
+     *
+     * @param baseDir the directory the name has to resolve inside of
+     * @param name the name from the topology conf
+     * @return the resolved file
+     * @throws IOException if the name is empty, is absolute, contains a ".." component, or resolves outside of baseDir
+     */
+    public static File resolveTopologyConfSuppliedName(File baseDir, String name) throws IOException {
+        if (StringUtils.isEmpty(name)) {
+            throw new IOException("Invalid local name, it can't be null or empty string");
+        }
+        Path namePath;
+        try {
+            namePath = Paths.get(name);
+        } catch (InvalidPathException e) {
+            throw new IOException("Invalid local name '" + name + "'", e);
+        }
+        if (namePath.isAbsolute()) {
+            throw new IOException("Invalid local name '" + name + "', it must be relative");
+        }
+        for (Path part : namePath) {
+            if ("..".equals(part.toString())) {
+                throw new IOException("Invalid local name '" + name + "', it can't contain \"..\"");
+            }
+        }
+        File ret = new File(baseDir, name);
+        Path base = baseDir.toPath().toAbsolutePath().normalize();
+        Path resolved = ret.toPath().toAbsolutePath().normalize();
+        if (resolved.equals(base) || !resolved.startsWith(base)) {
+            throw new IOException("Invalid local name '" + name + "', it does not resolve inside of " + baseDir);
+        }
+        return ret;
+    }
+
     // Non-static impl methods exist for mocking purposes.
     public String currentClasspathImpl() {
         return System.getProperty("java.class.path");

storm-server/src/test/java/org/apache/storm/daemon/supervisor/ContainerTest.java

@@ -37,7 +37,9 @@
 import org.yaml.snakeyaml.Yaml;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
@@ -172,6 +174,47 @@ public void testSetup() throws Exception {
         verify(ops, never()).createSymlink(new File(workerRoot, "resources"), new File(distRoot, "resources"));
     }
 
+    @Test
+    public void testCreateBlobstoreLinks() throws Exception {
+        final int port = 8080;
+        final String topoId = "test_topology";
+        final String workerId = "worker_id";
+        final String stormLocal = asAbsPath("tmp", "testing");
+        final File workerRoot = asAbsFile(stormLocal, "workers", workerId);
+        final File distRoot = asAbsFile(stormLocal, "supervisor", "stormdist", topoId);
+
+        final Map<String, Object> superConf = new HashMap<>();
+        superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
+        superConf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, stormLocal);
+
+        final Map<String, Object> topoConf = new HashMap<>();
+        Map<String, Object> blobInfo = new HashMap<>();
+        blobInfo.put("localname", "simple.txt");
+        topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, Collections.singletonMap("simple", blobInfo));
+
+        AdvancedFSOps ops = mock(AdvancedFSOps.class);
+        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
+
+        LocalAssignment la = new LocalAssignment();
+        la.set_topology_id(topoId);
+        ResourceIsolationInterface iso = mock(ResourceIsolationInterface.class);
+        MockContainer mc = new MockContainer(ContainerType.LAUNCH, superConf,
+                                             "SUPERVISOR", 6628, port, la, iso, workerId, topoConf, ops, new StormMetricsRegistry());
+
+        mc.createBlobstoreLinks();
+        verify(ops).createSymlink(new File(workerRoot, "simple.txt"), new File(distRoot, "simple.txt"));
+
+        //a localname that points outside of the worker root must not result in any link
+        AdvancedFSOps badOps = mock(AdvancedFSOps.class);
+        when(badOps.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
+        blobInfo.put("localname", asPath("..", "..", "escaped.txt"));
+        MockContainer badMc = new MockContainer(ContainerType.LAUNCH, superConf,
+                                                "SUPERVISOR", 6628, port, la, iso, workerId, topoConf, badOps, new StormMetricsRegistry());
+
+        assertThrows(IOException.class, () -> badMc.createBlobstoreLinks());
+        verify(badOps, never()).createSymlink(any(File.class), any(File.class));
+    }
+
     @Test
     public void testCleanup() throws Exception {
         final int supervisorPort = 6628;

storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
+import java.nio.file.LinkOption;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
@@ -30,6 +31,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -241,6 +243,76 @@ public void testRequestDownloadTopologyBlobs() throws Exception {
     }
 
 
+    @Test
+    public void testRequestDownloadTopologyBlobsWithLocalNameOutsideOfStormRoot() throws Exception {
+        ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
+        ConfigUtils previousConfigUtils = ConfigUtils.setInstance(mockedConfigUtils);
+
+        AsyncLocalizer victim = null;
+
+        try (TmpPath stormLocal = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {
+
+            Map<String, Object> conf = new HashMap<>();
+            conf.put(Config.STORM_LOCAL_DIR, stormLocal.getPath());
+
+            AdvancedFSOps ops = AdvancedFSOps.make(conf);
+            StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+
+            victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), metricsRegistry));
+
+            final String topoId = "TOPO-12345";
+            final String user = "user";
+
+            final Path userDir = Paths.get(stormLocal.getPath(), user);
+            final Path topologyDirRoot = Paths.get(stormLocal.getPath(), topoId);
+
+            // the localname comes from the topology conf and tries to point outside of the topology's dist dir
+            final String escapingLocalName = Joiner.on(File.separator).join("..", "escaped.txt");
+            final String simpleKey = "simple";
+            Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
+            Map<String, Object> simple = new HashMap<>();
+            simple.put("localname", escapingLocalName);
+            simple.put("uncompress", false);
+            topoBlobMap.put(simpleKey, simple);
+
+            final int port = 8080;
+
+            Map<String, Object> topoConf = new HashMap<>(conf);
+            topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
+            topoConf.put(Config.TOPOLOGY_NAME, "TOPO");
+
+            List<LocalizedResource> localizedList = new ArrayList<>();
+            LocalizedResource simpleLocal = new LocalizedResource(simpleKey, localizerRoot.getFile().toPath(), false, ops, conf, user,
+                metricsRegistry);
+            localizedList.add(simpleLocal);
+
+            when(mockedConfigUtils.supervisorStormDistRootImpl(conf, topoId)).thenReturn(topologyDirRoot.toString());
+            when(mockedConfigUtils.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
+            when(mockedConfigUtils.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(constructEmptyStormTopology());
+
+            //Write the mocking backwards so the actual method is not called on the spy object
+            doReturn(CompletableFuture.supplyAsync(() -> null)).when(victim)
+                    .requestDownloadBaseTopologyBlobs(any(), eq(null));
+
+            Files.createDirectories(topologyDirRoot);
+
+            doReturn(userDir.toFile()).when(victim).getLocalUserFileCacheDir(user);
+            doReturn(localizedList).when(victim).getBlobs(any(List.class), any(), any());
+
+            Future<Void> f = victim.requestDownloadTopologyBlobs(constructLocalAssignment(topoId, user), port, null);
+            assertThrows(ExecutionException.class, () -> f.get(20, TimeUnit.SECONDS));
+
+            // nothing was created outside of the topology's dist dir
+            assertFalse(Files.exists(topologyDirRoot.getParent().resolve("escaped.txt"), LinkOption.NOFOLLOW_LINKS));
+
+        } finally {
+            ConfigUtils.setInstance(previousConfigUtils);
+            if (victim != null) {
+                victim.close();
+            }
+        }
+    }
+
     @Test
     public void testRequestDownloadTopologyBlobsLocalMode() throws Exception {
         // tests download of topology blobs in local mode on a topology without resources folder

storm-server/src/test/java/org/apache/storm/utils/ServerUtilsTest.java

@@ -23,6 +23,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -397,4 +398,24 @@ private boolean sleepInterrupted(long milliSeconds) {
         }
         return false;
     }
+
+    @Test
+    public void testResolveTopologyConfSuppliedName() throws Exception {
+        File baseDir = new File(System.getProperty("java.io.tmpdir"), "stormdist/topo-1-1");
+
+        assertEquals(new File(baseDir, "myblob"), ServerUtils.resolveTopologyConfSuppliedName(baseDir, "myblob"));
+        assertEquals(new File(baseDir, "resources/myblob"), ServerUtils.resolveTopologyConfSuppliedName(baseDir, "resources/myblob"));
+
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, ".."));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, "../other-topo/stormjar.jar"));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, "a/../../../etc/passwd"));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, "a/../b/../.."));
+        //a sibling directory whose name only starts with the base directory name is not inside it
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, "../topo-1-1-evil/stormjar.jar"));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, "."));
+        assertThrows(IOException.class,
+            () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, new File("etc", "passwd").getAbsolutePath()));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, ""));
+        assertThrows(IOException.class, () -> ServerUtils.resolveTopologyConfSuppliedName(baseDir, null));
+    }
 }