Issue #3483 Add cert retrieval for requests

src/poetry/utils/authenticator.py

@@ -5,6 +5,7 @@
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
+from typing import Generator
 from typing import Optional
 from typing import Tuple
 
@@ -13,10 +14,14 @@
 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
@@ -31,6 +36,7 @@ def __init__(self, config: "Config", io: Optional["IO"] = None) -> None:
         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:
@@ -62,8 +68,16 @@ def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
 
         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
@@ -168,7 +182,7 @@ def _get_credentials_for_netloc(
     ) -> Tuple[Optional[str], Optional[str]]:
         credentials = (None, 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:
@@ -178,6 +192,25 @@ def _get_credentials_for_netloc(
 
         return credentials
 
+    def get_certs_for_url(self, url: str) -> Dict[str, "Path"]:
+        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) -> Generator[Tuple[str, str], None, None]:
+        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: Optional[str]
     ) -> Optional[Dict[str, str]]:
@@ -204,3 +237,14 @@ def _get_credentials_for_netloc_from_keyring(
             }
 
         return None
+
+    def _get_certs_for_netloc_from_config(self, netloc: str) -> Dict[str, "Path"]:
+        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

tests/utils/test_authenticator.py

@@ -1,6 +1,7 @@
 import re
 import uuid
 
+from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
@@ -308,3 +309,46 @@ def test_authenticator_uses_env_provided_credentials(
     request = http.last_request()
 
     assert request.headers["Authorization"] == "Basic YmFyOmJheg=="
+
+
+@pytest.mark.parametrize(
+    "cert,client_cert",
+    [
+        (None, None),
+        (None, "path/to/provided/client-cert"),
+        ("/path/to/provided/cert", None),
+        ("/path/to/provided/cert", "path/to/provided/client-cert"),
+    ],
+)
+def test_authenticator_uses_certs_from_config_if_not_provided(
+    config: "Config",
+    mock_remote: Type[httpretty.httpretty],
+    http: Type[httpretty.httpretty],
+    mocker: "MockerFixture",
+    cert: Union[str, None],
+    client_cert: Union[str, None],
+):
+    configured_cert = "/path/to/cert"
+    configured_client_cert = "/path/to/client-cert"
+    config.merge(
+        {
+            "repositories": {"foo": {"url": "https://foo.bar/simple/"}},
+            "http-basic": {"foo": {"username": "bar", "password": "baz"}},
+            "certificates": {
+                "foo": {"cert": configured_cert, "client-cert": configured_client_cert}
+            },
+        }
+    )
+
+    authenticator = Authenticator(config, NullIO())
+    session_send = mocker.patch.object(authenticator.session, "send")
+    authenticator.request(
+        "get",
+        "https://foo.bar/files/foo-0.1.0.tar.gz",
+        verify=cert,
+        cert=client_cert,
+    )
+    kwargs = session_send.call_args[1]
+
+    assert Path(kwargs["verify"]) == Path(cert or configured_cert)
+    assert Path(kwargs["cert"]) == Path(client_cert or configured_client_cert)