index_command.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. """
  2. Contains command classes which may interact with an index / the network.
  3. Unlike its sister module, req_command, this module still uses lazy imports
  4. so commands which don't always hit the network (e.g. list w/o --outdated or
  5. --uptodate) don't need waste time importing PipSession and friends.
  6. """
  7. from __future__ import annotations
  8. import logging
  9. import os
  10. import sys
  11. from functools import lru_cache
  12. from optparse import Values
  13. from typing import TYPE_CHECKING
  14. from pip._vendor import certifi
  15. from pip._internal.cli.base_command import Command
  16. from pip._internal.cli.command_context import CommandContextMixIn
  17. if TYPE_CHECKING:
  18. from ssl import SSLContext
  19. from pip._internal.network.session import PipSession
  20. logger = logging.getLogger(__name__)
  21. @lru_cache
  22. def _create_truststore_ssl_context() -> SSLContext | None:
  23. if sys.version_info < (3, 10):
  24. logger.debug("Disabling truststore because Python version isn't 3.10+")
  25. return None
  26. try:
  27. import ssl
  28. except ImportError:
  29. logger.warning("Disabling truststore since ssl support is missing")
  30. return None
  31. try:
  32. from pip._vendor import truststore
  33. except ImportError:
  34. logger.warning("Disabling truststore because platform isn't supported")
  35. return None
  36. ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
  37. ctx.load_verify_locations(certifi.where())
  38. return ctx
  39. class SessionCommandMixin(CommandContextMixIn):
  40. """
  41. A class mixin for command classes needing _build_session().
  42. """
  43. def __init__(self) -> None:
  44. super().__init__()
  45. self._session: PipSession | None = None
  46. @classmethod
  47. def _get_index_urls(cls, options: Values) -> list[str] | None:
  48. """Return a list of index urls from user-provided options."""
  49. index_urls = []
  50. if not getattr(options, "no_index", False):
  51. url = getattr(options, "index_url", None)
  52. if url:
  53. index_urls.append(url)
  54. urls = getattr(options, "extra_index_urls", None)
  55. if urls:
  56. index_urls.extend(urls)
  57. # Return None rather than an empty list
  58. return index_urls or None
  59. def get_default_session(self, options: Values) -> PipSession:
  60. """Get a default-managed session."""
  61. if self._session is None:
  62. self._session = self.enter_context(self._build_session(options))
  63. # there's no type annotation on requests.Session, so it's
  64. # automatically ContextManager[Any] and self._session becomes Any,
  65. # then https://github.com/python/mypy/issues/7696 kicks in
  66. assert self._session is not None
  67. return self._session
  68. def _build_session(
  69. self,
  70. options: Values,
  71. retries: int | None = None,
  72. timeout: int | None = None,
  73. ) -> PipSession:
  74. from pip._internal.network.session import PipSession
  75. cache_dir = options.cache_dir
  76. assert not cache_dir or os.path.isabs(cache_dir)
  77. if "legacy-certs" not in options.deprecated_features_enabled:
  78. ssl_context = _create_truststore_ssl_context()
  79. else:
  80. ssl_context = None
  81. session = PipSession(
  82. cache=os.path.join(cache_dir, "http-v2") if cache_dir else None,
  83. retries=retries if retries is not None else options.retries,
  84. trusted_hosts=options.trusted_hosts,
  85. index_urls=self._get_index_urls(options),
  86. ssl_context=ssl_context,
  87. )
  88. # Handle custom ca-bundles from the user
  89. if options.cert:
  90. session.verify = options.cert
  91. # Handle SSL client certificate
  92. if options.client_cert:
  93. session.cert = options.client_cert
  94. # Handle timeouts
  95. if options.timeout or timeout:
  96. session.timeout = timeout if timeout is not None else options.timeout
  97. # Handle configured proxies
  98. if options.proxy:
  99. session.proxies = {
  100. "http": options.proxy,
  101. "https": options.proxy,
  102. }
  103. session.trust_env = False
  104. session.pip_proxy = options.proxy
  105. # Determine if we can prompt the user for authentication or not
  106. session.auth.prompting = not options.no_input
  107. session.auth.keyring_provider = options.keyring_provider
  108. return session
  109. def _pip_self_version_check(session: PipSession, options: Values) -> None:
  110. from pip._internal.self_outdated_check import pip_self_version_check as check
  111. check(session, options)
  112. class IndexGroupCommand(Command, SessionCommandMixin):
  113. """
  114. Abstract base class for commands with the index_group options.
  115. This also corresponds to the commands that permit the pip version check.
  116. """
  117. def handle_pip_version_check(self, options: Values) -> None:
  118. """
  119. Do the pip version check if not disabled.
  120. This overrides the default behavior of not doing the check.
  121. """
  122. # Make sure the index_group options are present.
  123. assert hasattr(options, "no_index")
  124. if options.disable_pip_version_check or options.no_index:
  125. return
  126. try:
  127. # Otherwise, check if we're using the latest version of pip available.
  128. session = self._build_session(
  129. options,
  130. retries=0,
  131. timeout=min(5, options.timeout),
  132. )
  133. with session:
  134. _pip_self_version_check(session, options)
  135. except Exception:
  136. logger.warning("There was an error checking the latest version of pip.")
  137. logger.debug("See below for error", exc_info=True)