Add support for fetching Package URLs (fetch_package_url) #1383

Signed-off-by: tdruez <[email protected]>

Author: tdruez

scanpipe/pipes/fetch.py

@@ -39,6 +39,8 @@
 from commoncode import command
 from commoncode.hash import multi_checksums
 from commoncode.text import python_safe_name
+from packageurl import PackageURL
+from packageurl.contrib import purl2url
 from plugincode.location_provider import get_location
 from requests import auth as request_auth
 
@@ -356,6 +358,17 @@ def fetch_git_repo(url, to=None):
     )
 
 
+def fetch_package_url(url):
+    # Ensure the provided Package URL is valid, or raise a ValueError.
+    PackageURL.from_string(url)
+
+    # Resolve a Download URL using purl2url.
+    if download_url := purl2url.get_download_url(url):
+        return fetch_http(download_url)
+
+    raise ValueError(f"Could not resolve a download URL for {url}.")
+
+
 SCHEME_TO_FETCHER_MAPPING = {
     "http": fetch_http,
     "https": fetch_http,
@@ -371,6 +384,9 @@ def get_fetcher(url):
     if url.rstrip("/").endswith(".git"):
         return fetch_git_repo
 
+    if url.startswith("pkg:"):
+        return fetch_package_url
+
     # Not using `urlparse(url).scheme` for the scheme as it converts to lower case.
     scheme = url.split("://")[0]
 

scanpipe/tests/pipes/test_fetch.py

@@ -42,6 +42,7 @@ def test_scanpipe_pipes_fetch_get_fetcher(self):
         git_http_url = "https://github.com/aboutcode-org/scancode.io.git"
         self.assertEqual(fetch.fetch_git_repo, fetch.get_fetcher(git_http_url))
         self.assertEqual(fetch.fetch_git_repo, fetch.get_fetcher(git_http_url + "/"))
+        self.assertEqual(fetch.fetch_package_url, fetch.get_fetcher("pkg:npm/[email protected]"))
 
         with self.assertRaises(ValueError) as cm:
             fetch.get_fetcher("")
@@ -88,6 +89,25 @@ def test_scanpipe_pipes_fetch_http(self, mock_get):
         downloaded_file = fetch.fetch_http(url)
         self.assertTrue(Path(downloaded_file.directory, "another_name.zip").exists())
 
+    @mock.patch("requests.sessions.Session.get")
+    def test_scanpipe_pipes_fetch_package_url(self, mock_get):
+        package_url = "pkg:not_a_valid_purl"
+        with self.assertRaises(ValueError) as cm:
+            fetch.fetch_package_url(package_url)
+        expected = f"purl is missing the required type component: '{package_url}'."
+        self.assertEqual(expected, str(cm.exception))
+
+        package_url = "pkg:generic/name@version"
+        with self.assertRaises(ValueError) as cm:
+            fetch.fetch_package_url(package_url)
+        expected = f"Could not resolve a download URL for {package_url}."
+        self.assertEqual(expected, str(cm.exception))
+
+        package_url = "pkg:npm/[email protected]"
+        mock_get.return_value = make_mock_response(url="https://exa.com/filename.zip")
+        downloaded_file = fetch.fetch_package_url(package_url)
+        self.assertTrue(Path(downloaded_file.directory, "filename.zip").exists())
+
     @mock.patch("scanpipe.pipes.fetch.get_docker_image_platform")
     @mock.patch("scanpipe.pipes.fetch._get_skopeo_location")
     @mock.patch("scanpipe.pipes.fetch.run_command_safely")