index_command.py 5.5 KB

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