diff --git a/scanpipe/pipes/fetch.py b/scanpipe/pipes/fetch.py index 3cbbb13200..4b6b07a75e 100644 --- a/scanpipe/pipes/fetch.py +++ b/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: diff --git a/scanpipe/tests/pipes/test_fetch.py b/scanpipe/tests/pipes/test_fetch.py index e268c9ee4d..8c9916a5dc 100644 --- a/scanpipe/tests/pipes/test_fetch.py +++ b/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):