Handle auth failure when in locking and installation

pdm/models/auth.py

+from typing import List, Optional, Tuple
+
+from pdm._types import Source
+from pdm.exceptions import PdmException
+from pdm.models.pip_shims import MultiDomainBasicAuth
+from pdm.utils import expand_env_vars
+
+
+class PdmBasicAuth(MultiDomainBasicAuth):
+    """A custom auth class that differs from Pip's implementation in the
+    following ways:
+
+        1. It expands env variables in URL auth.
+        2. It shows an error message when credentials are not provided or correect.
+    """
+
+    def _get_url_and_credentials(
+        self, original_url: str
+    ) -> Tuple[str, Optional[str], Optional[str]]:
+        url, username, password = super()._get_url_and_credentials(original_url)
+
+        if username:
+            username = expand_env_vars(username)
+
+        if password:
+            password = expand_env_vars(password)
+
+        return url, username, password
+
+    def handle_401(self, resp, **kwargs):
+        if resp.status_code == 401 and not self.prompting:
+            raise PdmException(
+                f"The credentials for {resp.request.url} are not provided or correct. "
+                "Please run the command with `-v` option."
+            )
+        return super().handle_401(resp, **kwargs)
+
+
+def make_basic_auth(sources: List[Source], prompting: bool) -> PdmBasicAuth:
+    return PdmBasicAuth(prompting, [source["url"] for source in sources])

pdm/models/environment.py

@@ -8,7 +8,7 @@ import sys
 import sysconfig
 from contextlib import contextmanager
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, Generator, Iterator, List, Optional, Tuple
 
 from distlib.scripts import ScriptMaker
 from pip._internal.req import req_uninstall
@@ -20,6 +20,7 @@ from pythonfinder.environment import PYENV_INSTALLED, PYENV_ROOT
 from pdm.exceptions import NoPythonVersion
 from pdm.iostream import stream
 from pdm.models import pip_shims
+from pdm.models.auth import make_basic_auth
 from pdm.models.builders import EnvBuilder
 from pdm.models.in_process import (
     get_pep508_environment,
@@ -92,6 +93,9 @@ class Environment:
         self.python_requires = project.python_requires
         self.project = project
         self._essential_installed = False
+        self.auth = make_basic_auth(
+            self.project.sources, stream.verbosity >= stream.DETAIL
+        )
 
     @cached_property
     def python_executable(self) -> str:
@@ -242,7 +246,7 @@ class Environment:
         self,
         sources: Optional[List[Source]] = None,
         ignore_requires_python: bool = False,
-    ) -> pip_shims.PackageFinder:
+    ) -> Generator[pip_shims.PackageFinder, None, None]:
         """Return the package finder of given index sources.
 
         :param sources: a list of sources the finder should search in.
@@ -258,6 +262,8 @@ class Environment:
             python_version,
             ignore_requires_python,
         )
+        # Reuse the auth across sessions to avoid prompting repeatly.
+        finder.session.auth = self.auth
         yield finder
         finder.session.close()
 

pdm/models/pip_shims.py

@@ -18,6 +18,7 @@ from pip._internal.models.format_control import FormatControl
 from pip._internal.models.link import Link
 from pip._internal.models.target_python import TargetPython
 from pip._internal.models.wheel import Wheel as PipWheel
+from pip._internal.network.auth import MultiDomainBasicAuth
 from pip._internal.network.cache import SafeFileCache
 from pip._internal.network.download import Downloader
 from pip._internal.operations.prepare import unpack_url

pdm/utils.py

@@ -3,6 +3,7 @@ Utility functions
 """
 import atexit
 import os
+import re
 import shutil
 import subprocess
 import tempfile
@@ -349,3 +350,18 @@ def get_python_version_string(version: str, is_64bit: bool) -> str:
     if os.name == "nt" and not is_64bit:
         return f"{version}-32"
     return version
+
+
+def expand_env_vars(credential):
+    """A safe implementation of env var substitution.
+    It only supports the following forms:
+
+        ${ENV_VAR}
+
+    Neither $ENV_VAR and %ENV_VAR is not supported.
+    """
+
+    def replace_func(match):
+        return os.getenv(match.group(1), "")
+
+    return re.sub(r"\$\{(.+?)\}", replace_func, credential)