Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_internal/network/session.py: 56%
193 statements
« prev ^ index » next coverage.py v7.4.3, created at 2024-02-26 06:33 +0000
1"""PipSession and supporting code, containing all pip-specific
2network request configuration and behavior.
3"""
5import email.utils
6import io
7import ipaddress
8import json
9import logging
10import mimetypes
11import os
12import platform
13import shutil
14import subprocess
15import sys
16import urllib.parse
17import warnings
18from typing import (
19 TYPE_CHECKING,
20 Any,
21 Dict,
22 Generator,
23 List,
24 Mapping,
25 Optional,
26 Sequence,
27 Tuple,
28 Union,
29)
31from pip._vendor import requests, urllib3
32from pip._vendor.cachecontrol import CacheControlAdapter as _BaseCacheControlAdapter
33from pip._vendor.requests.adapters import DEFAULT_POOLBLOCK, BaseAdapter
34from pip._vendor.requests.adapters import HTTPAdapter as _BaseHTTPAdapter
35from pip._vendor.requests.models import PreparedRequest, Response
36from pip._vendor.requests.structures import CaseInsensitiveDict
37from pip._vendor.urllib3.connectionpool import ConnectionPool
38from pip._vendor.urllib3.exceptions import InsecureRequestWarning
40from pip import __version__
41from pip._internal.metadata import get_default_environment
42from pip._internal.models.link import Link
43from pip._internal.network.auth import MultiDomainBasicAuth
44from pip._internal.network.cache import SafeFileCache
46# Import ssl from compat so the initial import occurs in only one place.
47from pip._internal.utils.compat import has_tls
48from pip._internal.utils.glibc import libc_ver
49from pip._internal.utils.misc import build_url_from_netloc, parse_netloc
50from pip._internal.utils.urls import url_to_path
52if TYPE_CHECKING:
53 from ssl import SSLContext
55 from pip._vendor.urllib3.poolmanager import PoolManager
58logger = logging.getLogger(__name__)
60SecureOrigin = Tuple[str, str, Optional[Union[int, str]]]
63# Ignore warning raised when using --trusted-host.
64warnings.filterwarnings("ignore", category=InsecureRequestWarning)
67SECURE_ORIGINS: List[SecureOrigin] = [
68 # protocol, hostname, port
69 # Taken from Chrome's list of secure origins (See: http://bit.ly/1qrySKC)
70 ("https", "*", "*"),
71 ("*", "localhost", "*"),
72 ("*", "127.0.0.0/8", "*"),
73 ("*", "::1/128", "*"),
74 ("file", "*", None),
75 # ssh is always secure.
76 ("ssh", "*", "*"),
77]
80# These are environment variables present when running under various
81# CI systems. For each variable, some CI systems that use the variable
82# are indicated. The collection was chosen so that for each of a number
83# of popular systems, at least one of the environment variables is used.
84# This list is used to provide some indication of and lower bound for
85# CI traffic to PyPI. Thus, it is okay if the list is not comprehensive.
86# For more background, see: https://github.com/pypa/pip/issues/5499
87CI_ENVIRONMENT_VARIABLES = (
88 # Azure Pipelines
89 "BUILD_BUILDID",
90 # Jenkins
91 "BUILD_ID",
92 # AppVeyor, CircleCI, Codeship, Gitlab CI, Shippable, Travis CI
93 "CI",
94 # Explicit environment variable.
95 "PIP_IS_CI",
96)
99def looks_like_ci() -> bool:
100 """
101 Return whether it looks like pip is running under CI.
102 """
103 # We don't use the method of checking for a tty (e.g. using isatty())
104 # because some CI systems mimic a tty (e.g. Travis CI). Thus that
105 # method doesn't provide definitive information in either direction.
106 return any(name in os.environ for name in CI_ENVIRONMENT_VARIABLES)
109def user_agent() -> str:
110 """
111 Return a string representing the user agent.
112 """
113 data: Dict[str, Any] = {
114 "installer": {"name": "pip", "version": __version__},
115 "python": platform.python_version(),
116 "implementation": {
117 "name": platform.python_implementation(),
118 },
119 }
121 if data["implementation"]["name"] == "CPython":
122 data["implementation"]["version"] = platform.python_version()
123 elif data["implementation"]["name"] == "PyPy":
124 pypy_version_info = sys.pypy_version_info # type: ignore
125 if pypy_version_info.releaselevel == "final":
126 pypy_version_info = pypy_version_info[:3]
127 data["implementation"]["version"] = ".".join(
128 [str(x) for x in pypy_version_info]
129 )
130 elif data["implementation"]["name"] == "Jython":
131 # Complete Guess
132 data["implementation"]["version"] = platform.python_version()
133 elif data["implementation"]["name"] == "IronPython":
134 # Complete Guess
135 data["implementation"]["version"] = platform.python_version()
137 if sys.platform.startswith("linux"):
138 from pip._vendor import distro
140 linux_distribution = distro.name(), distro.version(), distro.codename()
141 distro_infos: Dict[str, Any] = dict(
142 filter(
143 lambda x: x[1],
144 zip(["name", "version", "id"], linux_distribution),
145 )
146 )
147 libc = dict(
148 filter(
149 lambda x: x[1],
150 zip(["lib", "version"], libc_ver()),
151 )
152 )
153 if libc:
154 distro_infos["libc"] = libc
155 if distro_infos:
156 data["distro"] = distro_infos
158 if sys.platform.startswith("darwin") and platform.mac_ver()[0]:
159 data["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]}
161 if platform.system():
162 data.setdefault("system", {})["name"] = platform.system()
164 if platform.release():
165 data.setdefault("system", {})["release"] = platform.release()
167 if platform.machine():
168 data["cpu"] = platform.machine()
170 if has_tls():
171 import _ssl as ssl
173 data["openssl_version"] = ssl.OPENSSL_VERSION
175 setuptools_dist = get_default_environment().get_distribution("setuptools")
176 if setuptools_dist is not None:
177 data["setuptools_version"] = str(setuptools_dist.version)
179 if shutil.which("rustc") is not None:
180 # If for any reason `rustc --version` fails, silently ignore it
181 try:
182 rustc_output = subprocess.check_output(
183 ["rustc", "--version"], stderr=subprocess.STDOUT, timeout=0.5
184 )
185 except Exception:
186 pass
187 else:
188 if rustc_output.startswith(b"rustc "):
189 # The format of `rustc --version` is:
190 # `b'rustc 1.52.1 (9bc8c42bb 2021-05-09)\n'`
191 # We extract just the middle (1.52.1) part
192 data["rustc_version"] = rustc_output.split(b" ")[1].decode()
194 # Use None rather than False so as not to give the impression that
195 # pip knows it is not being run under CI. Rather, it is a null or
196 # inconclusive result. Also, we include some value rather than no
197 # value to make it easier to know that the check has been run.
198 data["ci"] = True if looks_like_ci() else None
200 user_data = os.environ.get("PIP_USER_AGENT_USER_DATA")
201 if user_data is not None:
202 data["user_data"] = user_data
204 return "{data[installer][name]}/{data[installer][version]} {json}".format(
205 data=data,
206 json=json.dumps(data, separators=(",", ":"), sort_keys=True),
207 )
210class LocalFSAdapter(BaseAdapter):
211 def send(
212 self,
213 request: PreparedRequest,
214 stream: bool = False,
215 timeout: Optional[Union[float, Tuple[float, float]]] = None,
216 verify: Union[bool, str] = True,
217 cert: Optional[Union[str, Tuple[str, str]]] = None,
218 proxies: Optional[Mapping[str, str]] = None,
219 ) -> Response:
220 pathname = url_to_path(request.url)
222 resp = Response()
223 resp.status_code = 200
224 resp.url = request.url
226 try:
227 stats = os.stat(pathname)
228 except OSError as exc:
229 # format the exception raised as a io.BytesIO object,
230 # to return a better error message:
231 resp.status_code = 404
232 resp.reason = type(exc).__name__
233 resp.raw = io.BytesIO(f"{resp.reason}: {exc}".encode())
234 else:
235 modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
236 content_type = mimetypes.guess_type(pathname)[0] or "text/plain"
237 resp.headers = CaseInsensitiveDict(
238 {
239 "Content-Type": content_type,
240 "Content-Length": stats.st_size,
241 "Last-Modified": modified,
242 }
243 )
245 resp.raw = open(pathname, "rb")
246 resp.close = resp.raw.close
248 return resp
250 def close(self) -> None:
251 pass
254class _SSLContextAdapterMixin:
255 """Mixin to add the ``ssl_context`` constructor argument to HTTP adapters.
257 The additional argument is forwarded directly to the pool manager. This allows us
258 to dynamically decide what SSL store to use at runtime, which is used to implement
259 the optional ``truststore`` backend.
260 """
262 def __init__(
263 self,
264 *,
265 ssl_context: Optional["SSLContext"] = None,
266 **kwargs: Any,
267 ) -> None:
268 self._ssl_context = ssl_context
269 super().__init__(**kwargs)
271 def init_poolmanager(
272 self,
273 connections: int,
274 maxsize: int,
275 block: bool = DEFAULT_POOLBLOCK,
276 **pool_kwargs: Any,
277 ) -> "PoolManager":
278 if self._ssl_context is not None:
279 pool_kwargs.setdefault("ssl_context", self._ssl_context)
280 return super().init_poolmanager( # type: ignore[misc]
281 connections=connections,
282 maxsize=maxsize,
283 block=block,
284 **pool_kwargs,
285 )
288class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter):
289 pass
292class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter):
293 pass
296class InsecureHTTPAdapter(HTTPAdapter):
297 def cert_verify(
298 self,
299 conn: ConnectionPool,
300 url: str,
301 verify: Union[bool, str],
302 cert: Optional[Union[str, Tuple[str, str]]],
303 ) -> None:
304 super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
307class InsecureCacheControlAdapter(CacheControlAdapter):
308 def cert_verify(
309 self,
310 conn: ConnectionPool,
311 url: str,
312 verify: Union[bool, str],
313 cert: Optional[Union[str, Tuple[str, str]]],
314 ) -> None:
315 super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
318class PipSession(requests.Session):
319 timeout: Optional[int] = None
321 def __init__(
322 self,
323 *args: Any,
324 retries: int = 0,
325 cache: Optional[str] = None,
326 trusted_hosts: Sequence[str] = (),
327 index_urls: Optional[List[str]] = None,
328 ssl_context: Optional["SSLContext"] = None,
329 **kwargs: Any,
330 ) -> None:
331 """
332 :param trusted_hosts: Domains not to emit warnings for when not using
333 HTTPS.
334 """
335 super().__init__(*args, **kwargs)
337 # Namespace the attribute with "pip_" just in case to prevent
338 # possible conflicts with the base class.
339 self.pip_trusted_origins: List[Tuple[str, Optional[int]]] = []
341 # Attach our User Agent to the request
342 self.headers["User-Agent"] = user_agent()
344 # Attach our Authentication handler to the session
345 self.auth = MultiDomainBasicAuth(index_urls=index_urls)
347 # Create our urllib3.Retry instance which will allow us to customize
348 # how we handle retries.
349 retries = urllib3.Retry(
350 # Set the total number of retries that a particular request can
351 # have.
352 total=retries,
353 # A 503 error from PyPI typically means that the Fastly -> Origin
354 # connection got interrupted in some way. A 503 error in general
355 # is typically considered a transient error so we'll go ahead and
356 # retry it.
357 # A 500 may indicate transient error in Amazon S3
358 # A 502 may be a transient error from a CDN like CloudFlare or CloudFront
359 # A 520 or 527 - may indicate transient error in CloudFlare
360 status_forcelist=[500, 502, 503, 520, 527],
361 # Add a small amount of back off between failed requests in
362 # order to prevent hammering the service.
363 backoff_factor=0.25,
364 ) # type: ignore
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 if cache:
379 secure_adapter = CacheControlAdapter(
380 cache=SafeFileCache(cache),
381 max_retries=retries,
382 ssl_context=ssl_context,
383 )
384 self._trusted_host_adapter = InsecureCacheControlAdapter(
385 cache=SafeFileCache(cache),
386 max_retries=retries,
387 )
388 else:
389 secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context)
390 self._trusted_host_adapter = insecure_adapter
392 self.mount("https://", secure_adapter)
393 self.mount("http://", insecure_adapter)
395 # Enable file:// urls
396 self.mount("file://", LocalFSAdapter())
398 for host in trusted_hosts:
399 self.add_trusted_host(host, suppress_logging=True)
401 def update_index_urls(self, new_index_urls: List[str]) -> None:
402 """
403 :param new_index_urls: New index urls to update the authentication
404 handler with.
405 """
406 self.auth.index_urls = new_index_urls
408 def add_trusted_host(
409 self, host: str, source: Optional[str] = None, suppress_logging: bool = False
410 ) -> None:
411 """
412 :param host: It is okay to provide a host that has previously been
413 added.
414 :param source: An optional source string, for logging where the host
415 string came from.
416 """
417 if not suppress_logging:
418 msg = f"adding trusted host: {host!r}"
419 if source is not None:
420 msg += f" (from {source})"
421 logger.info(msg)
423 parsed_host, parsed_port = parse_netloc(host)
424 if parsed_host is None:
425 raise ValueError(f"Trusted host URL must include a host part: {host!r}")
426 if (parsed_host, parsed_port) not in self.pip_trusted_origins:
427 self.pip_trusted_origins.append((parsed_host, parsed_port))
429 self.mount(
430 build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter
431 )
432 self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter)
433 if not parsed_port:
434 self.mount(
435 build_url_from_netloc(host, scheme="http") + ":",
436 self._trusted_host_adapter,
437 )
438 # Mount wildcard ports for the same host.
439 self.mount(build_url_from_netloc(host) + ":", self._trusted_host_adapter)
441 def iter_secure_origins(self) -> Generator[SecureOrigin, None, None]:
442 yield from SECURE_ORIGINS
443 for host, port in self.pip_trusted_origins:
444 yield ("*", host, "*" if port is None else port)
446 def is_secure_origin(self, location: Link) -> bool:
447 # Determine if this url used a secure transport mechanism
448 parsed = urllib.parse.urlparse(str(location))
449 origin_protocol, origin_host, origin_port = (
450 parsed.scheme,
451 parsed.hostname,
452 parsed.port,
453 )
455 # The protocol to use to see if the protocol matches.
456 # Don't count the repository type as part of the protocol: in
457 # cases such as "git+ssh", only use "ssh". (I.e., Only verify against
458 # the last scheme.)
459 origin_protocol = origin_protocol.rsplit("+", 1)[-1]
461 # Determine if our origin is a secure origin by looking through our
462 # hardcoded list of secure origins, as well as any additional ones
463 # configured on this PackageFinder instance.
464 for secure_origin in self.iter_secure_origins():
465 secure_protocol, secure_host, secure_port = secure_origin
466 if origin_protocol != secure_protocol and secure_protocol != "*":
467 continue
469 try:
470 addr = ipaddress.ip_address(origin_host or "")
471 network = ipaddress.ip_network(secure_host)
472 except ValueError:
473 # We don't have both a valid address or a valid network, so
474 # we'll check this origin against hostnames.
475 if (
476 origin_host
477 and origin_host.lower() != secure_host.lower()
478 and secure_host != "*"
479 ):
480 continue
481 else:
482 # We have a valid address and network, so see if the address
483 # is contained within the network.
484 if addr not in network:
485 continue
487 # Check to see if the port matches.
488 if (
489 origin_port != secure_port
490 and secure_port != "*"
491 and secure_port is not None
492 ):
493 continue
495 # If we've gotten here, then this origin matches the current
496 # secure origin and we should return True
497 return True
499 # If we've gotten to this point, then the origin isn't secure and we
500 # will not accept it as a valid location to search. We will however
501 # log a warning that we are ignoring it.
502 logger.warning(
503 "The repository located at %s is not a trusted or secure host and "
504 "is being ignored. If this repository is available via HTTPS we "
505 "recommend you use HTTPS instead, otherwise you may silence "
506 "this warning and allow it anyway with '--trusted-host %s'.",
507 origin_host,
508 origin_host,
509 )
511 return False
513 def request(self, method: str, url: str, *args: Any, **kwargs: Any) -> Response:
514 # Allow setting a default timeout on a session
515 kwargs.setdefault("timeout", self.timeout)
516 # Allow setting a default proxies on a session
517 kwargs.setdefault("proxies", self.proxies)
519 # Dispatch the actual request
520 return super().request(method, url, *args, **kwargs)