Prevent URL cache path traversal

Summary: `HTTPURLHandler._get_local_path()` used URL path components when building cache destinations. Normalize URL path components with `PurePosixPath`, reject `..` traversal, and validate the resolved cache path remains under `get_cache_dir()`.

Reviewed By: nponte

Differential Revision: D108953882

fbshipit-source-id: 6f8b9704b0712ddbf3e1b97430448ebe0300ef75

Author: herman5

iopath/common/file_io.py

@@ -13,6 +13,7 @@
 import uuid
 from collections import OrderedDict
 from io import IOBase
+from pathlib import Path, PurePosixPath
 from types import TracebackType
 from typing import (
     Any,
@@ -917,9 +918,19 @@ def _get_local_path(
         ):
             logger = logging.getLogger(__name__)
             parsed_url = urlparse(path)
-            dirname = os.path.join(
-                get_cache_dir(cache_dir), os.path.dirname(parsed_url.path.lstrip("/"))
-            )
+            url_path_parts = [
+                part
+                for part in parsed_url.path.replace("\\", "/").split("/")
+                if part not in ("", ".")
+            ]
+            if ".." in url_path_parts:
+                raise ValueError(
+                    "URL path must not contain '..' components: {}".format(path)
+                )
+
+            url_path = PurePosixPath(*url_path_parts)
+            cache_root = Path(get_cache_dir(cache_dir))
+            dirname = cache_root / url_path.parent
             filename = path.split("/")[-1]
 
             if parsed_url.query:
@@ -928,13 +939,21 @@ def _get_local_path(
             if len(filename) > self.MAX_FILENAME_LEN:
                 filename = filename[:100] + "_" + uuid.uuid4().hex
 
-            cached = os.path.join(dirname, filename)
-            with file_lock(cached):
-                if not os.path.isfile(cached):
+            cached = dirname / filename
+            cache_root_resolved = cache_root.resolve()
+            cached_resolved = cached.resolve()
+            if not cached_resolved.is_relative_to(cache_root_resolved):
+                raise ValueError(
+                    "URL cache path must stay under cache directory: {}".format(path)
+                )
+
+            cached_path = os.fspath(cached)
+            with file_lock(cached_path):
+                if not os.path.isfile(cached_path):
                     logger.info("Downloading {} ...".format(path))
-                    cached = download(path, dirname, filename=filename)
-            logger.info("URL {} cached in {}".format(path, cached))
-            self.cache_map[path] = cached
+                    cached_path = download(path, os.fspath(dirname), filename=filename)
+            logger.info("URL {} cached in {}".format(path, cached_path))
+            self.cache_map[path] = cached_path
         return self.cache_map[path]
 
     def _open(
@@ -1595,7 +1614,6 @@ def register_handler(
             old_handler_type = type(self._path_handlers[prefix])
             if allow_override:
                 # if using the global PathManager, show the warnings
-                global g_pathmgr
                 if self == g_pathmgr:
                     logger.warning(
                         f"[PathManager] Attempting to register prefix '{prefix}' from "

tests/test_file_io.py

@@ -303,9 +303,9 @@ def test_open_read_async(self) -> None:
         with self.subTest("read binary"):
             test_data = {
                 "test_string": "test string",
-                1: 1,
-                1.0: 1.0,
-                True: True,
+                "test_int": 1,
+                "test_float": 1.0,
+                "test_bool": True,
             }
             tmp_binary_path = os.path.join(self._tmpdir, "test_binary.bin")  # type: ignore
             pickle.dump(test_data, open(tmp_binary_path, "wb"))
@@ -372,6 +372,35 @@ def test_get_local_path(self) -> None:
             self.assertTrue(os.path.exists(local_path))
             self.assertTrue(os.path.isfile(local_path))
 
+    def test_get_local_path_rejects_path_traversal(self) -> None:
+        with patch.object(
+            file_io, "get_cache_dir", return_value=self._cache_dir
+        ), patch.object(file_io, "download") as mock_download:
+            with self.assertRaisesRegex(ValueError, "must not contain"):
+                self._pathmgr.get_local_path(
+                    "https://example.com/models/../../escape.txt", force=True
+                )
+            mock_download.assert_not_called()
+
+    def test_get_local_path_rejects_cache_symlink_escape(self) -> None:
+        symlink_name = "escape_" + uuid.uuid4().hex
+        symlink_dir = os.path.join(self._cache_dir, symlink_name)
+        with tempfile.TemporaryDirectory() as outside_dir:
+            os.symlink(outside_dir, symlink_dir)
+            try:
+                with patch.object(
+                    file_io, "get_cache_dir", return_value=self._cache_dir
+                ), patch.object(file_io, "download") as mock_download:
+                    with self.assertRaisesRegex(ValueError, "cache path"):
+                        self._pathmgr.get_local_path(
+                            f"https://example.com/{symlink_name}/model.bin",
+                            force=True,
+                        )
+                    mock_download.assert_not_called()
+            finally:
+                if os.path.lexists(symlink_dir):
+                    os.unlink(symlink_dir)
+
     def test_open(self) -> None:
         with self._patch_download():
             with self._pathmgr.open(self._remote_uri, "rb") as f: