Add support for fetching authentications #620 (#1097)

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -48,6 +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
+  define credentials for various authentication types.
+  https://github.com/nexB/scancode.io/issues/620
+
 - Update matchcode-toolkit to v3.0.0
 
 v33.1.0 (2024-02-02)

docs/application-settings.rst

@@ -313,3 +313,63 @@ However, if authentication is enabled on your VulnerableCode instance,
 you can provide the API key using ``VULNERABLECODE_API_KEY``::
 
     VULNERABLECODE_API_KEY=insert_your_api_key_here
+
+.. _scancodeio_settings_fetch_authentication:
+
+Fetch Authentication
+--------------------
+
+Several settings are available to define the credentials required to access your
+private files, depending on the authentication type: Basic, Digest, Token header, etc.
+
+.. note::
+    The provided credentials are enabled for all projects on the ScanCode.io instance.
+
+.. warning::
+    Ensure that the provided ``host`` values are fully qualified, including the domain
+    and subdomain.
+
+.. _scancodeio_settings_fetch_basic_auth:
+
+SCANCODEIO_FETCH_BASIC_AUTH
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+You can provide credentials for input URLs protected by Basic Authentication using
+the ``host=user,password`` syntax::
+
+    SCANCODEIO_FETCH_BASIC_AUTH="www.host1.com=user,password;www.host2.com=user,password;"
+
+.. _scancodeio_settings_fetch_digest_auth:
+
+SCANCODEIO_FETCH_DIGEST_AUTH
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+You can provide credentials for input URLs protected by Digest Authentication using
+the ``host=user,password`` syntax::
+
+    SCANCODEIO_FETCH_DIGEST_AUTH="www.host1.com=user,password;www.host2.com=user,password;"
+
+.. _scancodeio_settings_fetch_headers:
+
+SCANCODEIO_FETCH_HEADERS
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+When authentication credentials can be provided through HTTP request headers, you can
+use the following syntax::
+
+    SCANCODEIO_FETCH_HEADERS="www.host1.com=Header1=value,Header2=value;"
+
+Example for a GitHub private repository::
+
+    SCANCODEIO_FETCH_HEADERS="raw.github.com=Authorization=token <YOUR_TOKEN>"
+
+.. _scancodeio_settings_netrc_location:
+
+SCANCODEIO_NETRC_LOCATION
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+If your credentials are stored in a
+`.netrc <https://everything.curl.dev/usingcurl/netrc>`_ file, you can provide its
+location on disk using::
+
+    SCANCODEIO_NETRC_LOCATION="~/.netrc"

docs/faq.rst

@@ -197,3 +197,19 @@ There are multiple ways to tag input files when uploading local files:
 
 - **Command Line Interface:** Tag uploaded files using the "filename:tag" syntax.
   Example: ``--input-file path/filename:tag``.
+
+How to fetch files from private sources and protected by credentials?
+---------------------------------------------------------------------
+
+Several :ref:`scancodeio_settings_fetch_authentication` settings are available to
+define the credentials required to access your private files, depending on the
+authentication type:
+
+- :ref:`Basic authentication <scancodeio_settings_fetch_basic_auth>`
+- :ref:`Digest authentication <scancodeio_settings_fetch_digest_auth>`
+- :ref:`HTTP request headers <scancodeio_settings_fetch_headers>`
+- :ref:`.netrc file <scancodeio_settings_netrc_location>`
+
+Example for GitHub private repository files::
+
+    SCANCODEIO_FETCH_HEADERS="github.com=Authorization=token <YOUR_TOKEN>"

scancodeio/settings.py

@@ -115,6 +115,36 @@
 # Default limit for "most common" entries in QuerySets.
 SCANCODEIO_MOST_COMMON_LIMIT = env.int("SCANCODEIO_MOST_COMMON_LIMIT", default=7)
 
+# Fetch authentication credentials
+
+# SCANCODEIO_FETCH_BASIC_AUTH="host=user,password;"
+SCANCODEIO_FETCH_BASIC_AUTH = env.dict(
+    "SCANCODEIO_FETCH_BASIC_AUTH",
+    cast={"value": tuple},
+    default={},
+)
+
+# SCANCODEIO_FETCH_DIGEST_AUTH="host=user,password;"
+SCANCODEIO_FETCH_DIGEST_AUTH = env.dict(
+    "SCANCODEIO_FETCH_DIGEST_AUTH",
+    cast={"value": tuple},
+    default={},
+)
+
+# SCANCODEIO_FETCH_HEADERS="host=Header1=value,Header2=value;"
+SCANCODEIO_FETCH_HEADERS = {}
+FETCH_HEADERS_STR = env.str("SCANCODEIO_FETCH_HEADERS", default="")
+for entry in FETCH_HEADERS_STR.split(";"):
+    if entry.strip():
+        host, headers = entry.split("=", 1)
+        SCANCODEIO_FETCH_HEADERS[host] = env.parse_value(headers, cast=dict)
+
+# SCANCODEIO_NETRC_LOCATION="~/.netrc"
+SCANCODEIO_NETRC_LOCATION = env.str("SCANCODEIO_NETRC_LOCATION", default="")
+if SCANCODEIO_NETRC_LOCATION:
+    # Propagate the location to the environ for `requests.utils.get_netrc_auth`
+    env.ENVIRON["NETRC"] = SCANCODEIO_NETRC_LOCATION
+
 # Application definition
 
 INSTALLED_APPS = [

scanpipe/api/views.py

@@ -343,7 +343,7 @@ def reset(self, request, *args, **kwargs):
         project = self.get_object()
 
         if self.request.method == "GET":
-            message = "POST on this URL to reset the project. " ""
+            message = "POST on this URL to reset the project."
             return Response({"status": message})
 
         try:

scanpipe/pipes/fetch.py

@@ -32,11 +32,14 @@
 from urllib.parse import unquote
 from urllib.parse import urlparse
 
+from django.conf import settings
+
 import requests
 from commoncode import command
 from commoncode.hash import multi_checksums
 from commoncode.text import python_safe_name
 from plugincode.location_provider import get_location
+from requests import auth as request_auth
 
 logger = logging.getLogger("scanpipe.pipes")
 
@@ -70,12 +73,30 @@ def run_command_safely(command_args):
     return result.stdout
 
 
+def get_request_session(uri):
+    """Return a Requests session setup with authentication and headers."""
+    session = requests.Session()
+    netloc = urlparse(uri).netloc
+
+    if credentials := settings.SCANCODEIO_FETCH_BASIC_AUTH.get(netloc):
+        session.auth = request_auth.HTTPBasicAuth(*credentials)
+
+    elif credentials := settings.SCANCODEIO_FETCH_DIGEST_AUTH.get(netloc):
+        session.auth = request_auth.HTTPDigestAuth(*credentials)
+
+    if headers := settings.SCANCODEIO_FETCH_HEADERS.get(netloc):
+        session.headers.update(headers)
+
+    return session
+
+
 def fetch_http(uri, to=None):
     """
     Download a given `uri` in a temporary directory and return the directory's
     path.
     """
-    response = requests.get(uri, timeout=5)
+    request_session = get_request_session(uri)
+    response = request_session.get(uri, timeout=5)
 
     if response.status_code != 200:
         raise requests.RequestException
@@ -329,8 +350,9 @@ def check_urls_availability(urls):
         if not url.startswith("http"):
             continue
 
+        request_session = get_request_session(url)
         try:
-            response = requests.head(url, timeout=3)
+            response = request_session.head(url, timeout=5)
             response.raise_for_status()
         except requests.exceptions.RequestException:
             errors.append(url)

scanpipe/tests/pipes/test_fetch.py

@@ -24,6 +24,9 @@
 from unittest import mock
 
 from django.test import TestCase
+from django.test import override_settings
+
+from requests import auth as request_auth
 
 from scanpipe.pipes import fetch
 
@@ -56,7 +59,7 @@ def test_scanpipe_pipes_fetch_get_fetcher(self):
         expected = "URL scheme 'DOCKER' is not supported. Did you mean: 'docker'?"
         self.assertEqual(expected, str(cm.exception))
 
-    @mock.patch("requests.get")
+    @mock.patch("requests.sessions.Session.get")
     def test_scanpipe_pipes_fetch_http(self, mock_get):
         url = "https://example.com/filename.zip"
 
@@ -120,7 +123,7 @@ def test_scanpipe_pipes_fetch_docker_image_string_injection_protection(self):
             fetch.fetch_docker_image(url)
         self.assertEqual("Invalid Docker reference.", str(cm.exception))
 
-    @mock.patch("requests.get")
+    @mock.patch("requests.sessions.Session.get")
     def test_scanpipe_pipes_fetch_fetch_urls(self, mock_get):
         urls = [
             "https://example.com/filename.zip",
@@ -141,3 +144,26 @@ def test_scanpipe_pipes_fetch_fetch_urls(self, mock_get):
         self.assertEqual(0, len(downloads))
         self.assertEqual(2, len(errors))
         self.assertEqual(urls, errors)
+
+    def test_scanpipe_pipes_fetch_get_request_session(self):
+        url = "https://example.com/filename.zip"
+        host = "example.com"
+        credentials = ("user", "pass")
+
+        session = fetch.get_request_session(url)
+        self.assertIsNone(session.auth)
+
+        with override_settings(SCANCODEIO_FETCH_BASIC_AUTH={host: credentials}):
+            session = fetch.get_request_session(url)
+            self.assertEqual(request_auth.HTTPBasicAuth(*credentials), session.auth)
+
+        with override_settings(SCANCODEIO_FETCH_DIGEST_AUTH={host: credentials}):
+            session = fetch.get_request_session(url)
+            self.assertEqual(request_auth.HTTPDigestAuth(*credentials), session.auth)
+
+        headers = {
+            host: {"Authorization": "token TOKEN"},
+        }
+        with override_settings(SCANCODEIO_FETCH_HEADERS=headers):
+            session = fetch.get_request_session(url)
+            self.assertEqual("token TOKEN", session.headers.get("Authorization"))

scanpipe/tests/test_commands.py

@@ -199,7 +199,7 @@ def test_scanpipe_management_command_add_input_file(self):
         with self.assertRaisesMessage(CommandError, expected):
             call_command("add-input", *options, stdout=out)
 
-    @mock.patch("requests.get")
+    @mock.patch("requests.sessions.Session.get")
     def test_scanpipe_management_command_add_input_url(self, mock_get):
         mock_get.side_effect = None
         mock_get.return_value = mock.Mock(

scanpipe/tests/test_forms.py

@@ -38,7 +38,7 @@ class ScanPipeFormsTest(TestCase):
     def setUp(self):
         self.project1 = Project.objects.create(name="Analysis")
 
-    @mock.patch("requests.head")
+    @mock.patch("requests.sessions.Session.head")
     def test_scanpipe_forms_inputs_base_form_input_urls(self, mock_head):
         data = {
             "input_urls": "Docker://debian",

scanpipe/tests/test_models.py

@@ -1207,7 +1207,7 @@ def test_scanpipe_input_source_model_delete_file(self):
         input_source.delete_file()
         self.assertFalse(input_source.exists())
 
-    @mock.patch("requests.get")
+    @mock.patch("requests.sessions.Session.get")
     def test_scanpipe_input_source_model_fetch(self, mock_get):
         download_url = "https://download.url/file.zip"
         mock_get.return_value = mock.Mock(