Validate and confine HTTP Content-Disposition filenames in fetch_http

scanpipe/pipes/fetch.py

@@ -77,6 +77,45 @@
 # urllib3-based HTTP client disagree on the host part of a URL, defeating the
 # ``is_safe_url`` check below.
 UNSAFE_URL_CHARS_RE = re.compile(r"[\x00-\x20\x7f\\]")
+UNSAFE_FILENAME_CHARS_RE = re.compile(r"[\x00-\x1f\x7f/\\]")
+
+
+def is_safe_filename(filename, download_directory):
+    """
+    Return True if `filename` is a safe single file name that resolves
+    strictly inside `download_directory` without traversing or escaping.
+    """
+    if not filename or not isinstance(filename, str):
+        return False
+
+    filename = filename.strip()
+    if not filename or filename in (".", ".."):
+        return False
+
+    # Reject null bytes, control characters, and POSIX / Windows path separators
+    if UNSAFE_FILENAME_CHARS_RE.search(filename):
+        return False
+
+    # Reject Windows drive prefixes or stream/colon syntax (e.g. C:, file:stream)
+    if ":" in filename:
+        return False
+
+    download_dir = Path(download_directory).resolve()
+    candidate_path = (download_dir / filename).resolve()
+
+    # Must resolve strictly inside download_dir and not be download_dir itself
+    if (
+        not candidate_path.is_relative_to(download_dir)
+        or candidate_path == download_dir
+    ):
+        return False
+
+    # Account for existing symlinks pointing outside download_dir
+    target = download_dir / filename
+    if target.is_symlink() and not target.resolve().is_relative_to(download_dir):
+        return False
+
+    return True
 
 
 def get_request_session(uri):
@@ -106,16 +145,27 @@ def fetch_http(uri, to=None):
     if response.status_code != 200:
         raise requests.RequestException
 
+    download_directory = to or tempfile.mkdtemp()
+    download_directory_path = Path(download_directory).resolve()
+    download_directory_path.mkdir(parents=True, exist_ok=True)
+
     content_disposition = response.headers.get("content-disposition", "")
     _, params = parse_header_parameters(content_disposition)
-    filename = params.get("filename")
-    if not filename:
+    raw_filename = params.get("filename")
+
+    if raw_filename is not None and raw_filename.strip():
+        filename = raw_filename.strip()
+        if not is_safe_filename(filename, download_directory_path):
+            raise ValueError(f"Invalid or unsafe filename: {raw_filename!r}")
+    else:
         # Using `response.url` in place of provided `Scan.uri` since the former
         # will be more accurate in case of HTTP redirect.
-        filename = unquote(Path(urlparse(response.url).path).name)
+        raw_url_filename = unquote(Path(urlparse(response.url).path).name)
+        filename = raw_url_filename.strip() if raw_url_filename else ""
+        if not is_safe_filename(filename, download_directory_path):
+            raise ValueError(f"Invalid or unsafe filename: {raw_url_filename!r}")
 
-    download_directory = to or tempfile.mkdtemp()
-    output_file = Path(download_directory, filename)
+    output_file = download_directory_path / filename
 
     file_content = response.content
     with open(output_file, "wb") as f:

scanpipe/tests/pipes/test_fetch.py

@@ -20,7 +20,9 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
+import os
 import socket
+import tempfile
 from pathlib import Path
 from unittest import mock
 
@@ -94,6 +96,172 @@ def test_scanpipe_pipes_fetch_http(self, mock_get, mock_is_safe_url):
         downloaded_file = fetch.fetch_http(url)
         self.assertTrue(Path(downloaded_file.directory, "another_name.zip").exists())
 
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_traversal_attempts(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/filename.zip"
+        invalid_filenames = [
+            "../../evil.zip",
+            "../evil.zip",
+            "sub/dir/evil.zip",
+            "..",
+            ".",
+        ]
+        for invalid_name in invalid_filenames:
+            headers = {"content-disposition": f'attachment; filename="{invalid_name}"'}
+            mock_get.return_value = make_mock_response(url=url, headers=headers)
+            with self.assertRaises(ValueError) as cm:
+                fetch.fetch_http(url)
+            self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_absolute_paths(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/filename.zip"
+        invalid_filenames = [
+            "/etc/passwd",
+            "/evil.zip",
+            "/var/test.zip",
+        ]
+        for invalid_name in invalid_filenames:
+            headers = {"content-disposition": f'attachment; filename="{invalid_name}"'}
+            mock_get.return_value = make_mock_response(url=url, headers=headers)
+            with self.assertRaises(ValueError) as cm:
+                fetch.fetch_http(url)
+            self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_windows_syntax(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/filename.zip"
+        invalid_filenames = [
+            r"..\..\evil.zip",
+            r"C:\evil.zip",
+            r"C:evil.zip",
+            r"D:/evil.zip",
+            r"\\server\share\evil.zip",
+        ]
+        for invalid_name in invalid_filenames:
+            headers = {"content-disposition": f'attachment; filename="{invalid_name}"'}
+            mock_get.return_value = make_mock_response(url=url, headers=headers)
+            with self.assertRaises(ValueError) as cm:
+                fetch.fetch_http(url)
+            self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_control_chars(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/filename.zip"
+        invalid_filenames = [
+            "evil\x00.zip",
+            "evil\n.zip",
+            "evil\r\ntest.zip",
+            "evil\x1f.zip",
+            "evil\x7f.zip",
+        ]
+        for invalid_name in invalid_filenames:
+            headers = {"content-disposition": f'attachment; filename="{invalid_name}"'}
+            mock_get.return_value = make_mock_response(url=url, headers=headers)
+            with self.assertRaises(ValueError) as cm:
+                fetch.fetch_http(url)
+            self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_empty_and_fallback(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/safe_fallback.zip"
+
+        # Empty filename parameter falls back to URL filename
+        headers = {"content-disposition": 'attachment; filename=""'}
+        mock_get.return_value = make_mock_response(url=url, headers=headers)
+        downloaded = fetch.fetch_http(url)
+        self.assertEqual("safe_fallback.zip", downloaded.filename)
+        self.assertTrue(Path(downloaded.directory, "safe_fallback.zip").exists())
+
+        # Whitespace-only filename parameter falls back to URL filename
+        headers = {"content-disposition": 'attachment; filename="   "'}
+        mock_get.return_value = make_mock_response(url=url, headers=headers)
+        downloaded = fetch.fetch_http(url)
+        self.assertEqual("safe_fallback.zip", downloaded.filename)
+        self.assertTrue(Path(downloaded.directory, "safe_fallback.zip").exists())
+
+        # Empty header filename and empty URL filename raises ValueError
+        empty_path_url = "https://example.com/"
+        headers = {"content-disposition": 'attachment; filename=""'}
+        mock_get.return_value = make_mock_response(url=empty_path_url, headers=headers)
+        with self.assertRaises(ValueError) as cm:
+            fetch.fetch_http(empty_path_url)
+        self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_safe_names(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/default.zip"
+        safe_names = [
+            "safe-release_v1.0.tar.gz",
+            "report (final).pdf",
+            "archive.bundle.zip",
+        ]
+        for safe_name in safe_names:
+            headers = {"content-disposition": f'attachment; filename="{safe_name}"'}
+            mock_get.return_value = make_mock_response(url=url, headers=headers)
+            downloaded = fetch.fetch_http(url)
+            self.assertEqual(safe_name, downloaded.filename)
+            self.assertEqual(downloaded.path, Path(downloaded.directory) / safe_name)
+            self.assertTrue(downloaded.path.exists())
+            self.assertGreater(downloaded.size, 0)
+            self.assertTrue(downloaded.sha1)
+            self.assertTrue(downloaded.md5)
+
+    @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_http_content_disposition_symlink(
+        self, mock_get, mock_is_safe_url
+    ):
+        url = "https://example.com/file.zip"
+        headers = {"content-disposition": 'attachment; filename="link.zip"'}
+        mock_get.return_value = make_mock_response(url=url, headers=headers)
+
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            tempfile.TemporaryDirectory() as outside_dir,
+        ):
+            download_dir = Path(temp_dir)
+            outside_target = Path(outside_dir) / "target.txt"
+            symlink_path = download_dir / "link.zip"
+
+            can_symlink = True
+            try:
+                os.symlink(outside_target, symlink_path)
+            except (OSError, NotImplementedError):
+                can_symlink = False
+
+            if can_symlink:
+                with self.assertRaises(ValueError) as cm:
+                    fetch.fetch_http(url, to=str(download_dir))
+                self.assertIn("Invalid or unsafe filename", str(cm.exception))
+                self.assertFalse(outside_target.exists())
+            else:
+                with mock.patch.object(Path, "is_symlink", return_value=True):
+                    with mock.patch.object(
+                        Path, "resolve", return_value=outside_target.resolve()
+                    ):
+                        with self.assertRaises(ValueError) as cm:
+                            fetch.fetch_http(url, to=str(download_dir))
+                        self.assertIn("Invalid or unsafe filename", str(cm.exception))
+
     @mock.patch("scanpipe.pipes.fetch.is_safe_url", return_value=True)
     @mock.patch("requests.sessions.Session.get")
     def test_scanpipe_pipes_fetch_package_url(self, mock_get, mock_is_safe_url):