Add authentication for private repos

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

setup.cfg

@@ -66,6 +66,7 @@ install_requires =
     requests >= 2.7.0
     resolvelib
     saneyaml >= 0.5.2
+    tinynetrc
     toml >= 0.10.0
 
 [options.packages.find]

src/python_inspector/resolve_cli.py

@@ -14,8 +14,10 @@
 
 import click
 from packaging.requirements import Requirement
+from tinynetrc import Netrc
 
 from python_inspector import dependencies
+from python_inspector import utils
 from python_inspector import utils_pypi
 from python_inspector.cli_utils import FileOptionType
 from python_inspector.resolution import get_resolved_dependencies
@@ -40,6 +42,15 @@
     help="Path to pip requirements file listing thirdparty packages. "
     "This option can be used multiple times.",
 )
+@click.option(
+    "-n",
+    "--netrc",
+    "netrc_file",
+    type=click.Path(exists=True, readable=True, path_type=str, dir_okay=False),
+    metavar="NETRC-FILE",
+    required=False,
+    help="Netrc file to use for authentication. ",
+)
 @click.option(
     "--spec",
     "--specifier",
@@ -111,6 +122,7 @@
 @click.help_option("-h", "--help")
 def resolve_dependencies(
     requirement_files,
+    netrc_file,
     specifiers,
     python_version,
     operating_system,
@@ -141,6 +153,9 @@ def resolve_dependencies(
 
     click.secho(f"Resolving dependencies...")
 
+    netrc = None
+    if netrc_file:
+        netrc = Netrc(file=netrc_file)
     # TODO: deduplicate me
     direct_dependencies = []
 
@@ -179,9 +194,16 @@ def resolve_dependencies(
                 existing.use_cached_index = use_cached_index
                 repos.append(existing)
             else:
+                credentials = None
+                if netrc:
+                    login, password = utils.get_netrc_auth(index_url, netrc)
+                    credentials = (
+                        dict(login=login, password=password) if login and password else None
+                    )
                 repo = utils_pypi.PypiSimpleRepository(
                     index_url=index_url,
                     use_cached_index=use_cached_index,
+                    credentials=credentials,
                 )
                 repos.append(repo)
 

src/python_inspector/utils.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# ScanCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/python-inspector for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+def get_netrc_auth(url, netrc):
+    """
+    Return login and password if url is in netrc
+    else return login and password as None
+    """
+    if netrc.get(url):
+        return (netrc[url].get("login"), netrc[url].get("password"))
+    return (None, None)

src/python_inspector/utils_pypi.py

@@ -494,6 +494,11 @@ class Distribution(NameVer):
         metadata=dict(help="Extra data"),
     )
 
+    credentials = attr.ib(
+        type=dict,
+        default=None,
+    )
+
     @property
     def package_url(self):
         """
@@ -565,6 +570,7 @@ def download(
         fetch_and_save(
             path_or_url=self.path_or_url,
             dest_dir=dest_dir,
+            credentials=self.credentials,
             filename=self.filename,
             as_text=False,
             verbose=verbose,
@@ -1387,6 +1393,8 @@ class PypiSimpleRepository:
         repr=False,
     )
 
+    credentials = attr.ib(type=dict, default=None)
+
     def _get_package_versions_map(
         self,
         name,
@@ -1483,6 +1491,7 @@ def fetch_links(
         package_url = f"{self.index_url}/{normalized_name}"
         text = CACHE.get(
             path_or_url=package_url,
+            credentials=self.credentials,
             as_text=True,
             force=not self.use_cached_index,
             verbose=verbose,
@@ -1520,6 +1529,7 @@ def __attrs_post_init__(self):
 
     def get(
         self,
+        credentials,
         path_or_url,
         as_text=True,
         force=False,
@@ -1540,6 +1550,7 @@ def get(
                 print(f"        FILE CACHE MISS: {path_or_url}")
             content = get_file_content(
                 path_or_url=path_or_url,
+                credentials=credentials,
                 as_text=as_text,
                 verbose=verbose,
                 echo_func=echo_func,
@@ -1559,6 +1570,7 @@ def get(
 
 def get_file_content(
     path_or_url,
+    credentials,
     as_text=True,
     verbose=False,
     echo_func=None,
@@ -1572,6 +1584,7 @@ def get_file_content(
             print(f"Fetching: {path_or_url}")
         _headers, content = get_remote_file_content(
             url=path_or_url,
+            credentials=credentials,
             as_text=as_text,
             verbose=verbose,
             echo_func=echo_func,
@@ -1606,6 +1619,7 @@ class RemoteNotFetchedException(Exception):
 
 def get_remote_file_content(
     url,
+    credentials,
     as_text=True,
     headers_only=False,
     headers=None,
@@ -1631,7 +1645,20 @@ def get_remote_file_content(
         echo_func = print
     if verbose:
         echo_func(f"DOWNLOADING: {url}")
-    with requests.get(url, allow_redirects=True, stream=True, headers=headers) as response:
+
+    auth = None
+    if credentials:
+        auth = (credentials.get("login"), credentials.get("password"))
+
+    stream = requests.get(
+        url,
+        allow_redirects=True,
+        stream=True,
+        headers=headers,
+        auth=auth,
+    )
+
+    with stream as response:
         status = response.status_code
         if status != requests.codes.ok:  # NOQA
             if status == 429 and _delay < 20:
@@ -1640,6 +1667,7 @@ def get_remote_file_content(
 
                 return get_remote_file_content(
                     url,
+                    credentials=credentials,
                     as_text=as_text,
                     headers_only=headers_only,
                     _delay=increased_delay,
@@ -1658,6 +1686,7 @@ def fetch_and_save(
     path_or_url,
     dest_dir,
     filename,
+    credentials,
     as_text=True,
     verbose=False,
     echo_func=None,
@@ -1670,6 +1699,7 @@ def fetch_and_save(
     """
     content = CACHE.get(
         path_or_url=path_or_url,
+        credentials=credentials,
         as_text=as_text,
         verbose=verbose,
         echo_func=echo_func,

tests/data/test.netrc

@@ -0,0 +1 @@
+machine https://pyp1.org/simple login test password test123

tests/test_cli.py

@@ -74,6 +74,27 @@ def test_cli_with_multiple_index_url_and_tilde_req():
     )
 
 
+@pytest.mark.online
+def test_cli_with_multiple_index_url_and_tilde_req_and_netrc_file_without_matching_url():
+    expected_file = test_env.get_test_loc("tilde_req-expected.json", must_exist=False)
+    netrc_file = test_env.get_test_loc("test.netrc", must_exist=False)
+    specifier = "zipp~=3.8.0"
+    extra_options = [
+        "--index-url",
+        "https://pypi.org/simple",
+        "--index-url",
+        "https://thirdparty.aboutcode.org/pypi/simple/",
+        "--netrc",
+        netrc_file,
+    ]
+    check_specs_resolution(
+        specifier=specifier,
+        expected_file=expected_file,
+        extra_options=extra_options,
+        regen=REGEN_TEST_FIXTURES,
+    )
+
+
 @pytest.mark.online
 def test_cli_with_pinned_requirements_file():
     requirements_file = test_env.get_test_loc("pinned-requirements.txt")

tests/test_utils.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# ScanCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/python-inspector for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+import os
+
+from commoncode.testcase import FileDrivenTesting
+from tinynetrc import Netrc
+
+from python_inspector.utils import get_netrc_auth
+
+test_env = FileDrivenTesting()
+test_env.test_data_dir = os.path.join(os.path.dirname(__file__), "data")
+
+
+def test_get_netrc_auth():
+    netrc_file = test_env.get_test_loc("test.netrc")
+    netrc = Netrc(netrc_file)
+    assert get_netrc_auth(url="https://pyp1.org/simple", netrc=netrc) == ("test", "test123")
+
+
+def test_get_netrc_auth_with_no_matching_url():
+    netrc_file = test_env.get_test_loc("test.netrc")
+    netrc = Netrc(netrc_file)
+    assert get_netrc_auth(url="https://pypi2.org/simple", netrc=netrc) == (None, None)