venv/lib/python3.11/site-packages/pip/_vendor/urllib3/util/connection.py

   1 from __future__ import absolute_import
   2
   3 import socket
   4
   5 from ..contrib import _appengine_environ
   6 from ..exceptions import LocationParseError
   7 from ..packages import six
   8 from .wait import NoWayToWaitForSocketError, wait_for_read
   9
  10
  11 def is_connection_dropped(conn):  # Platform-specific
  12     """
  13     Returns True if the connection is dropped and should be closed.
  14
  15     :param conn:
  16         :class:`http.client.HTTPConnection` object.
  17
  18     Note: For platforms like AppEngine, this will always return ``False`` to
  19     let the platform handle connection recycling transparently for us.
  20     """
  21     sock = getattr(conn, "sock", False)
  22     if sock is False:  # Platform-specific: AppEngine
  23         return False
  24     if sock is None:  # Connection already closed (such as by httplib).
  25         return True
  26     try:
  27         # Returns True if readable, which here means it's been dropped
  28         return wait_for_read(sock, timeout=0.0)
  29     except NoWayToWaitForSocketError:  # Platform-specific: AppEngine
  30         return False
  31
  32
  33 # This function is copied from socket.py in the Python 2.7 standard
  34 # library test suite. Added to its signature is only `socket_options`.
  35 # One additional modification is that we avoid binding to IPv6 servers
  36 # discovered in DNS if the system doesn't have IPv6 functionality.
  37 def create_connection(
  38     address,
  39     timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
  40     source_address=None,
  41     socket_options=None,
  42 ):
  43     """Connect to *address* and return the socket object.
  44
  45     Convenience function.  Connect to *address* (a 2-tuple ``(host,
  46     port)``) and return the socket object.  Passing the optional
  47     *timeout* parameter will set the timeout on the socket instance
  48     before attempting to connect.  If no *timeout* is supplied, the
  49     global default timeout setting returned by :func:`socket.getdefaulttimeout`
  50     is used.  If *source_address* is set it must be a tuple of (host, port)
  51     for the socket to bind as a source address before making the connection.
  52     An host of '' or port 0 tells the OS to use the default.
  53     """
  54
  55     host, port = address
  56     if host.startswith("["):
  57         host = host.strip("[]")
  58     err = None
  59
  60     # Using the value from allowed_gai_family() in the context of getaddrinfo lets
  61     # us select whether to work with IPv4 DNS records, IPv6 records, or both.
  62     # The original create_connection function always returns all records.
  63     family = allowed_gai_family()
  64
  65     try:
  66         host.encode("idna")
  67     except UnicodeError:
  68         return six.raise_from(
  69             LocationParseError(u"'%s', label empty or too long" % host), None
  70         )
  71
  72     for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
  73         af, socktype, proto, canonname, sa = res
  74         sock = None
  75         try:
  76             sock = socket.socket(af, socktype, proto)
  77
  78             # If provided, set socket level options before connecting.
  79             _set_socket_options(sock, socket_options)
  80
  81             if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
  82                 sock.settimeout(timeout)
  83             if source_address:
  84                 sock.bind(source_address)
  85             sock.connect(sa)
  86             return sock
  87
  88         except socket.error as e:
  89             err = e
  90             if sock is not None:
  91                 sock.close()
  92                 sock = None
  93
  94     if err is not None:
  95         raise err
  96
  97     raise socket.error("getaddrinfo returns an empty list")
  98
  99
 100 def _set_socket_options(sock, options):
 101     if options is None:
 102         return
 103
 104     for opt in options:
 105         sock.setsockopt(*opt)
 106
 107
 108 def allowed_gai_family():
 109     """This function is designed to work in the context of
 110     getaddrinfo, where family=socket.AF_UNSPEC is the default and
 111     will perform a DNS search for both IPv6 and IPv4 records."""
 112
 113     family = socket.AF_INET
 114     if HAS_IPV6:
 115         family = socket.AF_UNSPEC
 116     return family
 117
 118
 119 def _has_ipv6(host):
 120     """Returns True if the system can bind an IPv6 address."""
 121     sock = None
 122     has_ipv6 = False
 123
 124     # App Engine doesn't support IPV6 sockets and actually has a quota on the
 125     # number of sockets that can be used, so just early out here instead of
 126     # creating a socket needlessly.
 127     # See https://github.com/urllib3/urllib3/issues/1446
 128     if _appengine_environ.is_appengine_sandbox():
 129         return False
 130
 131     if socket.has_ipv6:
 132         # has_ipv6 returns true if cPython was compiled with IPv6 support.
 133         # It does not tell us if the system has IPv6 support enabled. To
 134         # determine that we must bind to an IPv6 address.
 135         # https://github.com/urllib3/urllib3/pull/611
 136         # https://bugs.python.org/issue658327
 137         try:
 138             sock = socket.socket(socket.AF_INET6)
 139             sock.bind((host, 0))
 140             has_ipv6 = True
 141         except Exception:
 142             pass
 143
 144     if sock:
 145         sock.close()
 146     return has_ipv6
 147
 148
 149 HAS_IPV6 = _has_ipv6("::1")