Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_internal/network/session.py: 57%
191 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:48 +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("utf8"))
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 520 or 527 - may indicate transient error in CloudFlare
359 status_forcelist=[500, 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
365 # Our Insecure HTTPAdapter disables HTTPS validation. It does not
366 # support caching so we'll use it for all http:// URLs.
367 # If caching is disabled, we will also use it for
368 # https:// hosts that we've marked as ignoring
369 # TLS errors for (trusted-hosts).
370 insecure_adapter = InsecureHTTPAdapter(max_retries=retries)
372 # We want to _only_ cache responses on securely fetched origins or when
373 # the host is specified as trusted. We do this because
374 # we can't validate the response of an insecurely/untrusted fetched
375 # origin, and we don't want someone to be able to poison the cache and
376 # require manual eviction from the cache to fix it.
377 if cache:
378 secure_adapter = CacheControlAdapter(
379 cache=SafeFileCache(cache),
380 max_retries=retries,
381 ssl_context=ssl_context,
382 )
383 self._trusted_host_adapter = InsecureCacheControlAdapter(
384 cache=SafeFileCache(cache),
385 max_retries=retries,
386 )
387 else:
388 secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context)
389 self._trusted_host_adapter = insecure_adapter
391 self.mount("https://", secure_adapter)
392 self.mount("http://", insecure_adapter)
394 # Enable file:// urls
395 self.mount("file://", LocalFSAdapter())
397 for host in trusted_hosts:
398 self.add_trusted_host(host, suppress_logging=True)
400 def update_index_urls(self, new_index_urls: List[str]) -> None:
401 """
402 :param new_index_urls: New index urls to update the authentication
403 handler with.
404 """
405 self.auth.index_urls = new_index_urls
407 def add_trusted_host(
408 self, host: str, source: Optional[str] = None, suppress_logging: bool = False
409 ) -> None:
410 """
411 :param host: It is okay to provide a host that has previously been
412 added.
413 :param source: An optional source string, for logging where the host
414 string came from.
415 """
416 if not suppress_logging:
417 msg = f"adding trusted host: {host!r}"
418 if source is not None:
419 msg += f" (from {source})"
420 logger.info(msg)
422 host_port = parse_netloc(host)
423 if host_port not in self.pip_trusted_origins:
424 self.pip_trusted_origins.append(host_port)
426 self.mount(
427 build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter
428 )
429 self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter)
430 if not host_port[1]:
431 self.mount(
432 build_url_from_netloc(host, scheme="http") + ":",
433 self._trusted_host_adapter,
434 )
435 # Mount wildcard ports for the same host.
436 self.mount(build_url_from_netloc(host) + ":", self._trusted_host_adapter)
438 def iter_secure_origins(self) -> Generator[SecureOrigin, None, None]:
439 yield from SECURE_ORIGINS
440 for host, port in self.pip_trusted_origins:
441 yield ("*", host, "*" if port is None else port)
443 def is_secure_origin(self, location: Link) -> bool:
444 # Determine if this url used a secure transport mechanism
445 parsed = urllib.parse.urlparse(str(location))
446 origin_protocol, origin_host, origin_port = (
447 parsed.scheme,
448 parsed.hostname,
449 parsed.port,
450 )
452 # The protocol to use to see if the protocol matches.
453 # Don't count the repository type as part of the protocol: in
454 # cases such as "git+ssh", only use "ssh". (I.e., Only verify against
455 # the last scheme.)
456 origin_protocol = origin_protocol.rsplit("+", 1)[-1]
458 # Determine if our origin is a secure origin by looking through our
459 # hardcoded list of secure origins, as well as any additional ones
460 # configured on this PackageFinder instance.
461 for secure_origin in self.iter_secure_origins():
462 secure_protocol, secure_host, secure_port = secure_origin
463 if origin_protocol != secure_protocol and secure_protocol != "*":
464 continue
466 try:
467 addr = ipaddress.ip_address(origin_host or "")
468 network = ipaddress.ip_network(secure_host)
469 except ValueError:
470 # We don't have both a valid address or a valid network, so
471 # we'll check this origin against hostnames.
472 if (
473 origin_host
474 and origin_host.lower() != secure_host.lower()
475 and secure_host != "*"
476 ):
477 continue
478 else:
479 # We have a valid address and network, so see if the address
480 # is contained within the network.
481 if addr not in network:
482 continue
484 # Check to see if the port matches.
485 if (
486 origin_port != secure_port
487 and secure_port != "*"
488 and secure_port is not None
489 ):
490 continue
492 # If we've gotten here, then this origin matches the current
493 # secure origin and we should return True
494 return True
496 # If we've gotten to this point, then the origin isn't secure and we
497 # will not accept it as a valid location to search. We will however
498 # log a warning that we are ignoring it.
499 logger.warning(
500 "The repository located at %s is not a trusted or secure host and "
501 "is being ignored. If this repository is available via HTTPS we "
502 "recommend you use HTTPS instead, otherwise you may silence "
503 "this warning and allow it anyway with '--trusted-host %s'.",
504 origin_host,
505 origin_host,
506 )
508 return False
510 def request(self, method: str, url: str, *args: Any, **kwargs: Any) -> Response:
511 # Allow setting a default timeout on a session
512 kwargs.setdefault("timeout", self.timeout)
513 # Allow setting a default proxies on a session
514 kwargs.setdefault("proxies", self.proxies)
516 # Dispatch the actual request
517 return super().request(method, url, *args, **kwargs)