Access keyring only when needed.

Requesting a network page is first tried without retrieving authentication from the keyring.
Credentials provided by other means are used already.
If no authentication was found and the server answers with 401 or 403 the keyring is queried for credentials.
This fixes #1917.

* src/poetry/utils/constants.py (STATUS_AUTHLIST): A list of HTTP errors requesting authentication. Currently 401 and 403.

* src/poetry/utils/authenticator.py (Authenticator.request): Retrieve credentials with disabled keyring first. On auth error try again with enabled keyring.

* src/poetry/utils/authenticator.py (Authenticator.get_credentials_for_url): Add `keyring` option.
* src/poetry/utils/authenticator.py (Authenticator._get_credentials_for_url: ^^^ same as above
* src/poetry/utils/authenticator.py (Authenticator._get_credentials_for_repository): ^^^
* src/poetry/utils/authenticator.py (AuthenticatorRepositoryConfig.get_http_credentials): ^^^
* src/poetry/utils/password_manager.py (PasswordManager.get_http_auth): ^^^

* src/poetry/utils/password_manager.py (HTTPAuthCredential): Implement `empty` property that one of username or password exists.

* src/poetry/utils/authenticator.py (Authenticator): Use `HTTPAuthCredential.empty`. Only cache non-empty credentials.

* tests/utils/test_authenticator.py : Fix tests.

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]

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,8 +293,8 @@ def callback(
     ["status", "attempts"],
     [
         (400, 0),
-        (401, 0),
-        (403, 0),
+        (401, 1),
+        (403, 1),
         (404, 0),
         (429, 5),
         (500, 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")
     )