jpayne@7: from __future__ import annotations
jpayne@7: 
jpayne@7: import socket
jpayne@7: import typing
jpayne@7: 
jpayne@7: from ..exceptions import LocationParseError
jpayne@7: from .timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
jpayne@7: 
jpayne@7: _TYPE_SOCKET_OPTIONS = typing.Sequence[typing.Tuple[int, int, typing.Union[int, bytes]]]
jpayne@7: 
jpayne@7: if typing.TYPE_CHECKING:
jpayne@7:     from .._base_connection import BaseHTTPConnection
jpayne@7: 
jpayne@7: 
jpayne@7: def is_connection_dropped(conn: BaseHTTPConnection) -> bool:  # Platform-specific
jpayne@7:     """
jpayne@7:     Returns True if the connection is dropped and should be closed.
jpayne@7:     :param conn: :class:`urllib3.connection.HTTPConnection` object.
jpayne@7:     """
jpayne@7:     return not conn.is_connected
jpayne@7: 
jpayne@7: 
jpayne@7: # This function is copied from socket.py in the Python 2.7 standard
jpayne@7: # library test suite. Added to its signature is only `socket_options`.
jpayne@7: # One additional modification is that we avoid binding to IPv6 servers
jpayne@7: # discovered in DNS if the system doesn't have IPv6 functionality.
jpayne@7: def create_connection(
jpayne@7:     address: tuple[str, int],
jpayne@7:     timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
jpayne@7:     source_address: tuple[str, int] | None = None,
jpayne@7:     socket_options: _TYPE_SOCKET_OPTIONS | None = None,
jpayne@7: ) -> socket.socket:
jpayne@7:     """Connect to *address* and return the socket object.
jpayne@7: 
jpayne@7:     Convenience function.  Connect to *address* (a 2-tuple ``(host,
jpayne@7:     port)``) and return the socket object.  Passing the optional
jpayne@7:     *timeout* parameter will set the timeout on the socket instance
jpayne@7:     before attempting to connect.  If no *timeout* is supplied, the
jpayne@7:     global default timeout setting returned by :func:`socket.getdefaulttimeout`
jpayne@7:     is used.  If *source_address* is set it must be a tuple of (host, port)
jpayne@7:     for the socket to bind as a source address before making the connection.
jpayne@7:     An host of '' or port 0 tells the OS to use the default.
jpayne@7:     """
jpayne@7: 
jpayne@7:     host, port = address
jpayne@7:     if host.startswith("["):
jpayne@7:         host = host.strip("[]")
jpayne@7:     err = None
jpayne@7: 
jpayne@7:     # Using the value from allowed_gai_family() in the context of getaddrinfo lets
jpayne@7:     # us select whether to work with IPv4 DNS records, IPv6 records, or both.
jpayne@7:     # The original create_connection function always returns all records.
jpayne@7:     family = allowed_gai_family()
jpayne@7: 
jpayne@7:     try:
jpayne@7:         host.encode("idna")
jpayne@7:     except UnicodeError:
jpayne@7:         raise LocationParseError(f"'{host}', label empty or too long") from None
jpayne@7: 
jpayne@7:     for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
jpayne@7:         af, socktype, proto, canonname, sa = res
jpayne@7:         sock = None
jpayne@7:         try:
jpayne@7:             sock = socket.socket(af, socktype, proto)
jpayne@7: 
jpayne@7:             # If provided, set socket level options before connecting.
jpayne@7:             _set_socket_options(sock, socket_options)
jpayne@7: 
jpayne@7:             if timeout is not _DEFAULT_TIMEOUT:
jpayne@7:                 sock.settimeout(timeout)
jpayne@7:             if source_address:
jpayne@7:                 sock.bind(source_address)
jpayne@7:             sock.connect(sa)
jpayne@7:             # Break explicitly a reference cycle
jpayne@7:             err = None
jpayne@7:             return sock
jpayne@7: 
jpayne@7:         except OSError as _:
jpayne@7:             err = _
jpayne@7:             if sock is not None:
jpayne@7:                 sock.close()
jpayne@7: 
jpayne@7:     if err is not None:
jpayne@7:         try:
jpayne@7:             raise err
jpayne@7:         finally:
jpayne@7:             # Break explicitly a reference cycle
jpayne@7:             err = None
jpayne@7:     else:
jpayne@7:         raise OSError("getaddrinfo returns an empty list")
jpayne@7: 
jpayne@7: 
jpayne@7: def _set_socket_options(
jpayne@7:     sock: socket.socket, options: _TYPE_SOCKET_OPTIONS | None
jpayne@7: ) -> None:
jpayne@7:     if options is None:
jpayne@7:         return
jpayne@7: 
jpayne@7:     for opt in options:
jpayne@7:         sock.setsockopt(*opt)
jpayne@7: 
jpayne@7: 
jpayne@7: def allowed_gai_family() -> socket.AddressFamily:
jpayne@7:     """This function is designed to work in the context of
jpayne@7:     getaddrinfo, where family=socket.AF_UNSPEC is the default and
jpayne@7:     will perform a DNS search for both IPv6 and IPv4 records."""
jpayne@7: 
jpayne@7:     family = socket.AF_INET
jpayne@7:     if HAS_IPV6:
jpayne@7:         family = socket.AF_UNSPEC
jpayne@7:     return family
jpayne@7: 
jpayne@7: 
jpayne@7: def _has_ipv6(host: str) -> bool:
jpayne@7:     """Returns True if the system can bind an IPv6 address."""
jpayne@7:     sock = None
jpayne@7:     has_ipv6 = False
jpayne@7: 
jpayne@7:     if socket.has_ipv6:
jpayne@7:         # has_ipv6 returns true if cPython was compiled with IPv6 support.
jpayne@7:         # It does not tell us if the system has IPv6 support enabled. To
jpayne@7:         # determine that we must bind to an IPv6 address.
jpayne@7:         # https://github.com/urllib3/urllib3/pull/611
jpayne@7:         # https://bugs.python.org/issue658327
jpayne@7:         try:
jpayne@7:             sock = socket.socket(socket.AF_INET6)
jpayne@7:             sock.bind((host, 0))
jpayne@7:             has_ipv6 = True
jpayne@7:         except Exception:
jpayne@7:             pass
jpayne@7: 
jpayne@7:     if sock:
jpayne@7:         sock.close()
jpayne@7:     return has_ipv6
jpayne@7: 
jpayne@7: 
jpayne@7: HAS_IPV6 = _has_ipv6("::1")