171 lines
5.5 KiB
Python
171 lines
5.5 KiB
Python
"""
|
|
Contains command classes which may interact with an index / the network.
|
|
|
|
Unlike its sister module, req_command, this module still uses lazy imports
|
|
so commands which don't always hit the network (e.g. list w/o --outdated or
|
|
--uptodate) don't need waste time importing PipSession and friends.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
from optparse import Values
|
|
from typing import TYPE_CHECKING, List, Optional
|
|
|
|
from pip._vendor import certifi
|
|
|
|
from pip._internal.cli.base_command import Command
|
|
from pip._internal.cli.command_context import CommandContextMixIn
|
|
|
|
if TYPE_CHECKING:
|
|
from ssl import SSLContext
|
|
|
|
from pip._internal.network.session import PipSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _create_truststore_ssl_context() -> Optional["SSLContext"]:
|
|
if sys.version_info < (3, 10):
|
|
logger.debug("Disabling truststore because Python version isn't 3.10+")
|
|
return None
|
|
|
|
try:
|
|
import ssl
|
|
except ImportError:
|
|
logger.warning("Disabling truststore since ssl support is missing")
|
|
return None
|
|
|
|
try:
|
|
from pip._vendor import truststore
|
|
except ImportError:
|
|
logger.warning("Disabling truststore because platform isn't supported")
|
|
return None
|
|
|
|
ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
ctx.load_verify_locations(certifi.where())
|
|
return ctx
|
|
|
|
|
|
class SessionCommandMixin(CommandContextMixIn):
|
|
"""
|
|
A class mixin for command classes needing _build_session().
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self._session: Optional["PipSession"] = None
|
|
|
|
@classmethod
|
|
def _get_index_urls(cls, options: Values) -> Optional[List[str]]:
|
|
"""Return a list of index urls from user-provided options."""
|
|
index_urls = []
|
|
if not getattr(options, "no_index", False):
|
|
url = getattr(options, "index_url", None)
|
|
if url:
|
|
index_urls.append(url)
|
|
urls = getattr(options, "extra_index_urls", None)
|
|
if urls:
|
|
index_urls.extend(urls)
|
|
# Return None rather than an empty list
|
|
return index_urls or None
|
|
|
|
def get_default_session(self, options: Values) -> "PipSession":
|
|
"""Get a default-managed session."""
|
|
if self._session is None:
|
|
self._session = self.enter_context(self._build_session(options))
|
|
# there's no type annotation on requests.Session, so it's
|
|
# automatically ContextManager[Any] and self._session becomes Any,
|
|
# then https://github.com/python/mypy/issues/7696 kicks in
|
|
assert self._session is not None
|
|
return self._session
|
|
|
|
def _build_session(
|
|
self,
|
|
options: Values,
|
|
retries: Optional[int] = None,
|
|
timeout: Optional[int] = None,
|
|
) -> "PipSession":
|
|
from pip._internal.network.session import PipSession
|
|
|
|
cache_dir = options.cache_dir
|
|
assert not cache_dir or os.path.isabs(cache_dir)
|
|
|
|
if "legacy-certs" not in options.deprecated_features_enabled:
|
|
ssl_context = _create_truststore_ssl_context()
|
|
else:
|
|
ssl_context = None
|
|
|
|
session = PipSession(
|
|
cache=os.path.join(cache_dir, "http-v2") if cache_dir else None,
|
|
retries=retries if retries is not None else options.retries,
|
|
trusted_hosts=options.trusted_hosts,
|
|
index_urls=self._get_index_urls(options),
|
|
ssl_context=ssl_context,
|
|
)
|
|
|
|
# Handle custom ca-bundles from the user
|
|
if options.cert:
|
|
session.verify = options.cert
|
|
|
|
# Handle SSL client certificate
|
|
if options.client_cert:
|
|
session.cert = options.client_cert
|
|
|
|
# Handle timeouts
|
|
if options.timeout or timeout:
|
|
session.timeout = timeout if timeout is not None else options.timeout
|
|
|
|
# Handle configured proxies
|
|
if options.proxy:
|
|
session.proxies = {
|
|
"http": options.proxy,
|
|
"https": options.proxy,
|
|
}
|
|
session.trust_env = False
|
|
|
|
# Determine if we can prompt the user for authentication or not
|
|
session.auth.prompting = not options.no_input
|
|
session.auth.keyring_provider = options.keyring_provider
|
|
|
|
return session
|
|
|
|
|
|
def _pip_self_version_check(session: "PipSession", options: Values) -> None:
|
|
from pip._internal.self_outdated_check import pip_self_version_check as check
|
|
|
|
check(session, options)
|
|
|
|
|
|
class IndexGroupCommand(Command, SessionCommandMixin):
|
|
"""
|
|
Abstract base class for commands with the index_group options.
|
|
|
|
This also corresponds to the commands that permit the pip version check.
|
|
"""
|
|
|
|
def handle_pip_version_check(self, options: Values) -> None:
|
|
"""
|
|
Do the pip version check if not disabled.
|
|
|
|
This overrides the default behavior of not doing the check.
|
|
"""
|
|
# Make sure the index_group options are present.
|
|
assert hasattr(options, "no_index")
|
|
|
|
if options.disable_pip_version_check or options.no_index:
|
|
return
|
|
|
|
try:
|
|
# Otherwise, check if we're using the latest version of pip available.
|
|
session = self._build_session(
|
|
options,
|
|
retries=0,
|
|
timeout=min(5, options.timeout),
|
|
)
|
|
with session:
|
|
_pip_self_version_check(session, options)
|
|
except Exception:
|
|
logger.warning("There was an error checking the latest version of pip.")
|
|
logger.debug("See below for error", exc_info=True)
|