Add settings for providing Skopeo credentials #203 (#1098)

tdruez · web-flow · commit adf24ef7556f · 2024-03-04T13:22:51.000+01:00
Signed-off-by: tdruez &lt;tdruez@nexb.com&gt;
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -48,9 +48,10 @@ v33.2.0 (unreleased)
   action providing a value for the new `output_format` parameter.
   https://github.com/nexB/scancode.io/issues/1091
 
-- Add multiple settings related to fetching private files. Those settings allow to
+- Add settings related to fetching private files. Those settings allow to
   define credentials for various authentication types.
   https://github.com/nexB/scancode.io/issues/620
+  https://github.com/nexB/scancode.io/issues/203
 
 - Update matchcode-toolkit to v3.0.0
 
diff --git a/docs/application-settings.rst b/docs/application-settings.rst
@@ -373,3 +373,22 @@ If your credentials are stored in a
 location on disk using::
 
     SCANCODEIO_NETRC_LOCATION="~/.netrc"
+
+.. _scancodeio_settings_skopeo_credentials:
+
+SCANCODEIO_SKOPEO_CREDENTIALS
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+You can define the username and password for Skopeo to access containers private
+registries using the ``host=user:password`` syntax::
+
+  SCANCODEIO_SKOPEO_CREDENTIALS="host1=user:password,host2=user:password"
+
+.. _scancodeio_settings_skopeo_authfile_location:
+
+SCANCODEIO_SKOPEO_AUTHFILE_LOCATION
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Specify the path of the Skopeo authentication file using the following setting::
+
+    SCANCODEIO_SKOPEO_AUTHFILE_LOCATION="/path/to/auth.json"
diff --git a/docs/faq.rst b/docs/faq.rst
@@ -209,7 +209,12 @@ authentication type:
 - :ref:`Digest authentication <scancodeio_settings_fetch_digest_auth>`
 - :ref:`HTTP request headers <scancodeio_settings_fetch_headers>`
 - :ref:`.netrc file <scancodeio_settings_netrc_location>`
+- :ref:`Docker private repository <scancodeio_settings_skopeo_credentials>`
 
 Example for GitHub private repository files::
 
     SCANCODEIO_FETCH_HEADERS="github.com=Authorization=token <YOUR_TOKEN>"
+
+Example for Docker private repository::
+
+    SCANCODEIO_SKOPEO_CREDENTIALS="registry.com=user:password"
diff --git a/scancodeio/settings.py b/scancodeio/settings.py
@@ -145,6 +145,14 @@
     # Propagate the location to the environ for `requests.utils.get_netrc_auth`
     env.ENVIRON["NETRC"] = SCANCODEIO_NETRC_LOCATION
 
+# SCANCODEIO_SKOPEO_CREDENTIALS="host1=user:password,host2=user:password"
+SCANCODEIO_SKOPEO_CREDENTIALS = env.dict("SCANCODEIO_SKOPEO_CREDENTIALS", default={})
+
+# SCANCODEIO_SKOPEO_AUTHFILE_LOCATION="/path/to/auth.json"
+SCANCODEIO_SKOPEO_AUTHFILE_LOCATION = env.str(
+    "SCANCODEIO_SKOPEO_AUTHFILE_LOCATION", default=""
+)
+
 # Application definition
 
 INSTALLED_APPS = [
diff --git a/scanpipe/pipes/fetch.py b/scanpipe/pipes/fetch.py
@@ -179,20 +179,35 @@ def _get_skopeo_location(_cache=[]):
     return cmd_loc
 
 
-def get_docker_image_platform(docker_reference):
+def get_docker_image_platform(docker_url):
     """
     Return a platform mapping of a docker reference.
     If there are more than one, return the first one by default.
     """
     skopeo_executable = _get_skopeo_location()
+
+    authentication_args = []
+    authfile = settings.SCANCODEIO_SKOPEO_AUTHFILE_LOCATION
+    if authfile:
+        authentication_args.append(f"--authfile {authfile}")
+
+    netloc = urlparse(docker_url).netloc
+    if credential := settings.SCANCODEIO_SKOPEO_CREDENTIALS.get(netloc):
+        # Username and password for accessing the registry.
+        authentication_args.append(f"--creds {credential}")
+    elif not authfile:
+        # Access the registry anonymously.
+        authentication_args.append("--no-creds")
+
     cmd_args = (
         str(skopeo_executable),
         "inspect",
         "--insecure-policy",
         "--raw",
-        "--no-creds",
-        docker_reference,
+        *authentication_args,
+        docker_url,
     )
+
     logger.info(f"Fetching image os/arch data: {cmd_args}")
     output = run_command_safely(cmd_args)
     logger.info(output)
@@ -232,28 +247,28 @@ def get_docker_image_platform(docker_reference):
             )
 
 
-def fetch_docker_image(download_url, to=None):
+def fetch_docker_image(docker_url, to=None):
     """
-    Fetch a docker image from the provided Docker image `download_url`,
+    Fetch a docker image from the provided Docker image `docker_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, download_url):
+    if not re.match(whitelist, docker_url):
         raise ValueError("Invalid Docker reference.")
 
-    reference = download_url.replace("docker://", "")
+    reference = docker_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(download_url):
+    if platform := get_docker_image_platform(docker_url):
         os, arch, variant = platform
         if os:
             platform_args.append(f"--override-os={os}")
@@ -262,12 +277,22 @@ def fetch_docker_image(download_url, to=None):
         if variant:
             platform_args.append(f"--override-variant={variant}")
 
+    authentication_args = []
+    if authfile := settings.SCANCODEIO_SKOPEO_AUTHFILE_LOCATION:
+        authentication_args.append(f"--authfile {authfile}")
+
+    netloc = urlparse(docker_url).netloc
+    if credential := settings.SCANCODEIO_SKOPEO_CREDENTIALS.get(netloc):
+        # Credentials for accessing the source registry.
+        authentication_args.append(f"--src-creds {credential}")
+
     cmd_args = (
         str(skopeo_executable),
         "copy",
         "--insecure-policy",
         *platform_args,
-        download_url,
+        *authentication_args,
+        docker_url,
         target,
     )
     logger.info(f"Fetching image with: {cmd_args}")
@@ -277,7 +302,7 @@ def fetch_docker_image(download_url, to=None):
     checksums = multi_checksums(output_file, ("md5", "sha1"))
 
     return Download(
-        uri=download_url,
+        uri=docker_url,
         directory=download_directory,
         filename=filename,
         path=output_file,
diff --git a/scanpipe/tests/pipes/test_fetch.py b/scanpipe/tests/pipes/test_fetch.py
@@ -96,7 +96,7 @@ def test_scanpipe_pipes_fetch_docker_image(
         expected = "Invalid Docker reference."
         self.assertEqual(expected, str(cm.exception))
 
-        url = "docker://debian:10.9"
+        url = "docker://registry.com/debian:10.9"
         mock_platform.return_value = "linux", "amd64", ""
         mock_skopeo.return_value = "skopeo"
         mock_run_command_safely.side_effect = Exception
@@ -112,11 +112,61 @@ def test_scanpipe_pipes_fetch_docker_image(
             "--insecure-policy",
             "--override-os=linux",
             "--override-arch=amd64",
-            "docker://debian:10.9",
+            url,
         )
         self.assertEqual(expected, cmd_args[0:6])
         self.assertTrue(cmd_args[-1].endswith("debian_10_9.tar"))
 
+        with override_settings(SCANCODEIO_SKOPEO_AUTHFILE_LOCATION="auth.json"):
+            with self.assertRaises(Exception):
+                fetch.fetch_docker_image(url)
+            cmd_args = mock_run_command_safely.call_args[0][0]
+            self.assertIn("--authfile auth.json", cmd_args)
+
+        credentials = {"registry.com": "user:password"}
+        with override_settings(SCANCODEIO_SKOPEO_CREDENTIALS=credentials):
+            with self.assertRaises(Exception):
+                fetch.fetch_docker_image(url)
+            cmd_args = mock_run_command_safely.call_args[0][0]
+            self.assertIn("--src-creds user:password", cmd_args)
+
+    @mock.patch("scanpipe.pipes.fetch._get_skopeo_location")
+    @mock.patch("scanpipe.pipes.fetch.run_command_safely")
+    def test_scanpipe_pipes_fetch_get_docker_image_platform(
+        self,
+        mock_run_command_safely,
+        mock_skopeo,
+    ):
+        url = "docker://registry.com/busybox"
+        mock_skopeo.return_value = "skopeo"
+        mock_run_command_safely.return_value = "{}"
+
+        fetch.get_docker_image_platform(url)
+        mock_run_command_safely.assert_called_once()
+        cmd_args = mock_run_command_safely.call_args[0][0]
+        expected = (
+            "skopeo",
+            "inspect",
+            "--insecure-policy",
+            "--raw",
+            "--no-creds",
+            url,
+        )
+        self.assertEqual(expected, cmd_args)
+
+        with override_settings(SCANCODEIO_SKOPEO_AUTHFILE_LOCATION="auth.json"):
+            fetch.get_docker_image_platform(url)
+            cmd_args = mock_run_command_safely.call_args[0][0]
+            self.assertIn("--authfile auth.json", cmd_args)
+            self.assertNotIn("--no-creds", cmd_args)
+
+        credentials = {"registry.com": "user:password"}
+        with override_settings(SCANCODEIO_SKOPEO_CREDENTIALS=credentials):
+            fetch.get_docker_image_platform(url)
+            cmd_args = mock_run_command_safely.call_args[0][0]
+            self.assertIn("--creds user:password", cmd_args)
+            self.assertNotIn("--no-creds", cmd_args)
+
     def test_scanpipe_pipes_fetch_docker_image_string_injection_protection(self):
         url = 'docker://;echo${IFS}"PoC"${IFS}"'
         with self.assertRaises(ValueError) as cm:
