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

1from __future__ import absolute_import 

2 

3import socket 

4 

5from ..contrib import _appengine_environ 

6from ..exceptions import LocationParseError 

7from ..packages import six 

8from .wait import NoWayToWaitForSocketError, wait_for_read 

9 

10 

11def 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. 

37def 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 

100def _set_socket_options(sock, options): 

101 if options is None: 

102 return 

103 

104 for opt in options: 

105 sock.setsockopt(*opt) 

106 

107 

108def 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 

119def _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 

149HAS_IPV6 = _has_ipv6("::1")