improve http request handling for sources and support multiple paths on same netloc

src/poetry/config/config.py

@@ -112,6 +112,20 @@ def _all(config: dict, parent_key: str = "") -> dict:
     def raw(self) -> dict[str, Any]:
         return self._config
 
+    @staticmethod
+    def _get_environment_repositories() -> dict[str, dict[str, str]]:
+        repositories = {}
+        pattern = re.compile(r"POETRY_REPOSITORIES_(?P<name>[A-Z_]+)_URL")
+
+        for env_key in os.environ.keys():
+            match = pattern.match(env_key)
+            if match:
+                repositories[match.group("name").lower().replace("_", "-")] = {
+                    "url": os.environ[env_key]
+                }
+
+        return repositories
+
     def get(self, setting_name: str, default: Any = None) -> Any:
         """
         Retrieve a setting value.
@@ -121,6 +135,12 @@ def get(self, setting_name: str, default: Any = None) -> Any:
         # Looking in the environment if the setting
         # is set via a POETRY_* environment variable
         if self._use_environment:
+            if setting_name == "repositories":
+                # repositories setting is special for now
+                repositories = self._get_environment_repositories()
+                if repositories:
+                    return repositories
+
             env = "POETRY_" + "_".join(k.upper().replace("-", "_") for k in keys)
             env_value = os.getenv(env)
             if env_value is not None:

src/poetry/factory.py

@@ -184,8 +184,6 @@ def create_legacy_repository(
         cls, source: dict[str, str], auth_config: Config, disable_cache: bool = False
     ) -> LegacyRepository:
         from poetry.repositories.legacy_repository import LegacyRepository
-        from poetry.utils.helpers import get_cert
-        from poetry.utils.helpers import get_client_cert
 
         if "url" not in source:
             raise RuntimeError("Unsupported source specified")
@@ -200,8 +198,6 @@ def create_legacy_repository(
             name,
             url,
             config=auth_config,
-            cert=get_cert(auth_config, name),
-            client_cert=get_client_cert(auth_config, name),
             disable_cache=disable_cache,
         )
 

src/poetry/installation/pip_installer.py

@@ -12,6 +12,7 @@
 from poetry.core.pyproject.toml import PyProjectTOML
 
 from poetry.installation.base_installer import BaseInstaller
+from poetry.repositories.http import HTTPRepository
 from poetry.utils._compat import encode
 from poetry.utils.helpers import remove_directory
 from poetry.utils.pip import pip_install
@@ -57,23 +58,28 @@ def install(self, package: Package, update: bool = False) -> None:
                 )
                 args += ["--trusted-host", parsed.hostname]
 
-            if repository.cert:
-                args += ["--cert", str(repository.cert)]
+            if isinstance(repository, HTTPRepository):
+                if repository.cert:
+                    args += ["--cert", str(repository.cert)]
 
-            if repository.client_cert:
-                args += ["--client-cert", str(repository.client_cert)]
+                if repository.client_cert:
+                    args += ["--client-cert", str(repository.client_cert)]
 
-            index_url = repository.authenticated_url
+                index_url = repository.authenticated_url
+
+                args += ["--index-url", index_url]
 
-            args += ["--index-url", index_url]
             if (
                 self._pool.has_default()
                 and repository.name != self._pool.repositories[0].name
             ):
-                args += [
-                    "--extra-index-url",
-                    self._pool.repositories[0].authenticated_url,
-                ]
+                first_repository = self._pool.repositories[0]
+
+                if isinstance(first_repository, HTTPRepository):
+                    args += [
+                        "--extra-index-url",
+                        first_repository.authenticated_url,
+                    ]
 
         if update:
             args.append("-U")

src/poetry/publishing/publisher.py

@@ -69,14 +69,14 @@ def publish(
                     logger.debug(
                         f"Found authentication information for {repository_name}."
                     )
-                    username = auth["username"]
-                    password = auth["password"]
+                    username = auth.username
+                    password = auth.password
 
         resolved_client_cert = client_cert or get_client_cert(
             self._poetry.config, repository_name
         )
         # Requesting missing credentials but only if there is not a client cert defined.
-        if not resolved_client_cert:
+        if not resolved_client_cert and hasattr(self._io, "ask"):
             if username is None:
                 username = self._io.ask("Username:")
 

src/poetry/repositories/cached.py

@@ -4,7 +4,6 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING
 
-from cachecontrol.caches import FileCache
 from cachy import CacheManager
 from poetry.core.semver.helpers import parse_constraint
 
@@ -21,7 +20,7 @@
 class CachedRepository(Repository, ABC):
     CACHE_VERSION = parse_constraint("1.0.0")
 
-    def __init__(self, name: str, cache_group: str, disable_cache: bool = False):
+    def __init__(self, name: str, disable_cache: bool = False):
         super().__init__(name)
         self._disable_cache = disable_cache
         self._cache_dir = REPOSITORY_CACHE_DIR / name
@@ -36,7 +35,6 @@ def __init__(self, name: str, cache_group: str, disable_cache: bool = False):
                 },
             }
         )
-        self._cache_control_cache = FileCache(str(self._cache_dir / cache_group))
 
     @abstractmethod
     def _get_release_info(self, name: str, version: str) -> dict:

src/poetry/repositories/http.py

@@ -1,21 +1,18 @@
 from __future__ import annotations
 
-import contextlib
 import hashlib
 import os
 import urllib
+import urllib.parse
 
 from abc import ABC
 from collections import defaultdict
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
-from urllib.parse import quote
 
 import requests
-import requests.auth
 
-from cachecontrol import CacheControl
 from poetry.core.packages.dependency import Dependency
 from poetry.core.packages.utils.link import Link
 from poetry.core.version.markers import parse_marker
@@ -42,64 +39,40 @@ def __init__(
         url: str,
         config: Config | None = None,
         disable_cache: bool = False,
-        cert: Path | None = None,
-        client_cert: Path | None = None,
     ) -> None:
-        super().__init__(name, "_http", disable_cache)
+        super().__init__(name, disable_cache)
         self._url = url
-        self._client_cert = client_cert
-        self._cert = cert
-
         self._authenticator = Authenticator(
-            config=config or Config(use_environment=True)
-        )
-
-        self._session = CacheControl(
-            self._authenticator.session, cache=self._cache_control_cache
+            config=config or Config(use_environment=True),
+            cache_id=name,
+            disable_cache=disable_cache,
         )
 
-        username, password = self._authenticator.get_credentials_for_url(self._url)
-        if username is not None and password is not None:
-            self._authenticator.session.auth = requests.auth.HTTPBasicAuth(
-                username, password
-            )
-
-        if self._cert:
-            self._authenticator.session.verify = str(self._cert)
-
-        if self._client_cert:
-            self._authenticator.session.cert = str(self._client_cert)
-
     @property
-    def session(self) -> CacheControl:
-        return self._session
-
-    def __del__(self) -> None:
-        with contextlib.suppress(AttributeError):
-            self._session.close()
+    def session(self) -> Authenticator:
+        return self._authenticator
 
     @property
     def url(self) -> str:
         return self._url
 
     @property
     def cert(self) -> Path | None:
-        return self._cert
+        cert = self._authenticator.get_certs_for_url(self.url).get("verify")
+        if cert:
+            return Path(cert)
+        return None
 
     @property
     def client_cert(self) -> Path | None:
-        return self._client_cert
+        cert = self._authenticator.get_certs_for_url(self.url).get("cert")
+        if cert:
+            return Path(cert)
+        return None
 
     @property
     def authenticated_url(self) -> str:
-        if not self._session.auth:
-            return self.url
-
-        parsed = urllib.parse.urlparse(self.url)
-        username = quote(self._session.auth.username, safe="")
-        password = quote(self._session.auth.password, safe="")
-
-        return f"{parsed.scheme}://{username}:{password}@{parsed.netloc}{parsed.path}"
+        return self._authenticator.authenticated_url(url=self.url)
 
     def _download(self, url: str, dest: str) -> None:
         return download_file(url, dest, session=self.session)
@@ -286,7 +259,7 @@ def _links_to_data(self, links: list[Link], data: PackageInfo) -> dict:
     def _get_response(self, endpoint: str) -> requests.Response | None:
         url = self._url + endpoint
         try:
-            response = self.session.get(url)
+            response = self.session.get(url, raise_for_status=False)
             if response.status_code in (401, 403):
                 self._log(
                     f"Authorization error accessing {url}",

src/poetry/repositories/legacy_repository.py

@@ -13,8 +13,6 @@
 
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from poetry.core.packages.dependency import Dependency
     from poetry.core.packages.utils.link import Link
 
@@ -28,15 +26,11 @@ def __init__(
         url: str,
         config: Config | None = None,
         disable_cache: bool = False,
-        cert: Path | None = None,
-        client_cert: Path | None = None,
     ) -> None:
         if name == "pypi":
             raise ValueError("The name [pypi] is reserved for repositories")
 
-        super().__init__(
-            name, url.rstrip("/"), config, disable_cache, cert, client_cert
-        )
+        super().__init__(name, url.rstrip("/"), config, disable_cache)
 
     def find_packages(self, dependency: Dependency) -> list[Package]:
         packages = []

src/poetry/repositories/pypi_repository.py

@@ -232,7 +232,7 @@ def _get(self, endpoint: str) -> dict | None:
         except requests.exceptions.TooManyRedirects:
             # Cache control redirect loop.
             # We try to remove the cache and try again
-            self._cache_control_cache.delete(self._base_url + endpoint)
+            self.session.delete_cache(self._base_url + endpoint)
             json_response = self.session.get(self._base_url + endpoint)
 
         if json_response.status_code == 404: