Merge pull request #470 from jacksmith15/feat/461/private-pypi-strip-…

…auth

Implement auth stripping for private PyPi packages

conda_lock/conda_lock.py

@@ -259,6 +259,7 @@ def make_lock_files(  # noqa: C901
     metadata_choices: AbstractSet[MetadataOption] = frozenset(),
     metadata_yamls: Sequence[pathlib.Path] = (),
     with_cuda: Optional[str] = None,
+    strip_auth: bool = False,
 ) -> None:
     """
     Generate a lock file from the src files provided
@@ -385,6 +386,7 @@ def make_lock_files(  # noqa: C901
                 update_spec=update_spec,
                 metadata_choices=metadata_choices,
                 metadata_yamls=metadata_yamls,
+                strip_auth=strip_auth,
             )
 
             if "lock" in kinds:
@@ -685,6 +687,7 @@ def _solve_for_arch(
     platform: str,
     channels: List[Channel],
     update_spec: Optional[UpdateSpecification] = None,
+    strip_auth: bool = False,
 ) -> List[LockedDependency]:
     """
     Solve specification for a single platform
@@ -725,6 +728,7 @@ def _solve_for_arch(
             python_version=conda_deps["python"].version,
             platform=platform,
             allow_pypi_requests=spec.allow_pypi_requests,
+            strip_auth=strip_auth,
         )
     else:
         pip_deps = {}
@@ -769,6 +773,7 @@ def create_lockfile_from_spec(
     update_spec: Optional[UpdateSpecification] = None,
     metadata_choices: AbstractSet[MetadataOption] = frozenset(),
     metadata_yamls: Sequence[pathlib.Path] = (),
+    strip_auth: bool = False,
 ) -> Lockfile:
     """
     Solve or update specification
@@ -785,6 +790,7 @@ def create_lockfile_from_spec(
             platform=platform,
             channels=[*spec.channels, virtual_package_channel],
             update_spec=update_spec,
+            strip_auth=strip_auth,
         )
 
         for dep in deps:
@@ -996,6 +1002,7 @@ def run_lock(
     filter_categories: bool = False,
     metadata_choices: AbstractSet[MetadataOption] = frozenset(),
     metadata_yamls: Sequence[pathlib.Path] = (),
+    strip_auth: bool = False,
 ) -> None:
     if environment_files == DEFAULT_FILES:
         if lockfile_path.exists():
@@ -1047,6 +1054,7 @@ def run_lock(
         filter_categories=filter_categories,
         metadata_choices=metadata_choices,
         metadata_yamls=metadata_yamls,
+        strip_auth=strip_auth,
     )
 
 
@@ -1309,16 +1317,17 @@ def lock(
         filter_categories=filter_categories,
         metadata_choices=metadata_enum_choices,
         metadata_yamls=metadata_yamls,
+        strip_auth=strip_auth,
     )
     if strip_auth:
         with tempfile.TemporaryDirectory() as tempdir:
             filename_template_temp = f"{tempdir}/{filename_template.split('/')[-1]}"
             lock_func(filename_template=filename_template_temp)
             filename_template_dir = "/".join(filename_template.split("/")[:-1])
             for file in os.listdir(tempdir):
-                lockfile = read_file(os.path.join(tempdir, file))
-                lockfile = _strip_auth_from_lockfile(lockfile)
-                write_file(lockfile, os.path.join(filename_template_dir, file))
+                lockfile_content = read_file(os.path.join(tempdir, file))
+                lockfile_content = _strip_auth_from_lockfile(lockfile_content)
+                write_file(lockfile_content, os.path.join(filename_template_dir, file))
     else:
         lock_func(
             filename_template=filename_template, check_input_hash=check_input_hash

conda_lock/pypi_solver.py

@@ -3,7 +3,7 @@
 
 from pathlib import Path
 from typing import TYPE_CHECKING, Dict, List, Optional
-from urllib.parse import urldefrag
+from urllib.parse import urldefrag, urlsplit, urlunsplit
 
 from clikit.api.io.flags import VERY_VERBOSE
 from clikit.io import ConsoleIO, NullIO
@@ -188,6 +188,7 @@ def get_requirements(
     platform: str,
     pool: Pool,
     env: Env,
+    strip_auth: bool = False,
 ) -> List[LockedDependency]:
     """Extract distributions from Poetry package plan, ignoring uninstalls
     (usually: conda package with no pypi equivalent) and skipped ops
@@ -230,7 +231,7 @@ def get_requirements(
                     dependencies={
                         dep.name: str(dep.constraint) for dep in op.package.requires
                     },
-                    url=url,
+                    url=url if not strip_auth else _strip_auth(url),
                     hash=hash,
                 )
             )
@@ -246,6 +247,7 @@ def solve_pypi(
     platform: str,
     allow_pypi_requests: bool = True,
     verbose: bool = False,
+    strip_auth: bool = False,
 ) -> Dict[str, LockedDependency]:
     """
     Solve pip dependencies for the given platform
@@ -270,6 +272,8 @@ def solve_pypi(
         Add pypi.org to the list of repositories (pip packages only)
     verbose :
         Print chatter from solver
+    strip_auth :
+        Whether to strip HTTP Basic auth from URLs.
 
     """
     dummy_package = PoetryProjectPackage("_dummy_package_", "0.0.0")
@@ -330,7 +334,7 @@ def solve_pypi(
     with s.use_environment(env):
         result = s.solve(use_latest=to_update)
 
-    requirements = get_requirements(result, platform, pool, env)
+    requirements = get_requirements(result, platform, pool, env, strip_auth=strip_auth)
 
     # use PyPI names of conda packages to walking the dependency tree and propagate
     # categories from explicit to transitive dependencies
@@ -377,3 +381,12 @@ def _prepare_repositories_pool(allow_pypi_requests: bool) -> Pool:
     if allow_pypi_requests:
         repos.append(PyPiRepository())
     return Pool(repositories=[*repos])
+
+
+def _strip_auth(url: str) -> str:
+    """Strip HTTP Basic authentication from a URL."""
+    parts = urlsplit(url, allow_fragments=True)
+    # Remove everything before and including the last '@' character in the part
+    # between 'scheme://' and the subsequent '/'.
+    netloc = parts.netloc.split("@")[-1]
+    return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))

tests/test_conda_lock.py

@@ -60,7 +60,7 @@
 )
 from conda_lock.models.channel import Channel
 from conda_lock.models.lock_spec import VCSDependency, VersionedDependency
-from conda_lock.pypi_solver import parse_pip_requirement, solve_pypi
+from conda_lock.pypi_solver import _strip_auth, parse_pip_requirement, solve_pypi
 from conda_lock.src_parser import (
     DEFAULT_PLATFORMS,
     LockSpecification,
@@ -1621,6 +1621,39 @@ def test__strip_auth_from_line(line: str, stripped: str):
     assert _strip_auth_from_line(line) == stripped
 
 
+@pytest.mark.parametrize(
+    "url,stripped",
+    (
+        (
+            "https://example.com/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+        (
+            "https://username:[email protected]/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+        (
+            "https://username:@example.com/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+        (
+            "https://:[email protected]/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+        (
+            "https://[email protected]:[email protected]/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+        (
+            "https://username:password@[email protected]/path?query=string#fragment",
+            "https://example.com/path?query=string#fragment",
+        ),
+    ),
+)
+def test_strip_auth_from_url(url: str, stripped: str):
+    assert _strip_auth(url) == stripped
+
+
 @pytest.mark.parametrize(
     "line,stripped",
     (