+netrc for authentication

morgan/__init__.py

@@ -3,7 +3,6 @@
 import argparse
 import configparser
 import hashlib
-import json
 import os
 import os.path
 import re
@@ -25,6 +24,7 @@
 from morgan.__about__ import __version__
 from morgan.utils import (
     Cache,
+    Index,
     ListExtendingOrderedDict,
     is_requirement_relevant,
     to_single_dash,
@@ -60,6 +60,7 @@ def __init__(self, args: argparse.Namespace):
             dict_type=ListExtendingOrderedDict,
         )
         self.config.read(args.config)
+        self.index = Index(args)
         self.envs = {}
         self._supported_pyversions = []
         self._supported_platforms = []
@@ -158,24 +159,7 @@ def _mirror(  # noqa: C901, PLR0912
         else:
             print(f"{requirement}")
 
-        data: dict | None = None
-
-        # get information about this package from the Simple API in JSON
-        # format as per PEP 691
-        request = urllib.request.Request(  # noqa: S310
-            f"{self.index_url}{requirement.name}/",
-            headers={
-                "Accept": "application/vnd.pypi.simple.v1+json",
-            },
-        )
-
-        response_url = ""
-        with urllib.request.urlopen(request) as response:  # noqa: S310
-            data = json.load(response)
-            response_url = str(response.url)
-            if not data:
-                msg = f"Failed loading metadata: {response}"
-                raise RuntimeError(msg)
+        data, response_url = self.index.get(requirement.name)
 
         # check metadata version ~1.0
         v_str = data["meta"]["api-version"]
@@ -449,7 +433,7 @@ def _matches_environments(  # noqa: C901, PLR0912
         if fileinfo.get("tags"):
             # At least one of the tags must match ALL of our environments
             for tag in fileinfo["tags"]:
-                (intrp_name, intrp_ver) = parse_interpreter(tag.interpreter)
+                intrp_name, intrp_ver = parse_interpreter(tag.interpreter)
                 if intrp_name not in ("py", "cp"):
                     continue
 
@@ -722,11 +706,7 @@ def mirror(args: argparse.Namespace):
         m.copy_server()
 
 
-def main():  # noqa: C901
-    """
-    Executes the command line interface of Morgan. Use -h for a full list of
-    flags, options and arguments.
-    """
+def create_arg_parser() -> argparse.ArgumentParser:
 
     def my_url(arg):
         # url -> url/ without params
@@ -802,6 +782,7 @@ def my_url(arg):
 
     server.add_arguments(parser)
     configurator.add_arguments(parser)
+    Index.add_arguments(parser)
 
     parser.add_argument(
         "command",
@@ -816,6 +797,16 @@ def my_url(arg):
         help="Command to execute",
     )
 
+    return parser
+
+
+def main():
+    """
+    Executes the command line interface of Morgan. Use -h for a full list of
+    flags, options and arguments.
+    """
+
+    parser = create_arg_parser()
     args = parser.parse_args()
 
     # These commands do not require a configuration file and therefore should

morgan/utils.py

@@ -1,13 +1,20 @@
 from __future__ import annotations
 
+import base64
+import json
+import netrc
 import os
 import re
+import urllib.parse
+import urllib.request
 from collections import OrderedDict
 from typing import TYPE_CHECKING, Iterable
 
 import dateutil  # type: ignore[import-untyped]
 
 if TYPE_CHECKING:
+    import argparse
+
     from packaging.requirements import Requirement
 
 
@@ -48,6 +55,7 @@ def is_simple_case(self, req: Requirement) -> bool:
             if not specifier:
                 return True
             # ruff: noqa: SLF001
+            # pylint: disable=protected-access
             if all(spec.operator in (">", ">=") for spec in specifier._specs):
                 return True
         return False
@@ -151,3 +159,57 @@ def __setitem__(self, key, value):
             self[key].extend(value)
         else:
             super().__setitem__(key, value)
+
+
+class Index:
+    def __init__(self, args: argparse.Namespace):
+        self.index_url = args.index_url.rstrip("/")
+        self.auth_header: tuple[str, str] | None = None
+        netrc_ = None
+        if args.netrc_file:
+            netrc_ = netrc.netrc(args.netrc_file)
+        elif args.netrc:
+            netrc_ = netrc.netrc()
+        if netrc_:
+            hostname = urllib.parse.urlsplit(self.index_url).hostname
+            t = netrc_.authenticators(hostname)  # login, account, passwd
+            if t:
+                # ruff: noqa: UP012
+                s = base64.b64encode(f"{t[0]}:{t[2]}".encode("utf-8")).decode("ascii")
+                self.auth_header = ("Authorization", f"Basic {s}")
+
+    def get(self, pkg_name: str) -> tuple[dict, str]:
+        # get information about this package from the Simple API in JSON
+        # format as per PEP 691
+        request = urllib.request.Request(  # noqa: S310
+            f"{self.index_url}/{pkg_name}/",
+            headers={
+                "Accept": "application/vnd.pypi.simple.v1+json",
+            },
+        )
+        if self.auth_header:
+            request.add_header(*self.auth_header)
+
+        with urllib.request.urlopen(request) as response:  # noqa: S310
+            data = json.load(response)
+            response_url = str(response.url)
+            if data:
+                return (data, response_url)
+            msg = f"Failed loading metadata: {response}"
+            raise RuntimeError(msg)
+
+    @staticmethod
+    def add_arguments(parser: argparse.ArgumentParser):
+        parser.add_argument(
+            "--netrc",
+            dest="netrc",
+            action="store_true",
+            help="Must read .netrc for username and password",
+        )
+
+        parser.add_argument(
+            "--netrc-file",
+            dest="netrc_file",
+            nargs="?",
+            help="Specify FILE for netrc",
+        )

pyproject.toml

@@ -93,3 +93,17 @@ ignore = [
 [tool.mypy]
 check_untyped_defs = true
 follow_untyped_imports = true
+
+[tool.isort]
+profile = "black"
+
+[tool.pylint]
+"messages control".disable = [
+    "line-too-long",
+    "missing-class-docstring",
+    "missing-function-docstring",
+    "missing-module-docstring",
+    "too-few-public-methods",
+    "too-many-instance-attributes",
+]
+reports.score = false

tests/test_init.py

@@ -1,15 +1,20 @@
 # pylint: disable=missing-function-docstring,missing-class-docstring,missing-module-docstring
 from __future__ import annotations
 
-import argparse
 import hashlib
 import os
 
 import packaging.requirements
 import packaging.version
 import pytest
 
-from morgan import PYPI_ADDRESS, Mirrorer, parse_interpreter, parse_requirement, server
+from morgan import (
+    Mirrorer,
+    create_arg_parser,
+    parse_interpreter,
+    parse_requirement,
+    server,
+)
 
 
 class TestParseInterpreter:
@@ -88,13 +93,14 @@ def temp_index_path(self, tmpdir):
         return tmpdir
 
     def test_mirrorer_initialization(self, temp_index_path):
-        args = argparse.Namespace(
-            index_path=temp_index_path,
-            index_url="https://pypi.org/simple/",
-            config=os.path.join(temp_index_path, "morgan.ini"),
-            mirror_all_versions=False,
-            package_type_regex="(whl|zip|tar.gz)",
-            mirror_all_wheels=False,
+        args = create_arg_parser().parse_args(
+            [
+                "mirror",
+                "--index-path",
+                str(temp_index_path),
+                "--config",
+                os.path.join(temp_index_path, "morgan.ini"),
+            ],
         )
 
         mirrorer = Mirrorer(args)
@@ -108,13 +114,14 @@ def test_mirrorer_initialization(self, temp_index_path):
         assert not mirrorer.mirror_all_versions
 
     def test_server_file_copying(self, temp_index_path):
-        args = argparse.Namespace(
-            index_path=temp_index_path,
-            index_url=PYPI_ADDRESS,
-            config=os.path.join(temp_index_path, "morgan.ini"),
-            mirror_all_versions=False,
-            package_type_regex="(whl|zip|tar.gz)",
-            mirror_all_wheels=False,
+        args = create_arg_parser().parse_args(
+            [
+                "mirror",
+                "--index-path",
+                str(temp_index_path),
+                "--config",
+                os.path.join(temp_index_path, "morgan.ini"),
+            ],
         )
         mirrorer = Mirrorer(args)
 
@@ -134,13 +141,14 @@ def test_server_file_copying(self, temp_index_path):
             )
 
     def test_file_hashing(self, temp_index_path):
-        args = argparse.Namespace(
-            index_path=temp_index_path,
-            index_url=PYPI_ADDRESS,
-            config=os.path.join(temp_index_path, "morgan.ini"),
-            mirror_all_versions=False,
-            package_type_regex="(whl|zip|tar.gz)",
-            mirror_all_wheels=False,
+        args = create_arg_parser().parse_args(
+            [
+                "mirror",
+                "--index-path",
+                str(temp_index_path),
+                "--config",
+                os.path.join(temp_index_path, "morgan.ini"),
+            ],
         )
         mirrorer = Mirrorer(args)
 
@@ -184,14 +192,20 @@ def temp_index_path(self, tmp_path):
     def make_mirrorer(self, temp_index_path):
         # Return a function that creates mirrorer instances
         def _make_mirrorer(mirror_all_versions, mirror_all_wheels=False):
-            args = argparse.Namespace(
-                index_path=temp_index_path,
-                index_url="https://example.com/simple",
-                config=os.path.join(temp_index_path, "morgan.ini"),
-                mirror_all_versions=mirror_all_versions,
-                package_type_regex=r"(whl|zip|tar\.gz)",
-                mirror_all_wheels=mirror_all_wheels,
-            )
+            args_list = [
+                "mirror",
+                "--index-path",
+                str(temp_index_path),
+                "--index-url",
+                "https://example.com/simple",
+                "--config",
+                os.path.join(temp_index_path, "morgan.ini"),
+            ]
+            if mirror_all_versions:
+                args_list.append("--mirror-all-versions")
+            if mirror_all_wheels:
+                args_list.append("--mirror-all-wheels")
+            args = create_arg_parser().parse_args(args_list)
             return Mirrorer(args)
 
         return _make_mirrorer

tests/test_init_wheelscore.py

@@ -3,13 +3,12 @@
 # pylint: disable=missing-function-docstring,missing-class-docstring,protected-access
 # ruff: noqa: SLF001
 
-import argparse
 import os
 from typing import NamedTuple
 
 import pytest
 
-from morgan import Mirrorer
+from morgan import Mirrorer, create_arg_parser
 
 
 class TestCalculateScoresForWheel:
@@ -32,13 +31,17 @@ def temp_index_path(self, tmp_path):
 
     @pytest.fixture
     def mirrorer(self, temp_index_path):
-        args = argparse.Namespace(
-            index_path=temp_index_path,
-            index_url="https://example.com/simple/",
-            config=os.path.join(temp_index_path, "morgan.ini"),
-            mirror_all_versions=False,
-            package_type_regex=r"(whl|zip|tar\.gz)",
-            mirror_all_wheels=True,
+        args = create_arg_parser().parse_args(
+            [
+                "mirror",
+                "--index-path",
+                str(temp_index_path),
+                "--index-url",
+                "https://example.com/simple",
+                "--config",
+                os.path.join(temp_index_path, "morgan.ini"),
+                "--mirror-all-wheels",
+            ],
         )
         return Mirrorer(args)