Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/util/connection.py: 36%

1from __future__ import annotations 

2 

3import socket 

4import typing 

5 

6from ..exceptions import LocationParseError 

7from .timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT 

8 

9_TYPE_SOCKET_OPTIONS = list[tuple[int, int, typing.Union[int, bytes]]] 

10 

11if typing.TYPE_CHECKING: 

12 from .._base_connection import BaseHTTPConnection 

13 

14 

15def is_connection_dropped(conn: BaseHTTPConnection) -> bool: # Platform-specific 

16 """ 

17 Returns True if the connection is dropped and should be closed. 

18 :param conn: :class:`urllib3.connection.HTTPConnection` object. 

19 """ 

20 return not conn.is_connected 

21 

22 

23# This function is copied from socket.py in the Python 2.7 standard 

24# library test suite. Added to its signature is only `socket_options`. 

25# One additional modification is that we avoid binding to IPv6 servers 

26# discovered in DNS if the system doesn't have IPv6 functionality. 

27def create_connection( 

28 address: tuple[str, int], 

29 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

30 source_address: tuple[str, int] | None = None, 

31 socket_options: _TYPE_SOCKET_OPTIONS | None = None, 

32) -> socket.socket: 

33 """Connect to *address* and return the socket object. 

34 

35 Convenience function. Connect to *address* (a 2-tuple ``(host, 

36 port)``) and return the socket object. Passing the optional 

37 *timeout* parameter will set the timeout on the socket instance 

38 before attempting to connect. If no *timeout* is supplied, the 

39 global default timeout setting returned by :func:`socket.getdefaulttimeout` 

40 is used. If *source_address* is set it must be a tuple of (host, port) 

41 for the socket to bind as a source address before making the connection. 

42 An host of '' or port 0 tells the OS to use the default. 

43 """ 

44 

45 host, port = address 

46 if host.startswith("["): 

47 host = host.strip("[]") 

48 err = None 

49 

50 # Using the value from allowed_gai_family() in the context of getaddrinfo lets 

51 # us select whether to work with IPv4 DNS records, IPv6 records, or both. 

52 # The original create_connection function always returns all records. 

53 family = allowed_gai_family() 

54 

55 try: 

56 host.encode("idna") 

57 except UnicodeError: 

58 raise LocationParseError(f"'{host}', label empty or too long") from None 

59 

60 for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM): 

61 af, socktype, proto, canonname, sa = res 

62 sock = None 

63 try: 

64 sock = socket.socket(af, socktype, proto) 

65 

66 # If provided, set socket level options before connecting. 

67 _set_socket_options(sock, socket_options) 

68 

69 if timeout is not _DEFAULT_TIMEOUT: 

70 sock.settimeout(timeout) 

71 if source_address: 

72 sock.bind(source_address) 

73 sock.connect(sa) 

74 # Break explicitly a reference cycle 

75 err = None 

76 return sock 

77 

78 except OSError as _: 

79 err = _ 

80 if sock is not None: 

81 sock.close() 

82 

83 if err is not None: 

84 try: 

85 raise err 

86 finally: 

87 # Break explicitly a reference cycle 

88 err = None 

89 else: 

90 raise OSError("getaddrinfo returns an empty list") 

91 

92 

93def _set_socket_options( 

94 sock: socket.socket, options: _TYPE_SOCKET_OPTIONS | None 

95) -> None: 

96 if options is None: 

97 return 

98 

99 for opt in options: 

100 sock.setsockopt(*opt) 

101 

102 

103def allowed_gai_family() -> socket.AddressFamily: 

104 """This function is designed to work in the context of 

105 getaddrinfo, where family=socket.AF_UNSPEC is the default and 

106 will perform a DNS search for both IPv6 and IPv4 records.""" 

107 

108 family = socket.AF_INET 

109 if HAS_IPV6: 

110 family = socket.AF_UNSPEC 

111 return family 

112 

113 

114def _has_ipv6(host: str) -> bool: 

115 """Returns True if the system can bind an IPv6 address.""" 

116 sock = None 

117 has_ipv6 = False 

118 

119 if socket.has_ipv6: 

120 # has_ipv6 returns true if cPython was compiled with IPv6 support. 

121 # It does not tell us if the system has IPv6 support enabled. To 

122 # determine that we must bind to an IPv6 address. 

123 # https://github.com/urllib3/urllib3/pull/611 

124 # https://bugs.python.org/issue658327 

125 try: 

126 sock = socket.socket(socket.AF_INET6) 

127 sock.bind((host, 0)) 

128 has_ipv6 = True 

129 except Exception: 

130 pass 

131 

132 if sock: 

133 sock.close() 

134 return has_ipv6 

135 

136 

137HAS_IPV6 = _has_ipv6("::1")