Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/pip/_internal/network»

Browse Source
This commit is contained in:
Polina
2026-07-02 19:04:36 +00:00
parent f3e67898e4
commit 2f756db6d1
3 changed files with 691 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

532
venv/Lib/site-packages/pip/_internal/network/session.py Normal file
View File

@@ -0,0 +1,532 @@
"""PipSession and supporting code, containing all pip-specific
network request configuration and behavior.
"""
from __future__ import annotations
import email.utils
import functools
import io
import ipaddress
import json
import logging
import mimetypes
import os
import platform
import shutil
import subprocess
import sys
import urllib.parse
import warnings
from collections.abc import Generator, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Any,
Optional,
Union,
)
from pip._vendor import requests, urllib3
from pip._vendor.cachecontrol import CacheControlAdapter as _BaseCacheControlAdapter
from pip._vendor.requests.adapters import DEFAULT_POOLBLOCK, BaseAdapter
from pip._vendor.requests.adapters import HTTPAdapter as _BaseHTTPAdapter
from pip._vendor.requests.models import PreparedRequest, Response
from pip._vendor.requests.structures import CaseInsensitiveDict
from pip._vendor.urllib3.connectionpool import ConnectionPool
from pip._vendor.urllib3.exceptions import InsecureRequestWarning
from pip import __version__
from pip._internal.metadata import get_default_environment
from pip._internal.models.link import Link
from pip._internal.network.auth import MultiDomainBasicAuth
from pip._internal.network.cache import SafeFileCache
# Import ssl from compat so the initial import occurs in only one place.
from pip._internal.utils.compat import has_tls
from pip._internal.utils.glibc import libc_ver
from pip._internal.utils.misc import build_url_from_netloc, parse_netloc
from pip._internal.utils.urls import url_to_path
if TYPE_CHECKING:
from ssl import SSLContext
from pip._vendor.urllib3 import ProxyManager
from pip._vendor.urllib3.poolmanager import PoolManager
logger = logging.getLogger(__name__)
SecureOrigin = tuple[str, str, Optional[Union[int, str]]]
# Ignore warning raised when using --trusted-host.
warnings.filterwarnings("ignore", category=InsecureRequestWarning)
SECURE_ORIGINS: list[SecureOrigin] = [
# protocol, hostname, port
# Taken from Chrome's list of secure origins (See: http://bit.ly/1qrySKC)
("https", "*", "*"),
("*", "localhost", "*"),
("*", "127.0.0.0/8", "*"),
("*", "::1/128", "*"),
("file", "*", None),
# ssh is always secure.
("ssh", "*", "*"),
]
# These are environment variables present when running under various
# CI systems. For each variable, some CI systems that use the variable
# are indicated. The collection was chosen so that for each of a number
# of popular systems, at least one of the environment variables is used.
# This list is used to provide some indication of and lower bound for
# CI traffic to PyPI. Thus, it is okay if the list is not comprehensive.
# For more background, see: https://github.com/pypa/pip/issues/5499
CI_ENVIRONMENT_VARIABLES = (
# Azure Pipelines
"BUILD_BUILDID",
# Jenkins
"BUILD_ID",
# AppVeyor, CircleCI, Codeship, Gitlab CI, Shippable, Travis CI
"CI",
# Explicit environment variable.
"PIP_IS_CI",
)
def looks_like_ci() -> bool:
"""
Return whether it looks like pip is running under CI.
"""
# We don't use the method of checking for a tty (e.g. using isatty())
# because some CI systems mimic a tty (e.g. Travis CI). Thus that
# method doesn't provide definitive information in either direction.
return any(name in os.environ for name in CI_ENVIRONMENT_VARIABLES)
@functools.lru_cache(maxsize=1)
def user_agent() -> str:
"""
Return a string representing the user agent.
"""
data: dict[str, Any] = {
"installer": {"name": "pip", "version": __version__},
"python": platform.python_version(),
"implementation": {
"name": platform.python_implementation(),
},
}
if data["implementation"]["name"] == "CPython":
data["implementation"]["version"] = platform.python_version()
elif data["implementation"]["name"] == "PyPy":
pypy_version_info = sys.pypy_version_info # type: ignore
if pypy_version_info.releaselevel == "final":
pypy_version_info = pypy_version_info[:3]
data["implementation"]["version"] = ".".join(
[str(x) for x in pypy_version_info]
)
elif data["implementation"]["name"] == "Jython":
# Complete Guess
data["implementation"]["version"] = platform.python_version()
elif data["implementation"]["name"] == "IronPython":
# Complete Guess
data["implementation"]["version"] = platform.python_version()
if sys.platform.startswith("linux"):
from pip._vendor import distro
linux_distribution = distro.name(), distro.version(), distro.codename()
distro_infos: dict[str, Any] = dict(
filter(
lambda x: x[1],
zip(["name", "version", "id"], linux_distribution),
)
)
libc = dict(
filter(
lambda x: x[1],
zip(["lib", "version"], libc_ver()),
)
)
if libc:
distro_infos["libc"] = libc
if distro_infos:
data["distro"] = distro_infos
if sys.platform.startswith("darwin") and platform.mac_ver()[0]:
data["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]}
if platform.system():
data.setdefault("system", {})["name"] = platform.system()
if platform.release():
data.setdefault("system", {})["release"] = platform.release()
if platform.machine():
data["cpu"] = platform.machine()
if has_tls():
import _ssl as ssl
data["openssl_version"] = ssl.OPENSSL_VERSION
setuptools_dist = get_default_environment().get_distribution("setuptools")
if setuptools_dist is not None:
data["setuptools_version"] = str(setuptools_dist.version)
if shutil.which("rustc") is not None:
# If for any reason `rustc --version` fails, silently ignore it
try:
rustc_output = subprocess.check_output(
["rustc", "--version"], stderr=subprocess.STDOUT, timeout=0.5
)
except Exception:
pass
else:
if rustc_output.startswith(b"rustc "):
# The format of `rustc --version` is:
# `b'rustc 1.52.1 (9bc8c42bb 2021-05-09)\n'`
# We extract just the middle (1.52.1) part
data["rustc_version"] = rustc_output.split(b" ")[1].decode()
# Use None rather than False so as not to give the impression that
# pip knows it is not being run under CI. Rather, it is a null or
# inconclusive result. Also, we include some value rather than no
# value to make it easier to know that the check has been run.
data["ci"] = True if looks_like_ci() else None
user_data = os.environ.get("PIP_USER_AGENT_USER_DATA")
if user_data is not None:
data["user_data"] = user_data
return "{data[installer][name]}/{data[installer][version]} {json}".format(
data=data,
json=json.dumps(data, separators=(",", ":"), sort_keys=True),
)
class LocalFSAdapter(BaseAdapter):
def send(
self,
request: PreparedRequest,
stream: bool = False,
timeout: float | tuple[float, float] | tuple[float, None] | None = None,
verify: bool | str = True,
cert: bytes | str | tuple[bytes | str, bytes | str] | None = None,
proxies: Mapping[str, str] | None = None,
) -> Response:
assert request.url is not None
pathname = url_to_path(request.url)
resp = Response()
resp.status_code = 200
resp.url = request.url
try:
stats = os.stat(pathname)
except OSError as exc:
# format the exception raised as a io.BytesIO object,
# to return a better error message:
resp.status_code = 404
resp.reason = type(exc).__name__
resp.raw = io.BytesIO(f"{resp.reason}: {exc}".encode())
else:
modified = email.utils.formatdate(stats.st_mtime, usegmt=True)
content_type = mimetypes.guess_type(pathname)[0] or "text/plain"
resp.headers = CaseInsensitiveDict(
{
"Content-Type": content_type,
"Content-Length": str(stats.st_size),
"Last-Modified": modified,
}
)
resp.raw = open(pathname, "rb")
resp.close = resp.raw.close # type: ignore[method-assign]
return resp
def close(self) -> None:
pass
class _SSLContextAdapterMixin:
"""Mixin to add the ``ssl_context`` constructor argument to HTTP adapters.
The additional argument is forwarded directly to the pool manager. This allows us
to dynamically decide what SSL store to use at runtime, which is used to implement
the optional ``truststore`` backend.
"""
def __init__(
self,
*,
ssl_context: SSLContext | None = None,
**kwargs: Any,
) -> None:
self._ssl_context = ssl_context
super().__init__(**kwargs)
def init_poolmanager(
self,
connections: int,
maxsize: int,
block: bool = DEFAULT_POOLBLOCK,
**pool_kwargs: Any,
) -> PoolManager:
if self._ssl_context is not None:
pool_kwargs.setdefault("ssl_context", self._ssl_context)
return super().init_poolmanager( # type: ignore[misc, no-any-return]
connections=connections,
maxsize=maxsize,
block=block,
**pool_kwargs,
)
def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> ProxyManager:
# Proxy manager replaces the pool manager, so inject our SSL
# context here too. https://github.com/pypa/pip/issues/13288
if self._ssl_context is not None:
proxy_kwargs.setdefault("ssl_context", self._ssl_context)
return super().proxy_manager_for(proxy, **proxy_kwargs) # type: ignore[misc, no-any-return]
class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter):
pass
class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter):
pass
class InsecureHTTPAdapter(HTTPAdapter):
def cert_verify(
self,
conn: ConnectionPool,
url: str,
verify: bool | str,
cert: str | tuple[str, str] | None,
) -> None:
super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
class InsecureCacheControlAdapter(CacheControlAdapter):
def cert_verify(
self,
conn: ConnectionPool,
url: str,
verify: bool | str,
cert: str | tuple[str, str] | None,
) -> None:
super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
class PipSession(requests.Session):
timeout: int | None = None
def __init__(
self,
*args: Any,
retries: int = 0,
resume_retries: int = 0,
cache: str | None = None,
trusted_hosts: Sequence[str] = (),
index_urls: list[str] | None = None,
ssl_context: SSLContext | None = None,
**kwargs: Any,
) -> None:
"""
:param trusted_hosts: Domains not to emit warnings for when not using
HTTPS.
"""
super().__init__(*args, **kwargs)
# Namespace the attribute with "pip_" just in case to prevent
# possible conflicts with the base class.
self.pip_trusted_origins: list[tuple[str, int | None]] = []
self.pip_proxy = None
# Attach our User Agent to the request
self.headers["User-Agent"] = user_agent()
# Attach our Authentication handler to the session
self.auth: MultiDomainBasicAuth = MultiDomainBasicAuth(index_urls=index_urls)
# Create our urllib3.Retry instance which will allow us to customize
# how we handle retries.
retries = urllib3.Retry(
# Set the total number of retries that a particular request can
# have.
total=retries,
# A 503 error from PyPI typically means that the Fastly -> Origin
# connection got interrupted in some way. A 503 error in general
# is typically considered a transient error so we'll go ahead and
# retry it.
# A 500 may indicate transient error in Amazon S3
# A 502 may be a transient error from a CDN like CloudFlare or CloudFront
# A 520 or 527 - may indicate transient error in CloudFlare
status_forcelist=[500, 502, 503, 520, 527],
# Add a small amount of back off between failed requests in
# order to prevent hammering the service.
backoff_factor=0.25,
) # type: ignore
self.resume_retries = resume_retries
# Our Insecure HTTPAdapter disables HTTPS validation. It does not
# support caching so we'll use it for all http:// URLs.
# If caching is disabled, we will also use it for
# https:// hosts that we've marked as ignoring
# TLS errors for (trusted-hosts).
insecure_adapter = InsecureHTTPAdapter(max_retries=retries)
# We want to _only_ cache responses on securely fetched origins or when
# the host is specified as trusted. We do this because
# we can't validate the response of an insecurely/untrusted fetched
# origin, and we don't want someone to be able to poison the cache and
# require manual eviction from the cache to fix it.
self._trusted_host_adapter: InsecureCacheControlAdapter | InsecureHTTPAdapter
if cache:
secure_adapter: _BaseHTTPAdapter = CacheControlAdapter(
cache=SafeFileCache(cache),
max_retries=retries,
ssl_context=ssl_context,
)
self._trusted_host_adapter = InsecureCacheControlAdapter(
cache=SafeFileCache(cache),
max_retries=retries,
)
else:
secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context)
self._trusted_host_adapter = insecure_adapter
self.mount("https://", secure_adapter)
self.mount("http://", insecure_adapter)
# Enable file:// urls
self.mount("file://", LocalFSAdapter())
for host in trusted_hosts:
self.add_trusted_host(host, suppress_logging=True)
def update_index_urls(self, new_index_urls: list[str]) -> None:
"""
:param new_index_urls: New index urls to update the authentication
handler with.
"""
self.auth.index_urls = new_index_urls
def add_trusted_host(
self, host: str, source: str | None = None, suppress_logging: bool = False
) -> None:
"""
:param host: It is okay to provide a host that has previously been
added.
:param source: An optional source string, for logging where the host
string came from.
"""
if not suppress_logging:
msg = f"adding trusted host: {host!r}"
if source is not None:
msg += f" (from {source})"
logger.info(msg)
parsed_host, parsed_port = parse_netloc(host)
if parsed_host is None:
raise ValueError(f"Trusted host URL must include a host part: {host!r}")
if (parsed_host, parsed_port) not in self.pip_trusted_origins:
self.pip_trusted_origins.append((parsed_host, parsed_port))
self.mount(
build_url_from_netloc(host, scheme="http") + "/", self._trusted_host_adapter
)
self.mount(build_url_from_netloc(host) + "/", self._trusted_host_adapter)
if not parsed_port:
self.mount(
build_url_from_netloc(host, scheme="http") + ":",
self._trusted_host_adapter,
)
# Mount wildcard ports for the same host.
self.mount(build_url_from_netloc(host) + ":", self._trusted_host_adapter)
def iter_secure_origins(self) -> Generator[SecureOrigin, None, None]:
yield from SECURE_ORIGINS
for host, port in self.pip_trusted_origins:
yield ("*", host, "*" if port is None else port)
def is_secure_origin(self, location: Link) -> bool:
# Determine if this url used a secure transport mechanism
parsed = urllib.parse.urlparse(str(location))
origin_protocol, origin_host, origin_port = (
parsed.scheme,
parsed.hostname,
parsed.port,
)
# The protocol to use to see if the protocol matches.
# Don't count the repository type as part of the protocol: in
# cases such as "git+ssh", only use "ssh". (I.e., Only verify against
# the last scheme.)
origin_protocol = origin_protocol.rsplit("+", 1)[-1]
# Determine if our origin is a secure origin by looking through our
# hardcoded list of secure origins, as well as any additional ones
# configured on this PackageFinder instance.
for secure_origin in self.iter_secure_origins():
secure_protocol, secure_host, secure_port = secure_origin
if origin_protocol != secure_protocol and secure_protocol != "*":
continue
try:
addr = ipaddress.ip_address(origin_host or "")
network = ipaddress.ip_network(secure_host)
except ValueError:
# We don't have both a valid address or a valid network, so
# we'll check this origin against hostnames.
if (
origin_host
and origin_host.lower() != secure_host.lower()
and secure_host != "*"
):
continue
else:
# We have a valid address and network, so see if the address
# is contained within the network.
if addr not in network:
continue
# Check to see if the port matches.
if (
origin_port != secure_port
and secure_port != "*"
and secure_port is not None
):
continue
# If we've gotten here, then this origin matches the current
# secure origin and we should return True
return True
# If we've gotten to this point, then the origin isn't secure and we
# will not accept it as a valid location to search. We will however
# log a warning that we are ignoring it.
logger.warning(
"The repository located at %s is not a trusted or secure host and "
"is being ignored. If this repository is available via HTTPS we "
"recommend you use HTTPS instead, otherwise you may silence "
"this warning and allow it anyway with '--trusted-host %s'.",
origin_host,
origin_host,
)
return False
def request(self, method: str, url: str, *args: Any, **kwargs: Any) -> Response: # type: ignore[override]
# Allow setting a default timeout on a session
kwargs.setdefault("timeout", self.timeout)
# Allow setting a default proxies on a session
kwargs.setdefault("proxies", self.proxies)
# Dispatch the actual request
return super().request(method, url, *args, **kwargs)

98
venv/Lib/site-packages/pip/_internal/network/utils.py Normal file
View File

@@ -0,0 +1,98 @@
from collections.abc import Generator
from pip._vendor.requests.models import Response
from pip._internal.exceptions import NetworkConnectionError
# The following comments and HTTP headers were originally added by
# Donald Stufft in git commit 22c562429a61bb77172039e480873fb239dd8c03.
#
# We use Accept-Encoding: identity here because requests defaults to
# accepting compressed responses. This breaks in a variety of ways
# depending on how the server is configured.
# - Some servers will notice that the file isn't a compressible file
# and will leave the file alone and with an empty Content-Encoding
# - Some servers will notice that the file is already compressed and
# will leave the file alone, adding a Content-Encoding: gzip header
# - Some servers won't notice anything at all and will take a file
# that's already been compressed and compress it again, and set
# the Content-Encoding: gzip header
# By setting this to request only the identity encoding we're hoping
# to eliminate the third case. Hopefully there does not exist a server
# which when given a file will notice it is already compressed and that
# you're not asking for a compressed file and will then decompress it
# before sending because if that's the case I don't think it'll ever be
# possible to make this work.
HEADERS: dict[str, str] = {"Accept-Encoding": "identity"}
DOWNLOAD_CHUNK_SIZE = 256 * 1024
def raise_for_status(resp: Response) -> None:
http_error_msg = ""
if isinstance(resp.reason, bytes):
# We attempt to decode utf-8 first because some servers
# choose to localize their reason strings. If the string
# isn't utf-8, we fall back to iso-8859-1 for all other
# encodings.
try:
reason = resp.reason.decode("utf-8")
except UnicodeDecodeError:
reason = resp.reason.decode("iso-8859-1")
else:
reason = resp.reason
if 400 <= resp.status_code < 500:
http_error_msg = (
f"{resp.status_code} Client Error: {reason} for url: {resp.url}"
)
elif 500 <= resp.status_code < 600:
http_error_msg = (
f"{resp.status_code} Server Error: {reason} for url: {resp.url}"
)
if http_error_msg:
raise NetworkConnectionError(http_error_msg, response=resp)
def response_chunks(
response: Response, chunk_size: int = DOWNLOAD_CHUNK_SIZE
) -> Generator[bytes, None, None]:
"""Given a requests Response, provide the data chunks."""
try:
# Special case for urllib3.
for chunk in response.raw.stream(
chunk_size,
# We use decode_content=False here because we don't
# want urllib3 to mess with the raw bytes we get
# from the server. If we decompress inside of
# urllib3 then we cannot verify the checksum
# because the checksum will be of the compressed
# file. This breakage will only occur if the
# server adds a Content-Encoding header, which
# depends on how the server was configured:
# - Some servers will notice that the file isn't a
# compressible file and will leave the file alone
# and with an empty Content-Encoding
# - Some servers will notice that the file is
# already compressed and will leave the file
# alone and will add a Content-Encoding: gzip
# header
# - Some servers won't notice anything at all and
# will take a file that's already been compressed
# and compress it again and set the
# Content-Encoding: gzip header
#
# By setting this not to decode automatically we
# hope to eliminate problems with the second case.
decode_content=False,
):
yield chunk
except AttributeError:
# Standard file-like object.
while True:
chunk = response.raw.read(chunk_size)
if not chunk:
break
yield chunk

61
venv/Lib/site-packages/pip/_internal/network/xmlrpc.py Normal file
View File

@@ -0,0 +1,61 @@
"""xmlrpclib.Transport implementation"""
import logging
import urllib.parse
import xmlrpc.client
from typing import TYPE_CHECKING
from pip._internal.exceptions import NetworkConnectionError
from pip._internal.network.session import PipSession
from pip._internal.network.utils import raise_for_status
if TYPE_CHECKING:
from xmlrpc.client import _HostType, _Marshallable
from _typeshed import SizedBuffer
logger = logging.getLogger(__name__)
class PipXmlrpcTransport(xmlrpc.client.Transport):
"""Provide a `xmlrpclib.Transport` implementation via a `PipSession`
object.
"""
def __init__(
self, index_url: str, session: PipSession, use_datetime: bool = False
) -> None:
super().__init__(use_datetime)
index_parts = urllib.parse.urlparse(index_url)
self._scheme = index_parts.scheme
self._session = session
def request(
self,
host: "_HostType",
handler: str,
request_body: "SizedBuffer",
verbose: bool = False,
) -> tuple["_Marshallable", ...]:
assert isinstance(host, str)
parts = (self._scheme, host, handler, None, None, None)
url = urllib.parse.urlunparse(parts)
try:
headers = {"Content-Type": "text/xml"}
response = self._session.post(
url,
data=request_body,
headers=headers,
stream=True,
)
raise_for_status(response)
self.verbose = verbose
return self.parse_response(response.raw)
except NetworkConnectionError as exc:
assert exc.response
logger.critical(
"HTTP error %s while getting %s",
exc.response.status_code,
url,
)
raise