Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""PipSession and supporting code, containing all pip-specific
2network request configuration and behavior.
3"""
5from __future__ import annotations
7import email.utils
8import functools
9import io
10import ipaddress
11import json
12import logging
13import mimetypes
14import os
15import platform
16import shutil
17import subprocess
18import sys
19import urllib.parse
20import warnings
21from collections.abc import Generator, Mapping, Sequence
22from typing import (
23 TYPE_CHECKING,
24 Any,
25)
27from pip._vendor import requests, urllib3
28from pip._vendor.cachecontrol import CacheControlAdapter as _BaseCacheControlAdapter
29from pip._vendor.requests.adapters import DEFAULT_POOLBLOCK, BaseAdapter
30from pip._vendor.requests.adapters import HTTPAdapter as _BaseHTTPAdapter
31from pip._vendor.requests.models import PreparedRequest, Response
32from pip._vendor.requests.structures import CaseInsensitiveDict
33from pip._vendor.urllib3.connectionpool import ConnectionPool
34from pip._vendor.urllib3.exceptions import InsecureRequestWarning
36from pip import __version__
37from pip._internal.exceptions import SSLMissingError
38from pip._internal.metadata import get_default_environment
39from pip._internal.models.link import Link
40from pip._internal.network.auth import MultiDomainBasicAuth
41from pip._internal.network.cache import SafeFileCache
42from pip._internal.network.utils import raise_connection_error
44# Import ssl from compat so the initial import occurs in only one place.
45from pip._internal.utils.compat import has_tls
46from pip._internal.utils.glibc import libc_ver
47from pip._internal.utils.misc import (
48 build_url_from_netloc,
49 looks_like_ci,
50 parse_netloc,
51 redact_auth_from_url,
52)
53from pip._internal.utils.urls import url_to_path
55if TYPE_CHECKING:
56 from ssl import SSLContext
58 from pip._vendor.urllib3 import ProxyManager
61logger = logging.getLogger(__name__)
63SecureOrigin = tuple[str, str, int | str | None]
66# Ignore warning raised when using --trusted-host.
67warnings.filterwarnings("ignore", category=InsecureRequestWarning)
70SECURE_ORIGINS: list[SecureOrigin] = [
71 # protocol, hostname, port
72 # Taken from Chrome's list of secure origins (See: http://bit.ly/1qrySKC)
73 ("https", "*", "*"),
74 ("*", "localhost", "*"),
75 ("*", "127.0.0.0/8", "*"),
76 ("*", "::1/128", "*"),
77 ("file", "*", None),
78 # ssh is always secure.
79 ("ssh", "*", "*"),
80]
83@functools.lru_cache(maxsize=1)
84def user_agent() -> str:
85 """
86 Return a string representing the user agent.
87 """
88 data: dict[str, Any] = {
89 "installer": {"name": "pip", "version": __version__},
90 "python": platform.python_version(),
91 "implementation": {
92 "name": platform.python_implementation(),
93 },
94 }
96 if data["implementation"]["name"] == "CPython":
97 data["implementation"]["version"] = platform.python_version()
98 elif data["implementation"]["name"] == "PyPy":
99 pypy_version_info = sys.pypy_version_info # type: ignore
100 if pypy_version_info.releaselevel == "final":
101 pypy_version_info = pypy_version_info[:3]
102 data["implementation"]["version"] = ".".join(
103 [str(x) for x in pypy_version_info]
104 )
105 elif data["implementation"]["name"] == "Jython":
106 # Complete Guess
107 data["implementation"]["version"] = platform.python_version()
108 elif data["implementation"]["name"] == "IronPython":
109 # Complete Guess
110 data["implementation"]["version"] = platform.python_version()
112 if sys.platform.startswith("linux"):
113 from pip._vendor import distro
115 linux_distribution = distro.name(), distro.version(), distro.codename()
116 distro_infos: dict[str, Any] = dict(
117 filter(
118 lambda x: x[1],
119 zip(["name", "version", "id"], linux_distribution),
120 )
121 )
122 libc = dict(
123 filter(
124 lambda x: x[1],
125 zip(["lib", "version"], libc_ver()),
126 )
127 )
128 if libc:
129 distro_infos["libc"] = libc
130 if distro_infos:
131 data["distro"] = distro_infos
133 if sys.platform.startswith("darwin") and platform.mac_ver()[0]:
134 data["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]}
136 if platform.system():
137 data.setdefault("system", {})["name"] = platform.system()
139 if platform.release():
140 data.setdefault("system", {})["release"] = platform.release()
142 if platform.machine():
143 data["cpu"] = platform.machine()
145 if has_tls():
146 import _ssl as ssl
148 data["openssl_version"] = ssl.OPENSSL_VERSION
150 setuptools_dist = get_default_environment().get_distribution("setuptools")
151 if setuptools_dist is not None:
152 data["setuptools_version"] = str(setuptools_dist.version)
154 if shutil.which("rustc") is not None:
155 # If for any reason `rustc --version` fails, silently ignore it
156 try:
157 rustc_output = subprocess.check_output(
158 ["rustc", "--version"], stderr=subprocess.STDOUT, timeout=0.5
159 )
160 except Exception:
161 pass
162 else:
163 if rustc_output.startswith(b"rustc "):
164 # The format of `rustc --version` is:
165 # `b'rustc 1.52.1 (9bc8c42bb 2021-05-09)\n'`
166 # We extract just the middle (1.52.1) part
167 data["rustc_version"] = rustc_output.split(b" ")[1].decode()
169 # Use None rather than False so as not to give the impression that
170 # pip knows it is not being run under CI. Rather, it is a null or
171 # inconclusive result. Also, we include some value rather than no
172 # value to make it easier to know that the check has been run.
173 data["ci"] = True if looks_like_ci() else None
175 user_data = os.environ.get("PIP_USER_AGENT_USER_DATA")
176 if user_data is not None:
177 data["user_data"] = user_data
179 return "{data[installer][name]}/{data[installer][version]} {json}".format(
180 data=data,
181 json=json.dumps(data, separators=(",", ":"), sort_keys=True),
182 )
185class LocalFSAdapter(BaseAdapter):
186 def send(
187 self,
188 request: PreparedRequest,
189 stream: bool = False,
190 timeout: float | tuple[float | None, float | None] | None = None,
191 verify: bool | str = True,
192 cert: bytes | str | tuple[bytes | str, bytes | str] | None = None,
193 proxies: Mapping[str, str] | None = None,
194 ) -> Response:
195 assert request.url is not None
196 pathname = url_to_path(request.url)
198 resp = Response()
199 resp.status_code = 200
200 resp.url = request.url
202 try:
203 stats = os.stat(pathname)
204 except OSError as exc:
205 # format the exception raised as a io.BytesIO object,
206 # to return a better error message:
207 resp.status_code = 404
208 resp.reason = type(exc).__name__
209 resp.raw = io.BytesIO(f"{resp.reason}: {exc}".encode())
210 else:
211 modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
212 content_type = mimetypes.guess_type(pathname)[0] or "text/plain"
213 resp.headers = CaseInsensitiveDict(
214 {
215 "Content-Type": content_type,
216 "Content-Length": str(stats.st_size),
217 "Last-Modified": modified,
218 }
219 )
221 resp.raw = open(pathname, "rb")
222 resp.close = resp.raw.close # type: ignore[method-assign]
224 return resp
226 def close(self) -> None:
227 pass
230class _SSLContextAdapterMixin:
231 """Mixin to add the ``ssl_context`` constructor argument to HTTP adapters.
233 The additional argument is forwarded directly to the pool manager. This allows us
234 to dynamically decide what SSL store to use at runtime, which is used to implement
235 the optional ``truststore`` backend.
236 """
238 def __init__(
239 self,
240 *,
241 ssl_context: SSLContext | None = None,
242 **kwargs: Any,
243 ) -> None:
244 self._ssl_context = ssl_context
245 super().__init__(**kwargs)
247 def init_poolmanager(
248 self,
249 connections: int,
250 maxsize: int,
251 block: bool = DEFAULT_POOLBLOCK,
252 **pool_kwargs: Any,
253 ) -> None:
254 if self._ssl_context is not None:
255 pool_kwargs.setdefault("ssl_context", self._ssl_context)
256 super().init_poolmanager( # type: ignore[misc]
257 connections=connections,
258 maxsize=maxsize,
259 block=block,
260 **pool_kwargs,
261 )
263 def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> ProxyManager:
264 # Proxy manager replaces the pool manager, so inject our SSL
265 # context here too. https://github.com/pypa/pip/issues/13288
266 if self._ssl_context is not None:
267 proxy_kwargs.setdefault("ssl_context", self._ssl_context)
268 # For HTTPS proxies, urllib3 also opens a separate TLS connection
269 # to the proxy itself (before tunnelling to the destination) and
270 # uses "proxy_ssl_context" for that handshake.
271 # https://github.com/pypa/pip/issues/13465
272 proxy_kwargs.setdefault("proxy_ssl_context", self._ssl_context)
273 return super().proxy_manager_for(proxy, **proxy_kwargs) # type: ignore[misc]
276class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter):
277 pass
280class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter):
281 pass
284class InsecureHTTPAdapter(HTTPAdapter):
285 def cert_verify(
286 self,
287 conn: ConnectionPool,
288 url: str,
289 verify: bool | str,
290 cert: str | tuple[str, str] | None,
291 ) -> None:
292 super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
295class InsecureCacheControlAdapter(CacheControlAdapter):
296 def cert_verify(
297 self,
298 conn: ConnectionPool,
299 url: str,
300 verify: bool | str,
301 cert: str | tuple[str, str] | None,
302 ) -> None:
303 super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
306class PipSession(requests.Session):
307 timeout: int | None = None
309 def __init__(
310 self,
311 *args: Any,
312 retries: int = 0,
313 resume_retries: int = 0,
314 cache: str | None = None,
315 trusted_hosts: Sequence[str] = (),
316 index_urls: list[str] | None = None,
317 ssl_context: SSLContext | None = None,
318 refresh_package: set[str] | None = None,
319 **kwargs: Any,
320 ) -> None:
321 """
322 :param trusted_hosts: Domains not to emit warnings for when not using
323 HTTPS.
324 """
325 super().__init__(*args, **kwargs)
327 # Namespace the attribute with "pip_" just in case to prevent
328 # possible conflicts with the base class.
329 self.pip_trusted_origins: list[tuple[str, int | None]] = []
330 # "" disables proxying; None means no --proxy was given.
331 self.pip_proxy: str | None = None
332 self.pip_no_proxy_env = False
333 self.refresh_package: set[str] = refresh_package or set()
335 # Attach our User Agent to the request
336 self.headers["User-Agent"] = user_agent()
338 # Pin Accept-Encoding so it doesn't vary with zstd availability (Python
339 # 3.14+ or backports.zstd); a varying value misses the cache for "Vary:
340 # Accept-Encoding" responses shared across interpreters (pypa/pip#13979).
341 self.headers["Accept-Encoding"] = "gzip, deflate"
343 # Attach our Authentication handler to the session
344 self.auth: MultiDomainBasicAuth = MultiDomainBasicAuth(index_urls=index_urls)
346 # Create our urllib3.Retry instance which will allow us to customize
347 # how we handle retries.
348 retries = urllib3.Retry(
349 # Set the total number of retries that a particular request can
350 # have.
351 total=retries,
352 # A 503 error from PyPI typically means that the Fastly -> Origin
353 # connection got interrupted in some way. A 503 error in general
354 # is typically considered a transient error so we'll go ahead and
355 # retry it.
356 # A 500 may indicate transient error in Amazon S3
357 # A 502 may be a transient error from a CDN like CloudFlare or CloudFront
358 # A 520 or 527 - may indicate transient error in CloudFlare
359 status_forcelist=[500, 502, 503, 520, 527],
360 # Add a small amount of back off between failed requests in
361 # order to prevent hammering the service.
362 backoff_factor=0.25,
363 ) # type: ignore
364 self.resume_retries = resume_retries
366 # Our Insecure HTTPAdapter disables HTTPS validation. It does not
367 # support caching so we'll use it for all http:// URLs.
368 # If caching is disabled, we will also use it for
369 # https:// hosts that we've marked as ignoring
370 # TLS errors for (trusted-hosts).
371 insecure_adapter = InsecureHTTPAdapter(max_retries=retries)
373 # We want to _only_ cache responses on securely fetched origins or when
374 # the host is specified as trusted. We do this because
375 # we can't validate the response of an insecurely/untrusted fetched
376 # origin, and we don't want someone to be able to poison the cache and
377 # require manual eviction from the cache to fix it.
378 self._trusted_host_adapter: InsecureCacheControlAdapter | InsecureHTTPAdapter
379 if cache:
380 secure_adapter: _BaseHTTPAdapter = CacheControlAdapter(
381 cache=SafeFileCache(cache),
382 max_retries=retries,
383 ssl_context=ssl_context,
384 )
385 self._trusted_host_adapter = InsecureCacheControlAdapter(
386 cache=SafeFileCache(cache),
387 max_retries=retries,
388 )
389 else:
390 secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context)
391 self._trusted_host_adapter = insecure_adapter
393 self.mount("https://", secure_adapter)
394 self.mount("http://", insecure_adapter)
396 # Enable file:// urls
397 self.mount("file://", LocalFSAdapter())
399 for host in trusted_hosts:
400 self.add_trusted_host(host, suppress_logging=True)
402 def update_index_urls(self, new_index_urls: list[str]) -> None:
403 """
404 :param new_index_urls: New index urls to update the authentication
405 handler with.
406 """
407 self.auth.index_urls = new_index_urls
409 def add_trusted_host(
410 self, host: str, source: str | None = None, suppress_logging: bool = False
411 ) -> None:
412 """
413 :param host: It is okay to provide a host that has previously been
414 added.
415 :param source: An optional source string, for logging where the host
416 string came from.
417 """
418 if not suppress_logging:
419 msg = f"adding trusted host: {host!r}"
420 if source is not None:
421 msg += f" (from {source})"
422 logger.info(msg)
424 parsed_host, parsed_port = parse_netloc(host)
425 if parsed_host is None:
426 raise ValueError(f"Trusted host URL must include a host part: {host!r}")
427 if (parsed_host, parsed_port) not in self.pip_trusted_origins:
428 self.pip_trusted_origins.append((parsed_host, parsed_port))
430 self.mount(
431 build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter
432 )
433 self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter)
434 if not parsed_port:
435 self.mount(
436 build_url_from_netloc(host, scheme="http") + ":",
437 self._trusted_host_adapter,
438 )
439 # Mount wildcard ports for the same host.
440 self.mount(build_url_from_netloc(host) + ":", self._trusted_host_adapter)
442 def iter_secure_origins(self) -> Generator[SecureOrigin, None, None]:
443 yield from SECURE_ORIGINS
444 for host, port in self.pip_trusted_origins:
445 yield ("*", host, "*" if port is None else port)
447 def is_secure_origin(self, location: Link) -> bool:
448 # Determine if this url used a secure transport mechanism
449 parsed = urllib.parse.urlparse(str(location))
450 origin_protocol, origin_host, origin_port = (
451 parsed.scheme,
452 parsed.hostname,
453 parsed.port,
454 )
456 # The protocol to use to see if the protocol matches.
457 # Don't count the repository type as part of the protocol: in
458 # cases such as "git+ssh", only use "ssh". (I.e., Only verify against
459 # the last scheme.)
460 origin_protocol = origin_protocol.rsplit("+", 1)[-1]
462 # Determine if our origin is a secure origin by looking through our
463 # hardcoded list of secure origins, as well as any additional ones
464 # configured on this PackageFinder instance.
465 for secure_origin in self.iter_secure_origins():
466 secure_protocol, secure_host, secure_port = secure_origin
467 if origin_protocol != secure_protocol and secure_protocol != "*":
468 continue
470 try:
471 addr = ipaddress.ip_address(origin_host or "")
472 network = ipaddress.ip_network(secure_host)
473 except ValueError:
474 # We don't have both a valid address or a valid network, so
475 # we'll check this origin against hostnames.
476 if (
477 origin_host
478 and origin_host.lower() != secure_host.lower()
479 and secure_host != "*"
480 ):
481 continue
482 else:
483 # We have a valid address and network, so see if the address
484 # is contained within the network.
485 if addr not in network:
486 continue
488 # Check to see if the port matches.
489 if (
490 origin_port != secure_port
491 and secure_port != "*"
492 and secure_port is not None
493 ):
494 continue
496 # If we've gotten here, then this origin matches the current
497 # secure origin and we should return True
498 return True
500 # If we've gotten to this point, then the origin isn't secure and we
501 # will not accept it as a valid location to search. We will however
502 # log a warning that we are ignoring it.
503 logger.warning(
504 "The repository located at %s is not a trusted or secure host and "
505 "is being ignored. If this repository is available via HTTPS we "
506 "recommend you use HTTPS instead, otherwise you may silence "
507 "this warning and allow it anyway with '--trusted-host %s'.",
508 origin_host,
509 origin_host,
510 )
512 return False
514 def request(self, method: str, url: str, *args: Any, **kwargs: Any) -> Response: # type: ignore[override]
515 # Allow setting a default timeout on a session
516 kwargs.setdefault("timeout", self.timeout)
517 # Allow setting a default proxies on a session
518 kwargs.setdefault("proxies", self.proxies)
520 # Dispatch the actual request
521 try:
522 return super().request(method, url, *args, **kwargs)
523 except (requests.ConnectionError, requests.Timeout) as e:
524 request = getattr(e, "request", None)
525 failed_url = getattr(request, "url", None) or url
526 raise_connection_error(e, url=failed_url, timeout=kwargs["timeout"])
527 except ImportError as e:
528 if "ssl" in str(e).lower():
529 # Unfortunately, if this TLS error was the result of a redirect from
530 # a HTTP to a HTTPS url, we don't know what the final url was.
531 raise SSLMissingError(redact_auth_from_url(url))
532 raise