Access keyring only when needed.

src/poetry/utils/authenticator.py

@@ -18,11 +18,13 @@
 
 from cachecontrol import CacheControlAdapter
 from cachecontrol.caches import FileCache
+from requests import Request
 
 from poetry.config.config import Config
 from poetry.exceptions import PoetryException
 from poetry.utils.constants import REQUESTS_TIMEOUT
 from poetry.utils.constants import RETRY_AFTER_HEADER
+from poetry.utils.constants import STATUS_AUTHLIST
 from poetry.utils.constants import STATUS_FORCELIST
 from poetry.utils.password_manager import HTTPAuthCredential
 from poetry.utils.password_manager import PasswordManager
@@ -80,14 +82,17 @@ def http_credential_keys(self) -> list[str]:
         return [self.url, self.netloc, self.name]
 
     def get_http_credentials(
-        self, password_manager: PasswordManager, username: str | None = None
+        self,
+        password_manager: PasswordManager,
+        username: str | None = None,
+        keyring: bool = True,
     ) -> HTTPAuthCredential:
         # try with the repository name via the password manager
         credential = HTTPAuthCredential(
-            **(password_manager.get_http_auth(self.name) or {})
+            **(password_manager.get_http_auth(self.name, keyring=keyring) or {})
         )
 
-        if credential.password is None:
+        if credential.password is None and keyring:
             # fallback to url and netloc based keyring entries
             credential = password_manager.keyring.get_credential(
                 self.url, self.netloc, username=credential.username
@@ -190,19 +195,31 @@ def authenticated_url(self, url: str) -> str:
 
         return url
 
+    def _auth_request(
+        self, url: str, request: Request, keyring: bool
+    ) -> tuple[Request, bool]:
+        """Try to authenticate http request."""
+        credential = self.get_credentials_for_url(url, keyring=keyring)
+        if credential.empty:
+            return request, False
+
+        return (
+            requests.auth.HTTPBasicAuth(
+                credential.username or "", credential.password or ""
+            )(request),
+            True,
+        )
+
     def request(
         self, method: str, url: str, raise_for_status: bool = True, **kwargs: Any
     ) -> requests.Response:
         headers = kwargs.get("headers")
         request = requests.Request(method, url, headers=headers)
-        credential = self.get_credentials_for_url(url)
+        session = self.get_session(url=url)
 
-        if credential.username is not None or credential.password is not None:
-            request = requests.auth.HTTPBasicAuth(
-                credential.username or "", credential.password or ""
-            )(request)
+        # check config for credentials
+        request, authenticated = self._auth_request(url, request, False)
 
-        session = self.get_session(url=url)
         prepared_request = session.prepare_request(request)
 
         proxies: dict[str, str] = kwargs.get("proxies", {})
@@ -218,7 +235,7 @@ def request(
         verify = str(verify) if isinstance(verify, Path) else verify
 
         settings = session.merge_environment_settings(
-            prepared_request.url, proxies, stream, verify, cert
+            url, proxies, stream, verify, cert
         )
 
         # Send the request.
@@ -239,11 +256,19 @@ def request(
                 if is_last_attempt:
                     raise e
             else:
-                if resp.status_code not in STATUS_FORCELIST or is_last_attempt:
+                if (resp.status_code not in STATUS_FORCELIST or is_last_attempt) and (
+                    authenticated or resp.status_code not in STATUS_AUTHLIST
+                ):
                     if raise_for_status:
                         resp.raise_for_status()
                     return resp
 
+                if resp.status_code in STATUS_AUTHLIST:
+                    # ask keyring for authentication
+                    request, _ = self._auth_request(url, request, True)
+                    authenticated = True
+                    prepared_request = session.prepare_request(request)
+
             if not is_last_attempt:
                 attempt += 1
                 delay = self._get_backoff(resp, attempt)
@@ -269,31 +294,39 @@ def post(self, url: str, **kwargs: Any) -> requests.Response:
         return self.request("post", url, **kwargs)
 
     def _get_credentials_for_repository(
-        self, repository: AuthenticatorRepositoryConfig, username: str | None = None
+        self,
+        repository: AuthenticatorRepositoryConfig,
+        username: str | None = None,
+        keyring: bool = True,
     ) -> HTTPAuthCredential:
         # cache repository credentials by repository url to avoid multiple keyring
         # backend queries when packages are being downloaded from the same source
         key = f"{repository.url}#username={username or ''}"
 
         if key not in self._credentials:
-            self._credentials[key] = repository.get_http_credentials(
-                password_manager=self._password_manager, username=username
+            credential = repository.get_http_credentials(
+                password_manager=self._password_manager,
+                username=username,
+                keyring=keyring,
             )
+            if credential.empty:
+                return credential
+            self._credentials[key] = credential
 
         return self._credentials[key]
 
     def _get_credentials_for_url(
-        self, url: str, exact_match: bool = False
+        self, url: str, exact_match: bool = False, keyring: bool = True
     ) -> HTTPAuthCredential:
         repository = self.get_repository_config_for_url(url, exact_match)
 
         credential = (
-            self._get_credentials_for_repository(repository=repository)
+            self._get_credentials_for_repository(repository=repository, keyring=keyring)
             if repository is not None
             else HTTPAuthCredential()
         )
 
-        if credential.password is None:
+        if credential.password is None and keyring:
             parsed_url = urllib.parse.urlsplit(url)
             netloc = parsed_url.netloc
             credential = self._password_manager.keyring.get_credential(
@@ -319,15 +352,20 @@ def get_credentials_for_git_url(self, url: str) -> HTTPAuthCredential:
 
         return self._credentials[key]
 
-    def get_credentials_for_url(self, url: str) -> HTTPAuthCredential:
+    def get_credentials_for_url(
+        self, url: str, keyring: bool = True
+    ) -> HTTPAuthCredential:
         parsed_url = urllib.parse.urlsplit(url)
         netloc = parsed_url.netloc
 
         if url not in self._credentials:
             if "@" not in netloc:
                 # no credentials were provided in the url, try finding the
                 # best repository configuration
-                self._credentials[url] = self._get_credentials_for_url(url)
+                credential = self._get_credentials_for_url(url, keyring=keyring)
+                if credential.empty:
+                    return credential
+                self._credentials[url] = credential
             else:
                 # Split from the right because that's how urllib.parse.urlsplit()
                 # behaves if more than one @ is present (which can be checked using

src/poetry/utils/constants.py

@@ -10,3 +10,6 @@
 
 # Server response codes to retry requests on.
 STATUS_FORCELIST = [429, 500, 501, 502, 503, 504]
+
+# Server response code to try to retrieve authentication on.
+STATUS_AUTHLIST = [401, 403, 404]

src/poetry/utils/password_manager.py

@@ -26,6 +26,10 @@ class HTTPAuthCredential:
     username: str | None = dataclasses.field(default=None)
     password: str | None = dataclasses.field(default=None)
 
+    @property
+    def empty(self) -> bool:
+        return self.username is None and self.password is None
+
 
 class PoetryKeyring:
     def __init__(self, namespace: str) -> None:
@@ -192,13 +196,15 @@ def delete_pypi_token(self, name: str) -> None:
 
         self.keyring.delete_password(name, "__token__")
 
-    def get_http_auth(self, name: str) -> dict[str, str | None] | None:
+    def get_http_auth(
+        self, name: str, keyring: bool = True
+    ) -> dict[str, str | None] | None:
         username = self._config.get(f"http-basic.{name}.username")
         password = self._config.get(f"http-basic.{name}.password")
         if not username and not password:
             return None
 
-        if not password:
+        if not password and keyring:
             password = self.keyring.get_password(name, username)
 
         return {

tests/utils/test_authenticator.py

@@ -8,6 +8,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Callable
+from typing import Iterable
 
 import httpretty
 import pytest
@@ -21,6 +23,7 @@
 
 if TYPE_CHECKING:
     from _pytest.monkeypatch import MonkeyPatch
+    from httpretty.core import HTTPrettyRequest
     from pytest_mock import MockerFixture
 
     from tests.conftest import Config
@@ -148,9 +151,23 @@ def test_authenticator_uses_empty_strings_as_default_username(
     assert request.headers["Authorization"] == "Basic OmJhcg=="
 
 
+def status_callback(
+    codes: Iterable[int],
+) -> Callable[[HTTPrettyRequest, str, dict[str, str]], tuple[int, dict[str, str], str]]:
+    """Return a httpretty callback that returns a sequence of status codes."""
+    it = iter(codes)
+
+    def callback(
+        request: HTTPrettyRequest, url: str, headers: dict[str, str]
+    ) -> tuple[int, dict[str, str], str]:
+        v = next(it)
+        return v, headers, ""
+
+    return callback
+
+
 def test_authenticator_falls_back_to_keyring_url(
     config: Config,
-    mock_remote: None,
     repo: dict[str, dict[str, str]],
     http: type[httpretty.httpretty],
     with_simple_keyring: None,
@@ -162,21 +179,23 @@ def test_authenticator_falls_back_to_keyring_url(
         }
     )
 
+    http.register_uri(http.GET, re.compile(r".*"), body=status_callback((401, 200)))
+
     dummy_keyring.set_password(
         "https://foo.bar/simple/", None, SimpleCredential("foo", "bar")
     )
 
     authenticator = Authenticator(config, NullIO())
-    authenticator.request("get", "https://foo.bar/files/foo-0.1.0.tar.gz")
+    authenticator.request("GET", "https://foo.bar/files/foo-0.1.0.tar.gz")
 
     request = http.last_request()
+    assert request
 
     assert request.headers["Authorization"] == "Basic Zm9vOmJhcg=="
 
 
 def test_authenticator_falls_back_to_keyring_netloc(
     config: Config,
-    mock_remote: None,
     repo: dict[str, dict[str, str]],
     http: type[httpretty.httpretty],
     with_simple_keyring: None,
@@ -187,6 +206,7 @@ def test_authenticator_falls_back_to_keyring_netloc(
             "repositories": repo,
         }
     )
+    http.register_uri(http.GET, re.compile(r".*"), body=status_callback((401, 200)))
 
     dummy_keyring.set_password("foo.bar", None, SimpleCredential("foo", "bar"))
 
@@ -273,9 +293,9 @@ def callback(
     ["status", "attempts"],
     [
         (400, 0),
-        (401, 0),
-        (403, 0),
-        (404, 0),
+        (401, 1),
+        (403, 1),
+        (404, 1),
         (429, 5),
         (500, 5),
         (501, 5),
@@ -435,7 +455,6 @@ def test_authenticator_uses_credentials_from_config_with_at_sign_in_path(
 
 def test_authenticator_falls_back_to_keyring_url_matched_by_path(
     config: Config,
-    mock_remote: None,
     http: type[httpretty.httpretty],
     with_simple_keyring: None,
     dummy_keyring: DummyBackend,
@@ -449,6 +468,8 @@ def test_authenticator_falls_back_to_keyring_url_matched_by_path(
         }
     )
 
+    http.register_uri(http.GET, re.compile(r".*"), body=status_callback((401, 200) * 2))
+
     dummy_keyring.set_password(
         "https://foo.bar/alpha/files/simple/", None, SimpleCredential("foo", "bar")
     )