Add URL scheme validation with explicit error messages #1047 (#1085)

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

Author: tdruez

CHANGELOG.rst

@@ -41,6 +41,9 @@ v33.2.0 (unreleased)
 - Extract all archives recursively in the `scan_single_package` pipeline.
   https://github.com/nexB/scancode.io/issues/1081
 
+- Add URL scheme validation with explicit error messages for input URLs.
+  https://github.com/nexB/scancode.io/issues/1047
+
 v33.1.0 (2024-02-02)
 --------------------
 

docs/built-in-pipelines.rst

@@ -58,8 +58,8 @@ Find Vulnerabilities (addon)
 
 .. _pipeline_inspect_elf:
 
-Inspect ELF Binaries
---------------------
+Inspect ELF Binaries (addon)
+----------------------------
 .. autoclass:: scanpipe.pipelines.inspect_elf_binaries.InspectELFBinaries()
     :members:
     :member-order: bysource

scanpipe/forms.py

@@ -29,7 +29,7 @@
 from taggit.forms import TagWidget
 
 from scanpipe.models import Project
-from scanpipe.pipes.fetch import check_urls_availability
+from scanpipe.pipes import fetch
 
 scanpipe_app = apps.get_app_config("scanpipe")
 
@@ -76,6 +76,23 @@ class InputsBaseForm(forms.Form):
     class Media:
         js = ("add-inputs.js",)
 
+    @staticmethod
+    def validate_scheme(input_urls):
+        """
+        Raise a validation error if some of the `input_urls` have a scheme that is not
+        supported.
+        """
+        errors = []
+
+        for url in input_urls:
+            try:
+                fetch.get_fetcher(url)
+            except ValueError as e:
+                errors.append(str(e))
+
+        if errors:
+            raise ValidationError("\n".join(errors))
+
     def clean_input_urls(self):
         """
         Fetch the `input_urls` and sets the `downloads` objects in the cleaned_data.
@@ -84,8 +101,9 @@ def clean_input_urls(self):
         input_urls_str = self.cleaned_data.get("input_urls", "")
         input_urls = input_urls_str.split()
 
-        errors = check_urls_availability(input_urls)
-        if errors:
+        self.validate_scheme(input_urls)
+
+        if errors := fetch.check_urls_availability(input_urls):
             raise ValidationError("Could not fetch:\n" + "\n".join(errors))
 
         return input_urls

scanpipe/pipes/fetch.py

@@ -211,27 +211,28 @@ def get_docker_image_platform(docker_reference):
             )
 
 
-def fetch_docker_image(docker_reference, to=None):
+def fetch_docker_image(download_url, to=None):
     """
-    Fetch a docker image from the provided Docker image `docker_reference`
-    docker:// reference URL. Return a `download` object.
+    Fetch a docker image from the provided Docker image `download_url`,
+    using the "docker://reference" URL syntax.
+    Return a `Download` object.
 
     Docker references are documented here:
     https://github.com/containers/skopeo/blob/0faf16017/docs/skopeo.1.md#image-names
     """
     whitelist = r"^docker://[a-zA-Z0-9_.:/@-]+$"
-    if not re.match(whitelist, docker_reference):
+    if not re.match(whitelist, download_url):
         raise ValueError("Invalid Docker reference.")
 
-    name = python_safe_name(docker_reference.replace("docker://", ""))
-    filename = f"{name}.tar"
+    reference = download_url.replace("docker://", "")
+    filename = f"{python_safe_name(reference)}.tar"
     download_directory = to or tempfile.mkdtemp()
     output_file = Path(download_directory, filename)
     target = f"docker-archive:{output_file}"
     skopeo_executable = _get_skopeo_location()
 
     platform_args = []
-    if platform := get_docker_image_platform(docker_reference):
+    if platform := get_docker_image_platform(download_url):
         os, arch, variant = platform
         if os:
             platform_args.append(f"--override-os={os}")
@@ -245,7 +246,7 @@ def fetch_docker_image(docker_reference, to=None):
         "copy",
         "--insecure-policy",
         *platform_args,
-        docker_reference,
+        download_url,
         target,
     )
     logger.info(f"Fetching image with: {cmd_args}")
@@ -255,7 +256,7 @@ def fetch_docker_image(docker_reference, to=None):
     checksums = multi_checksums(output_file, ("md5", "sha1"))
 
     return Download(
-        uri=docker_reference,
+        uri=download_url,
         directory=download_directory,
         filename=filename,
         path=output_file,
@@ -265,16 +266,30 @@ def fetch_docker_image(docker_reference, to=None):
     )
 
 
-def _get_fetcher(url):
-    """Return the fetcher function based on the provided `url`."""
-    if url.startswith("docker://"):
-        return fetch_docker_image
-    return fetch_http
+SCHEME_TO_FETCHER_MAPPING = {
+    "http": fetch_http,
+    "https": fetch_http,
+    "docker": fetch_docker_image,
+}
+
+
+def get_fetcher(url):
+    """Return the fetcher function based on the provided `url` scheme."""
+    # Not using `urlparse(url).scheme` for the scheme as it converts to lower case.
+    scheme = url.split("://")[0]
+
+    if fetcher := SCHEME_TO_FETCHER_MAPPING.get(scheme):
+        return fetcher
+
+    error_msg = f"URL scheme '{scheme}' is not supported."
+    if scheme.lower() in SCHEME_TO_FETCHER_MAPPING:
+        error_msg += f" Did you mean: '{scheme.lower()}'?"
+    raise ValueError(error_msg)
 
 
 def fetch_url(url):
     """Fetch provided `url` and returns the result as a `Download` object."""
-    fetcher = _get_fetcher(url)
+    fetcher = get_fetcher(url)
     logger.info(f'Fetching "{url}" using {fetcher.__name__}')
     downloaded = fetcher(url)
     return downloaded

scanpipe/templates/scanpipe/project_form.html

@@ -85,7 +85,7 @@ <h2 class="subtitle mb-0 mb-4">
           {{ form.media }}
         </div>
 
-        <div class="column has-background-light has-border-radius">
+        <div class="column has-background-light has-border-radius mb-3">
           <h3 class="subtitle mb-3">Pipelines:</h3>
           {% for pipeline_name, pipeline_info in pipelines.items %}
             <div {% if not forloop.last %}class="mb-3"{% endif %}>

scanpipe/tests/pipes/test_fetch.py

@@ -31,6 +31,31 @@
 class ScanPipeFetchPipesTest(TestCase):
     data_location = Path(__file__).parent.parent / "data"
 
+    def test_scanpipe_pipes_fetch_get_fetcher(self):
+        self.assertEqual(fetch.fetch_http, fetch.get_fetcher("http://a.b/f.z"))
+        self.assertEqual(fetch.fetch_http, fetch.get_fetcher("https://a.b/f.z"))
+        self.assertEqual(fetch.fetch_docker_image, fetch.get_fetcher("docker://image"))
+
+        with self.assertRaises(ValueError) as cm:
+            fetch.get_fetcher("")
+        expected = "URL scheme '' is not supported."
+        self.assertEqual(expected, str(cm.exception))
+
+        with self.assertRaises(ValueError) as cm:
+            fetch.get_fetcher("abcd://a.b/f.z")
+        expected = "URL scheme 'abcd' is not supported."
+        self.assertEqual(expected, str(cm.exception))
+
+        with self.assertRaises(ValueError) as cm:
+            fetch.get_fetcher("Docker://image")
+        expected = "URL scheme 'Docker' is not supported. Did you mean: 'docker'?"
+        self.assertEqual(expected, str(cm.exception))
+
+        with self.assertRaises(ValueError) as cm:
+            fetch.get_fetcher("DOCKER://image")
+        expected = "URL scheme 'DOCKER' is not supported. Did you mean: 'docker'?"
+        self.assertEqual(expected, str(cm.exception))
+
     @mock.patch("requests.get")
     def test_scanpipe_pipes_fetch_http(self, mock_get):
         url = "https://example.com/filename.zip"
@@ -63,8 +88,12 @@ def test_scanpipe_pipes_fetch_http(self, mock_get):
     def test_scanpipe_pipes_fetch_docker_image(
         self, mock_run_command_safely, mock_skopeo, mock_platform
     ):
-        url = "docker://debian:10.9"
+        with self.assertRaises(ValueError) as cm:
+            fetch.fetch_docker_image("Docker://debian")
+        expected = "Invalid Docker reference."
+        self.assertEqual(expected, str(cm.exception))
 
+        url = "docker://debian:10.9"
         mock_platform.return_value = "linux", "amd64", ""
         mock_skopeo.return_value = "skopeo"
         mock_run_command_safely.side_effect = Exception

scanpipe/tests/test_forms.py

@@ -41,9 +41,16 @@ def setUp(self):
     @mock.patch("requests.head")
     def test_scanpipe_forms_inputs_base_form_input_urls(self, mock_head):
         data = {
-            "input_urls": "https://example.com/archive.zip",
+            "input_urls": "Docker://debian",
         }
+        form = InputsBaseForm(data=data)
+        self.assertFalse(form.is_valid())
+        error_msg = "URL scheme 'Docker' is not supported. Did you mean: 'docker'?"
+        self.assertEqual({"input_urls": [error_msg]}, form.errors)
 
+        data = {
+            "input_urls": "https://example.com/archive.zip",
+        }
         mock_head.side_effect = requests.exceptions.RequestException
         form = InputsBaseForm(data=data)
         self.assertFalse(form.is_valid())