We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Upgrading packages from private repositories fails using cert and client-cert 
<!--
  Hi there! Thank you for discovering and submitting an issue.

  Before you submit this; let's make sure of a few things.
  Please make sure the following boxes are ticked if they are correct.
  If not, please try and fulfill these first.
-->

<!-- Checked checkbox should look like this: [x] -->
- [x] I am on the [latest](https://github.com/python-poetry/poetry/releases/latest) Poetry version.
- [x] I have searched the [issues](https://github.com/python-poetry/poetry/issues) of this repo and believe that this is not a duplicate.
- [x] If an exception occurs when executing a command, I executed it again in debug mode (`-vvv` option).

<!--
  Once those are done, if you're able to fill in the following list with your information,
  it'd be very helpful to whoever handles the issue.
-->

- **Poetry version**: **1.1.4** <!-- Replace with version -->

## Issue
<!-- Now feel free to write your issue, but please be descriptive! Thanks again 🙌 ❤️ -->

To configure the cert and client-cert I am using:

    poetry config certificates.private-pypi.cert /path/to/ca.crt
    poetry config certificates.private-pypi.client-cert /path/to/client.pem

... from the custom CA & TLS [instructions](https://python-poetry.org/docs/repositories/#custom-certificate-authority-and-mutual-tls-authentication).

And from pyproject.toml:

    [[tool.poetry.source]]
    name = "private-pypi"
    url = "https://private-pypi/simple"

When doing a `poetry update` to get the latest packages, it fails to get packages from private repositories. 

It fails with:

    Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify
     failed: self signed certificate in certificate chain (_ssl.c:1091)'))

... I started to debug that error it looked like the cert and CA were simply not provided in the http request. Actually here's what I saw:

A single successful request where the cert and CA were passed:

    GET https://private-pypi/simple/<package_name>/

And then 5 failed retries that had the SSLCertVerificationError:

    GET https://private-pypi/packages/<package_name>/<package>-py3-none-any.whl#md5=blah

And in those 5 requests it looked like the cert and CA were not provided. So in my CI build I am not able to install `<package>` which happens to be an upgrade. I believe this started to happen with 1.1.0 as I can force version 1.0.10 and then the CI build will succeed and the package will install successfully.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/poetry/utils/authenticator.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 import time
5 import urllib.parse
6 
7 from typing import TYPE_CHECKING
8 from typing import Any
9 
10 import requests
11 import requests.auth
12 import requests.exceptions
13 
14 from poetry.exceptions import PoetryException
15 from poetry.utils.password_manager import PasswordManager
16 
17 
18 if TYPE_CHECKING:
19     from cleo.io.io import IO
20 
21     from poetry.config.config import Config
22 
23 
24 logger = logging.getLogger()
25 
26 
27 class Authenticator:
28     def __init__(self, config: Config, io: IO | None = None) -> None:
29         self._config = config
30         self._io = io
31         self._session = None
32         self._credentials = {}
33         self._password_manager = PasswordManager(self._config)
34 
35     def _log(self, message: str, level: str = "debug") -> None:
36         if self._io is not None:
37             self._io.write_line(f"<{level}>{message}</{level}>")
38         else:
39             getattr(logger, level, logger.debug)(message)
40 
41     @property
42     def session(self) -> requests.Session:
43         if self._session is None:
44             self._session = requests.Session()
45 
46         return self._session
47 
48     def __del__(self) -> None:
49         if self._session is not None:
50             self._session.close()
51 
52     def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
53         request = requests.Request(method, url)
54         username, password = self.get_credentials_for_url(url)
55 
56         if username is not None and password is not None:
57             request = requests.auth.HTTPBasicAuth(username, password)(request)
58 
59         session = self.session
60         prepared_request = session.prepare_request(request)
61 
62         proxies = kwargs.get("proxies", {})
63         stream = kwargs.get("stream")
64         verify = kwargs.get("verify")
65         cert = kwargs.get("cert")
66 
67         settings = session.merge_environment_settings(
68             prepared_request.url, proxies, stream, verify, cert
69         )
70 
71         # Send the request.
72         send_kwargs = {
73             "timeout": kwargs.get("timeout"),
74             "allow_redirects": kwargs.get("allow_redirects", True),
75         }
76         send_kwargs.update(settings)
77 
78         attempt = 0
79 
80         while True:
81             is_last_attempt = attempt >= 5
82             try:
83                 resp = session.send(prepared_request, **send_kwargs)
84             except (requests.exceptions.ConnectionError, OSError) as e:
85                 if is_last_attempt:
86                     raise e
87             else:
88                 if resp.status_code not in [502, 503, 504] or is_last_attempt:
89                     resp.raise_for_status()
90                     return resp
91 
92             if not is_last_attempt:
93                 attempt += 1
94                 delay = 0.5 * attempt
95                 self._log(f"Retrying HTTP request in {delay} seconds.", level="debug")
96                 time.sleep(delay)
97                 continue
98 
99         # this should never really be hit under any sane circumstance
100         raise PoetryException("Failed HTTP {} request", method.upper())
101 
102     def get_credentials_for_url(self, url: str) -> tuple[str | None, str | None]:
103         parsed_url = urllib.parse.urlsplit(url)
104 
105         netloc = parsed_url.netloc
106 
107         credentials = self._credentials.get(netloc, (None, None))
108 
109         if credentials == (None, None):
110             if "@" not in netloc:
111                 credentials = self._get_credentials_for_netloc(netloc)
112             else:
113                 # Split from the right because that's how urllib.parse.urlsplit()
114                 # behaves if more than one @ is present (which can be checked using
115                 # the password attribute of urlsplit()'s return value).
116                 auth, netloc = netloc.rsplit("@", 1)
117                 # Split from the left because that's how urllib.parse.urlsplit()
118                 # behaves if more than one : is present (which again can be checked
119                 # using the password attribute of the return value)
120                 credentials = auth.split(":", 1) if ":" in auth else (auth, None)
121                 credentials = tuple(
122                     None if x is None else urllib.parse.unquote(x) for x in credentials
123                 )
124 
125         if credentials[0] is not None or credentials[1] is not None:
126             credentials = (credentials[0] or "", credentials[1] or "")
127 
128             self._credentials[netloc] = credentials
129 
130         return credentials[0], credentials[1]
131 
132     def get_pypi_token(self, name: str) -> str:
133         return self._password_manager.get_pypi_token(name)
134 
135     def get_http_auth(self, name: str) -> dict[str, str] | None:
136         return self._get_http_auth(name, None)
137 
138     def _get_http_auth(self, name: str, netloc: str | None) -> dict[str, str] | None:
139         if name == "pypi":
140             url = "https://upload.pypi.org/legacy/"
141         else:
142             url = self._config.get(f"repositories.{name}.url")
143             if not url:
144                 return None
145 
146         parsed_url = urllib.parse.urlsplit(url)
147 
148         if netloc is None or netloc == parsed_url.netloc:
149             auth = self._password_manager.get_http_auth(name)
150 
151             if auth is None or auth["password"] is None:
152                 username = auth["username"] if auth else None
153                 auth = self._get_credentials_for_netloc_from_keyring(
154                     url, parsed_url.netloc, username
155                 )
156 
157             return auth
158 
159     def _get_credentials_for_netloc(self, netloc: str) -> tuple[str | None, str | None]:
160         for repository_name in self._config.get("repositories", []):
161             auth = self._get_http_auth(repository_name, netloc)
162 
163             if auth is None:
164                 continue
165 
166             return auth["username"], auth["password"]
167 
168         return None, None
169 
170     def _get_credentials_for_netloc_from_keyring(
171         self, url: str, netloc: str, username: str | None
172     ) -> dict[str, str] | None:
173         import keyring
174 
175         cred = keyring.get_credential(url, username)
176         if cred is not None:
177             return {
178                 "username": cred.username,
179                 "password": cred.password,
180             }
181 
182         cred = keyring.get_credential(netloc, username)
183         if cred is not None:
184             return {
185                 "username": cred.username,
186                 "password": cred.password,
187             }
188 
189         if username:
190             return {
191                 "username": username,
192                 "password": None,
193             }
194 
195         return None
196
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/poetry/utils/authenticator.py b/src/poetry/utils/authenticator.py
--- a/src/poetry/utils/authenticator.py
+++ b/src/poetry/utils/authenticator.py
@@ -6,16 +6,21 @@
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Iterator
 
 import requests
 import requests.auth
 import requests.exceptions
 
 from poetry.exceptions import PoetryException
+from poetry.utils.helpers import get_cert
+from poetry.utils.helpers import get_client_cert
 from poetry.utils.password_manager import PasswordManager
 
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from cleo.io.io import IO
 
     from poetry.config.config import Config
@@ -30,6 +35,7 @@
         self._io = io
         self._session = None
         self._credentials = {}
+        self._certs = {}
         self._password_manager = PasswordManager(self._config)
 
     def _log(self, message: str, level: str = "debug") -> None:
@@ -61,8 +67,16 @@
 
         proxies = kwargs.get("proxies", {})
         stream = kwargs.get("stream")
-        verify = kwargs.get("verify")
-        cert = kwargs.get("cert")
+
+        certs = self.get_certs_for_url(url)
+        verify = kwargs.get("verify") or certs.get("verify")
+        cert = kwargs.get("cert") or certs.get("cert")
+
+        if cert is not None:
+            cert = str(cert)
+
+        if verify is not None:
+            verify = str(verify)
 
         settings = session.merge_environment_settings(
             prepared_request.url, proxies, stream, verify, cert
@@ -157,7 +171,7 @@
             return auth
 
     def _get_credentials_for_netloc(self, netloc: str) -> tuple[str | None, str | None]:
-        for repository_name in self._config.get("repositories", []):
+        for repository_name, _ in self._get_repository_netlocs():
             auth = self._get_http_auth(repository_name, netloc)
 
             if auth is None:
@@ -167,6 +181,22 @@
 
         return None, None
 
+    def get_certs_for_url(self, url: str) -> dict[str, Path | None]:
+        parsed_url = urllib.parse.urlsplit(url)
+
+        netloc = parsed_url.netloc
+
+        return self._certs.setdefault(
+            netloc,
+            self._get_certs_for_netloc_from_config(netloc),
+        )
+
+    def _get_repository_netlocs(self) -> Iterator[tuple[str, str]]:
+        for repository_name in self._config.get("repositories", []):
+            url = self._config.get(f"repositories.{repository_name}.url")
+            parsed_url = urllib.parse.urlsplit(url)
+            yield repository_name, parsed_url.netloc
+
     def _get_credentials_for_netloc_from_keyring(
         self, url: str, netloc: str, username: str | None
     ) -> dict[str, str] | None:
@@ -193,3 +223,14 @@
             }
 
         return None
+
+    def _get_certs_for_netloc_from_config(self, netloc: str) -> dict[str, Path | None]:
+        certs = {"cert": None, "verify": None}
+
+        for repository_name, repository_netloc in self._get_repository_netlocs():
+            if netloc == repository_netloc:
+                certs["cert"] = get_client_cert(self._config, repository_name)
+                certs["verify"] = get_cert(self._config, repository_name)
+                break
+
+        return certs

{"golden_diff": "diff --git a/src/poetry/utils/authenticator.py b/src/poetry/utils/authenticator.py\n--- a/src/poetry/utils/authenticator.py\n+++ b/src/poetry/utils/authenticator.py\n@@ -6,16 +6,21 @@\n \n from typing import TYPE_CHECKING\n from typing import Any\n+from typing import Iterator\n \n import requests\n import requests.auth\n import requests.exceptions\n \n from poetry.exceptions import PoetryException\n+from poetry.utils.helpers import get_cert\n+from poetry.utils.helpers import get_client_cert\n from poetry.utils.password_manager import PasswordManager\n \n \n if TYPE_CHECKING:\n+    from pathlib import Path\n+\n     from cleo.io.io import IO\n \n     from poetry.config.config import Config\n@@ -30,6 +35,7 @@\n         self._io = io\n         self._session = None\n         self._credentials = {}\n+        self._certs = {}\n         self._password_manager = PasswordManager(self._config)\n \n     def _log(self, message: str, level: str = \"debug\") -> None:\n@@ -61,8 +67,16 @@\n \n         proxies = kwargs.get(\"proxies\", {})\n         stream = kwargs.get(\"stream\")\n-        verify = kwargs.get(\"verify\")\n-        cert = kwargs.get(\"cert\")\n+\n+        certs = self.get_certs_for_url(url)\n+        verify = kwargs.get(\"verify\") or certs.get(\"verify\")\n+        cert = kwargs.get(\"cert\") or certs.get(\"cert\")\n+\n+        if cert is not None:\n+            cert = str(cert)\n+\n+        if verify is not None:\n+            verify = str(verify)\n \n         settings = session.merge_environment_settings(\n             prepared_request.url, proxies, stream, verify, cert\n@@ -157,7 +171,7 @@\n             return auth\n \n     def _get_credentials_for_netloc(self, netloc: str) -> tuple[str | None, str | None]:\n-        for repository_name in self._config.get(\"repositories\", []):\n+        for repository_name, _ in self._get_repository_netlocs():\n             auth = self._get_http_auth(repository_name, netloc)\n \n             if auth is None:\n@@ -167,6 +181,22 @@\n \n         return None, None\n \n+    def get_certs_for_url(self, url: str) -> dict[str, Path | None]:\n+        parsed_url = urllib.parse.urlsplit(url)\n+\n+        netloc = parsed_url.netloc\n+\n+        return self._certs.setdefault(\n+            netloc,\n+            self._get_certs_for_netloc_from_config(netloc),\n+        )\n+\n+    def _get_repository_netlocs(self) -> Iterator[tuple[str, str]]:\n+        for repository_name in self._config.get(\"repositories\", []):\n+            url = self._config.get(f\"repositories.{repository_name}.url\")\n+            parsed_url = urllib.parse.urlsplit(url)\n+            yield repository_name, parsed_url.netloc\n+\n     def _get_credentials_for_netloc_from_keyring(\n         self, url: str, netloc: str, username: str | None\n     ) -> dict[str, str] | None:\n@@ -193,3 +223,14 @@\n             }\n \n         return None\n+\n+    def _get_certs_for_netloc_from_config(self, netloc: str) -> dict[str, Path | None]:\n+        certs = {\"cert\": None, \"verify\": None}\n+\n+        for repository_name, repository_netloc in self._get_repository_netlocs():\n+            if netloc == repository_netloc:\n+                certs[\"cert\"] = get_client_cert(self._config, repository_name)\n+                certs[\"verify\"] = get_cert(self._config, repository_name)\n+                break\n+\n+        return certs\n", "issue": "Upgrading packages from private repositories fails using cert and client-cert \n<!--\r\n  Hi there! Thank you for discovering and submitting an issue.\r\n\r\n  Before you submit this; let's make sure of a few things.\r\n  Please make sure the following boxes are ticked if they are correct.\r\n  If not, please try and fulfill these first.\r\n-->\r\n\r\n<!-- Checked checkbox should look like this: [x] -->\r\n- [x] I am on the [latest](https://github.com/python-poetry/poetry/releases/latest) Poetry version.\r\n- [x] I have searched the [issues](https://github.com/python-poetry/poetry/issues) of this repo and believe that this is not a duplicate.\r\n- [x] If an exception occurs when executing a command, I executed it again in debug mode (`-vvv` option).\r\n\r\n<!--\r\n  Once those are done, if you're able to fill in the following list with your information,\r\n  it'd be very helpful to whoever handles the issue.\r\n-->\r\n\r\n- **Poetry version**: **1.1.4** <!-- Replace with version -->\r\n\r\n## Issue\r\n<!-- Now feel free to write your issue, but please be descriptive! Thanks again \ud83d\ude4c \u2764\ufe0f -->\r\n\r\nTo configure the cert and client-cert I am using:\r\n\r\n    poetry config certificates.private-pypi.cert /path/to/ca.crt\r\n    poetry config certificates.private-pypi.client-cert /path/to/client.pem\r\n\r\n... from the custom CA & TLS [instructions](https://python-poetry.org/docs/repositories/#custom-certificate-authority-and-mutual-tls-authentication).\r\n\r\nAnd from pyproject.toml:\r\n\r\n    [[tool.poetry.source]]\r\n    name = \"private-pypi\"\r\n    url = \"https://private-pypi/simple\"\r\n\r\nWhen doing a `poetry update` to get the latest packages, it fails to get packages from private repositories. \r\n\r\nIt fails with:\r\n\r\n    Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify\r\n     failed: self signed certificate in certificate chain (_ssl.c:1091)'))\r\n\r\n... I started to debug that error it looked like the cert and CA were simply not provided in the http request. Actually here's what I saw:\r\n\r\nA single successful request where the cert and CA were passed:\r\n\r\n    GET https://private-pypi/simple/<package_name>/\r\n\r\nAnd then 5 failed retries that had the SSLCertVerificationError:\r\n\r\n    GET https://private-pypi/packages/<package_name>/<package>-py3-none-any.whl#md5=blah\r\n\r\nAnd in those 5 requests it looked like the cert and CA were not provided. So in my CI build I am not able to install `<package>` which happens to be an upgrade. I believe this started to happen with 1.1.0 as I can force version 1.0.10 and then the CI build will succeed and the package will install successfully.\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport time\nimport urllib.parse\n\nfrom typing import TYPE_CHECKING\nfrom typing import Any\n\nimport requests\nimport requests.auth\nimport requests.exceptions\n\nfrom poetry.exceptions import PoetryException\nfrom poetry.utils.password_manager import PasswordManager\n\n\nif TYPE_CHECKING:\n    from cleo.io.io import IO\n\n    from poetry.config.config import Config\n\n\nlogger = logging.getLogger()\n\n\nclass Authenticator:\n    def __init__(self, config: Config, io: IO | None = None) -> None:\n        self._config = config\n        self._io = io\n        self._session = None\n        self._credentials = {}\n        self._password_manager = PasswordManager(self._config)\n\n    def _log(self, message: str, level: str = \"debug\") -> None:\n        if self._io is not None:\n            self._io.write_line(f\"<{level}>{message}</{level}>\")\n        else:\n            getattr(logger, level, logger.debug)(message)\n\n    @property\n    def session(self) -> requests.Session:\n        if self._session is None:\n            self._session = requests.Session()\n\n        return self._session\n\n    def __del__(self) -> None:\n        if self._session is not None:\n            self._session.close()\n\n    def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:\n        request = requests.Request(method, url)\n        username, password = self.get_credentials_for_url(url)\n\n        if username is not None and password is not None:\n            request = requests.auth.HTTPBasicAuth(username, password)(request)\n\n        session = self.session\n        prepared_request = session.prepare_request(request)\n\n        proxies = kwargs.get(\"proxies\", {})\n        stream = kwargs.get(\"stream\")\n        verify = kwargs.get(\"verify\")\n        cert = kwargs.get(\"cert\")\n\n        settings = session.merge_environment_settings(\n            prepared_request.url, proxies, stream, verify, cert\n        )\n\n        # Send the request.\n        send_kwargs = {\n            \"timeout\": kwargs.get(\"timeout\"),\n            \"allow_redirects\": kwargs.get(\"allow_redirects\", True),\n        }\n        send_kwargs.update(settings)\n\n        attempt = 0\n\n        while True:\n            is_last_attempt = attempt >= 5\n            try:\n                resp = session.send(prepared_request, **send_kwargs)\n            except (requests.exceptions.ConnectionError, OSError) as e:\n                if is_last_attempt:\n                    raise e\n            else:\n                if resp.status_code not in [502, 503, 504] or is_last_attempt:\n                    resp.raise_for_status()\n                    return resp\n\n            if not is_last_attempt:\n                attempt += 1\n                delay = 0.5 * attempt\n                self._log(f\"Retrying HTTP request in {delay} seconds.\", level=\"debug\")\n                time.sleep(delay)\n                continue\n\n        # this should never really be hit under any sane circumstance\n        raise PoetryException(\"Failed HTTP {} request\", method.upper())\n\n    def get_credentials_for_url(self, url: str) -> tuple[str | None, str | None]:\n        parsed_url = urllib.parse.urlsplit(url)\n\n        netloc = parsed_url.netloc\n\n        credentials = self._credentials.get(netloc, (None, None))\n\n        if credentials == (None, None):\n            if \"@\" not in netloc:\n                credentials = self._get_credentials_for_netloc(netloc)\n            else:\n                # Split from the right because that's how urllib.parse.urlsplit()\n                # behaves if more than one @ is present (which can be checked using\n                # the password attribute of urlsplit()'s return value).\n                auth, netloc = netloc.rsplit(\"@\", 1)\n                # Split from the left because that's how urllib.parse.urlsplit()\n                # behaves if more than one : is present (which again can be checked\n                # using the password attribute of the return value)\n                credentials = auth.split(\":\", 1) if \":\" in auth else (auth, None)\n                credentials = tuple(\n                    None if x is None else urllib.parse.unquote(x) for x in credentials\n                )\n\n        if credentials[0] is not None or credentials[1] is not None:\n            credentials = (credentials[0] or \"\", credentials[1] or \"\")\n\n            self._credentials[netloc] = credentials\n\n        return credentials[0], credentials[1]\n\n    def get_pypi_token(self, name: str) -> str:\n        return self._password_manager.get_pypi_token(name)\n\n    def get_http_auth(self, name: str) -> dict[str, str] | None:\n        return self._get_http_auth(name, None)\n\n    def _get_http_auth(self, name: str, netloc: str | None) -> dict[str, str] | None:\n        if name == \"pypi\":\n            url = \"https://upload.pypi.org/legacy/\"\n        else:\n            url = self._config.get(f\"repositories.{name}.url\")\n            if not url:\n                return None\n\n        parsed_url = urllib.parse.urlsplit(url)\n\n        if netloc is None or netloc == parsed_url.netloc:\n            auth = self._password_manager.get_http_auth(name)\n\n            if auth is None or auth[\"password\"] is None:\n                username = auth[\"username\"] if auth else None\n                auth = self._get_credentials_for_netloc_from_keyring(\n                    url, parsed_url.netloc, username\n                )\n\n            return auth\n\n    def _get_credentials_for_netloc(self, netloc: str) -> tuple[str | None, str | None]:\n        for repository_name in self._config.get(\"repositories\", []):\n            auth = self._get_http_auth(repository_name, netloc)\n\n            if auth is None:\n                continue\n\n            return auth[\"username\"], auth[\"password\"]\n\n        return None, None\n\n    def _get_credentials_for_netloc_from_keyring(\n        self, url: str, netloc: str, username: str | None\n    ) -> dict[str, str] | None:\n        import keyring\n\n        cred = keyring.get_credential(url, username)\n        if cred is not None:\n            return {\n                \"username\": cred.username,\n                \"password\": cred.password,\n            }\n\n        cred = keyring.get_credential(netloc, username)\n        if cred is not None:\n            return {\n                \"username\": cred.username,\n                \"password\": cred.password,\n            }\n\n        if username:\n            return {\n                \"username\": username,\n                \"password\": None,\n            }\n\n        return None\n", "path": "src/poetry/utils/authenticator.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport time\nimport urllib.parse\n\nfrom typing import TYPE_CHECKING\nfrom typing import Any\nfrom typing import Iterator\n\nimport requests\nimport requests.auth\nimport requests.exceptions\n\nfrom poetry.exceptions import PoetryException\nfrom poetry.utils.helpers import get_cert\nfrom poetry.utils.helpers import get_client_cert\nfrom poetry.utils.password_manager import PasswordManager\n\n\nif TYPE_CHECKING:\n    from pathlib import Path\n\n    from cleo.io.io import IO\n\n    from poetry.config.config import Config\n\n\nlogger = logging.getLogger()\n\n\nclass Authenticator:\n    def __init__(self, config: Config, io: IO | None = None) -> None:\n        self._config = config\n        self._io = io\n        self._session = None\n        self._credentials = {}\n        self._certs = {}\n        self._password_manager = PasswordManager(self._config)\n\n    def _log(self, message: str, level: str = \"debug\") -> None:\n        if self._io is not None:\n            self._io.write_line(f\"<{level}>{message}</{level}>\")\n        else:\n            getattr(logger, level, logger.debug)(message)\n\n    @property\n    def session(self) -> requests.Session:\n        if self._session is None:\n            self._session = requests.Session()\n\n        return self._session\n\n    def __del__(self) -> None:\n        if self._session is not None:\n            self._session.close()\n\n    def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:\n        request = requests.Request(method, url)\n        username, password = self.get_credentials_for_url(url)\n\n        if username is not None and password is not None:\n            request = requests.auth.HTTPBasicAuth(username, password)(request)\n\n        session = self.session\n        prepared_request = session.prepare_request(request)\n\n        proxies = kwargs.get(\"proxies\", {})\n        stream = kwargs.get(\"stream\")\n\n        certs = self.get_certs_for_url(url)\n        verify = kwargs.get(\"verify\") or certs.get(\"verify\")\n        cert = kwargs.get(\"cert\") or certs.get(\"cert\")\n\n        if cert is not None:\n            cert = str(cert)\n\n        if verify is not None:\n            verify = str(verify)\n\n        settings = session.merge_environment_settings(\n            prepared_request.url, proxies, stream, verify, cert\n        )\n\n        # Send the request.\n        send_kwargs = {\n            \"timeout\": kwargs.get(\"timeout\"),\n            \"allow_redirects\": kwargs.get(\"allow_redirects\", True),\n        }\n        send_kwargs.update(settings)\n\n        attempt = 0\n\n        while True:\n            is_last_attempt = attempt >= 5\n            try:\n                resp = session.send(prepared_request, **send_kwargs)\n            except (requests.exceptions.ConnectionError, OSError) as e:\n                if is_last_attempt:\n                    raise e\n            else:\n                if resp.status_code not in [502, 503, 504] or is_last_attempt:\n                    resp.raise_for_status()\n                    return resp\n\n            if not is_last_attempt:\n                attempt += 1\n                delay = 0.5 * attempt\n                self._log(f\"Retrying HTTP request in {delay} seconds.\", level=\"debug\")\n                time.sleep(delay)\n                continue\n\n        # this should never really be hit under any sane circumstance\n        raise PoetryException(\"Failed HTTP {} request\", method.upper())\n\n    def get_credentials_for_url(self, url: str) -> tuple[str | None, str | None]:\n        parsed_url = urllib.parse.urlsplit(url)\n\n        netloc = parsed_url.netloc\n\n        credentials = self._credentials.get(netloc, (None, None))\n\n        if credentials == (None, None):\n            if \"@\" not in netloc:\n                credentials = self._get_credentials_for_netloc(netloc)\n            else:\n                # Split from the right because that's how urllib.parse.urlsplit()\n                # behaves if more than one @ is present (which can be checked using\n                # the password attribute of urlsplit()'s return value).\n                auth, netloc = netloc.rsplit(\"@\", 1)\n                # Split from the left because that's how urllib.parse.urlsplit()\n                # behaves if more than one : is present (which again can be checked\n                # using the password attribute of the return value)\n                credentials = auth.split(\":\", 1) if \":\" in auth else (auth, None)\n                credentials = tuple(\n                    None if x is None else urllib.parse.unquote(x) for x in credentials\n                )\n\n        if credentials[0] is not None or credentials[1] is not None:\n            credentials = (credentials[0] or \"\", credentials[1] or \"\")\n\n            self._credentials[netloc] = credentials\n\n        return credentials[0], credentials[1]\n\n    def get_pypi_token(self, name: str) -> str:\n        return self._password_manager.get_pypi_token(name)\n\n    def get_http_auth(self, name: str) -> dict[str, str] | None:\n        return self._get_http_auth(name, None)\n\n    def _get_http_auth(self, name: str, netloc: str | None) -> dict[str, str] | None:\n        if name == \"pypi\":\n            url = \"https://upload.pypi.org/legacy/\"\n        else:\n            url = self._config.get(f\"repositories.{name}.url\")\n            if not url:\n                return None\n\n        parsed_url = urllib.parse.urlsplit(url)\n\n        if netloc is None or netloc == parsed_url.netloc:\n            auth = self._password_manager.get_http_auth(name)\n\n            if auth is None or auth[\"password\"] is None:\n                username = auth[\"username\"] if auth else None\n                auth = self._get_credentials_for_netloc_from_keyring(\n                    url, parsed_url.netloc, username\n                )\n\n            return auth\n\n    def _get_credentials_for_netloc(self, netloc: str) -> tuple[str | None, str | None]:\n        for repository_name, _ in self._get_repository_netlocs():\n            auth = self._get_http_auth(repository_name, netloc)\n\n            if auth is None:\n                continue\n\n            return auth[\"username\"], auth[\"password\"]\n\n        return None, None\n\n    def get_certs_for_url(self, url: str) -> dict[str, Path | None]:\n        parsed_url = urllib.parse.urlsplit(url)\n\n        netloc = parsed_url.netloc\n\n        return self._certs.setdefault(\n            netloc,\n            self._get_certs_for_netloc_from_config(netloc),\n        )\n\n    def _get_repository_netlocs(self) -> Iterator[tuple[str, str]]:\n        for repository_name in self._config.get(\"repositories\", []):\n            url = self._config.get(f\"repositories.{repository_name}.url\")\n            parsed_url = urllib.parse.urlsplit(url)\n            yield repository_name, parsed_url.netloc\n\n    def _get_credentials_for_netloc_from_keyring(\n        self, url: str, netloc: str, username: str | None\n    ) -> dict[str, str] | None:\n        import keyring\n\n        cred = keyring.get_credential(url, username)\n        if cred is not None:\n            return {\n                \"username\": cred.username,\n                \"password\": cred.password,\n            }\n\n        cred = keyring.get_credential(netloc, username)\n        if cred is not None:\n            return {\n                \"username\": cred.username,\n                \"password\": cred.password,\n            }\n\n        if username:\n            return {\n                \"username\": username,\n                \"password\": None,\n            }\n\n        return None\n\n    def _get_certs_for_netloc_from_config(self, netloc: str) -> dict[str, Path | None]:\n        certs = {\"cert\": None, \"verify\": None}\n\n        for repository_name, repository_netloc in self._get_repository_netlocs():\n            if netloc == repository_netloc:\n                certs[\"cert\"] = get_client_cert(self._config, repository_name)\n                certs[\"verify\"] = get_cert(self._config, repository_name)\n                break\n\n        return certs\n", "path": "src/poetry/utils/authenticator.py"}]}