520 lines
18 KiB
Python
520 lines
18 KiB
Python
|
"""PipSession and supporting code, containing all pip-specific
|
||
|
|
network request configuration and behavior.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import email.utils
|
||
|
|
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 typing import (
|
||
|
|
TYPE_CHECKING,
|
||
|
|
Any,
|
||
|
|
Dict,
|
||
|
|
Generator,
|
||
|
|
List,
|
||
|
|
Mapping,
|
||
|
|
Optional,
|
||
|
|
Sequence,
|
||
|
|
Tuple,
|
||
|
|
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.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)
|
||
|
|
|
||
|
|
|
||
|
|
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: Optional[Union[float, Tuple[float, float]]] = None,
|
||
|
|
verify: Union[bool, str] = True,
|
||
|
|
cert: Optional[Union[str, Tuple[str, str]]] = None,
|
||
|
|
proxies: Optional[Mapping[str, str]] = None,
|
||
|
|
) -> Response:
|
||
|
|
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("utf8"))
|
||
|
|
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": stats.st_size,
|
||
|
|
"Last-Modified": modified,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
resp.raw = open(pathname, "rb")
|
||
|
|
resp.close = resp.raw.close
|
||
|
|
|
||
|
|
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: Optional["SSLContext"] = 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]
|
||
|
|
connections=connections,
|
||
|
|
maxsize=maxsize,
|
||
|
|
block=block,
|
||
|
|
**pool_kwargs,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class InsecureHTTPAdapter(HTTPAdapter):
|
||
|
|
def cert_verify(
|
||
|
|
self,
|
||
|
|
conn: ConnectionPool,
|
||
|
|
url: str,
|
||
|
|
verify: Union[bool, str],
|
||
|
|
cert: Optional[Union[str, Tuple[str, str]]],
|
||
|
|
) -> None:
|
||
|
|
super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
|
||
|
|
|
||
|
|
|
||
|
|
class InsecureCacheControlAdapter(CacheControlAdapter):
|
||
|
|
def cert_verify(
|
||
|
|
self,
|
||
|
|
conn: ConnectionPool,
|
||
|
|
url: str,
|
||
|
|
verify: Union[bool, str],
|
||
|
|
cert: Optional[Union[str, Tuple[str, str]]],
|
||
|
|
) -> None:
|
||
|
|
super().cert_verify(conn=conn, url=url, verify=False, cert=cert)
|
||
|
|
|
||
|
|
|
||
|
|
class PipSession(requests.Session):
|
||
|
|
timeout: Optional[int] = None
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
*args: Any,
|
||
|
|
retries: int = 0,
|
||
|
|
cache: Optional[str] = None,
|
||
|
|
trusted_hosts: Sequence[str] = (),
|
||
|
|
index_urls: Optional[List[str]] = None,
|
||
|
|
ssl_context: Optional["SSLContext"] = 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, Optional[int]]] = []
|
||
|
|
|
||
|
|
# Attach our User Agent to the request
|
||
|
|
self.headers["User-Agent"] = user_agent()
|
||
|
|
|
||
|
|
# Attach our Authentication handler to the session
|
||
|
|
self.auth = 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 520 or 527 - may indicate transient error in CloudFlare
|
||
|
|
status_forcelist=[500, 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
|
||
|
|
|
||
|
|
# 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.
|
||
|
|
if cache:
|
||
|
|
secure_adapter = 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: Optional[str] = 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:
|
||
|
|
# 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)
|