Ensure HTTP auth persists throughout session

Author: cquick01

src/proxpi/_cache.py

@@ -307,6 +307,22 @@ def __getitem__(self, k: str) -> threading.Lock:
         return self._locks[k]
 
 
+def _parse_basic_auth(url: str) -> t.Tuple[str, str]:
+    """Parse HTTP Basic username and password from URL.
+
+    Args:
+        url: URL to process
+
+    Returns:
+        Tuple containing the username and password
+            (or containing empty strings if the URL has no username and/or password)
+    """
+    parsed = urllib.parse.urlsplit(url)
+    username = parsed.username if parsed.username else ""
+    password = parsed.password if parsed.password else ""
+    return username, password
+
+
 def _mask_password(url: str) -> str:
     """Mask HTTP basic auth password in URL.
 
@@ -359,17 +375,31 @@ def __init__(self, index_url: str, ttl: int, session: requests.Session = None):
         self._index = {}
         self._packages = {}
         self._index_url_masked = _mask_password(index_url)
+        username, password = _parse_basic_auth(index_url)
+        if username:
+            # password either supplied or empty str
+            self._auth = (username, password)
+        else:
+            self._auth = None
 
     def __repr__(self):
         return f"{self.__class__.__name__}({self._index_url_masked!r}, {self.ttl!r})"
 
+    def get(self, url, **kwargs):
+        """Wrapper for self.session.get to configure auth if provided."""
+        if self._auth:
+            response = self.session.get(url, auth=self._auth, **kwargs)
+        else:
+            response = self.session.get(url, **kwargs)
+        return response
+
     def _list_packages(self):
         """List projects using or updating cache."""
         if self._index_t is not None and _now() < self._index_t + self.ttl:
             return
 
         logger.info(f"Listing packages in index '{self._index_url_masked}'")
-        response = self.session.get(self.index_url, headers=self._headers)
+        response = self.get(self.index_url, headers=self._headers)
         response.raise_for_status()
         self._index_t = _now()
 
@@ -430,15 +460,15 @@ def _list_files(self, package_name: str):
         if self._index_t is None or _now() > self._index_t + self.ttl:
             url = urllib.parse.urljoin(self.index_url, package_name)
             logger.debug(f"Refreshing '{package_name}'")
-            response = self.session.get(url, headers=self._headers)
+            response = self.get(url, headers=self._headers)
         if not response or not response.ok:
             logger.debug(f"List-files response: {response}")
             package_name_normalised = _name_normalise_re.sub("-", package_name).lower()
             if package_name_normalised not in self.list_projects():
                 raise NotFound(package_name)
             package_url = self._index[package_name]
             url = urllib.parse.urljoin(self.index_url, package_url)
-            response = self.session.get(url, headers=self._headers)
+            response = self.get(url, headers=self._headers)
             response.raise_for_status()
 
         package = Package(package_name, files={}, refreshed=_now())

tests/_utils.py

@@ -67,11 +67,22 @@ def run(self):
             self.exc = e
 
 
-def make_server(app: "flask.Flask") -> t.Generator[str, None, None]:
+def make_server(
+    app: "flask.Flask",
+    auth: t.Tuple[t.Optional[str], t.Optional[str]] = (None, None),
+) -> t.Generator[str, None, None]:
     server = werkzeug.serving.make_server(host="localhost", port=0, app=app)
     thread = Thread(target=server.serve_forever, args=(0.05,))
     thread.start()
-    yield f"http://localhost:{server.port}"
+    username, password = auth
+    creds = ""
+    if username is not None:
+        creds = username
+        if password is not None:
+            creds += f":{password}"
+        creds += "@"
+
+    yield f"http://{creds}localhost:{server.port}"
     server.shutdown()
     thread.join(timeout=0.1)
     if thread.exc:

tests/test_integration.py

@@ -26,6 +26,12 @@
 
 mock_index_response_is_json = False
 
+CREDENTIALS = [
+    ("username", "password"),
+    ("username", None),
+    (None, None),
+]
+
 
 @contextlib.contextmanager
 def set_mock_index_response_is_json(value: t.Union[bool, str]):
@@ -94,20 +100,25 @@ def get_file(project_name: str, file_name: str) -> flask.Response:
     return app
 
 
+@pytest.fixture(scope="module", params=CREDENTIALS)
+def mock_auth(request):
+    return request.param
+
+
 @pytest.fixture(scope="module")
-def mock_root_index():
+def mock_root_index(mock_auth):
     app = make_mock_index_app(index_name="root")
-    yield from _utils.make_server(app)
+    yield from _utils.make_server(app, mock_auth)
 
 
 @pytest.fixture(scope="module")
-def mock_extra_index():
+def mock_extra_index(mock_auth):
     app = make_mock_index_app(index_name="extra")
-    yield from _utils.make_server(app)
+    yield from _utils.make_server(app, mock_auth)
 
 
 @pytest.fixture(scope="module")
-def server(mock_root_index, mock_extra_index):
+def server(mock_root_index, mock_extra_index, mock_auth):
     session = proxpi.server.cache.root_cache.session
     # noinspection PyProtectedMember
     root_patch = mock.patch.object(
@@ -122,7 +133,7 @@ def server(mock_root_index, mock_extra_index):
         [proxpi.server.cache._index_cache_cls(f"{mock_extra_index}/", 10, session)],
     )
     with root_patch, extras_patch:
-        yield from _utils.make_server(proxpi_server.app)
+        yield from _utils.make_server(proxpi_server.app, mock_auth)
 
 
 @pytest.fixture