Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/websocket/_url.py: 44%

1import os 

2import socket 

3import struct 

4 

5from urllib.parse import unquote, urlparse 

6 

7""" 

8_url.py 

9websocket - WebSocket client library for Python 

10 

11Copyright 2022 engn33r 

12 

13Licensed under the Apache License, Version 2.0 (the "License"); 

14you may not use this file except in compliance with the License. 

15You may obtain a copy of the License at 

16 

17 http://www.apache.org/licenses/LICENSE-2.0 

18 

19Unless required by applicable law or agreed to in writing, software 

20distributed under the License is distributed on an "AS IS" BASIS, 

21WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

22See the License for the specific language governing permissions and 

23limitations under the License. 

24""" 

25 

26__all__ = ["parse_url", "get_proxy_info"] 

27 

28 

29def parse_url(url: str) -> tuple: 

30 """ 

31 parse url and the result is tuple of 

32 (hostname, port, resource path and the flag of secure mode) 

33 

34 Parameters 

35 ---------- 

36 url: str 

37 url string. 

38 """ 

39 if ":" not in url: 

40 raise ValueError("url is invalid") 

41 

42 scheme, url = url.split(":", 1) 

43 

44 parsed = urlparse(url, scheme="http") 

45 if parsed.hostname: 

46 hostname = parsed.hostname 

47 else: 

48 raise ValueError("hostname is invalid") 

49 port = 0 

50 if parsed.port: 

51 port = parsed.port 

52 

53 is_secure = False 

54 if scheme == "ws": 

55 if not port: 

56 port = 80 

57 elif scheme == "wss": 

58 is_secure = True 

59 if not port: 

60 port = 443 

61 else: 

62 raise ValueError("scheme %s is invalid" % scheme) 

63 

64 if parsed.path: 

65 resource = parsed.path 

66 else: 

67 resource = "/" 

68 

69 if parsed.query: 

70 resource += "?" + parsed.query 

71 

72 return hostname, port, resource, is_secure 

73 

74 

75DEFAULT_NO_PROXY_HOST = ["localhost", "127.0.0.1"] 

76 

77 

78def _is_ip_address(addr: str) -> bool: 

79 try: 

80 socket.inet_aton(addr) 

81 except socket.error: 

82 return False 

83 else: 

84 return True 

85 

86 

87def _is_subnet_address(hostname: str) -> bool: 

88 try: 

89 addr, netmask = hostname.split("/") 

90 return _is_ip_address(addr) and 0 <= int(netmask) < 32 

91 except ValueError: 

92 return False 

93 

94 

95def _is_address_in_network(ip: str, net: str) -> bool: 

96 ipaddr = struct.unpack('!I', socket.inet_aton(ip))[0] 

97 netaddr, netmask = net.split('/') 

98 netaddr = struct.unpack('!I', socket.inet_aton(netaddr))[0] 

99 

100 netmask = (0xFFFFFFFF << (32 - int(netmask))) & 0xFFFFFFFF 

101 return ipaddr & netmask == netaddr 

102 

103 

104def _is_no_proxy_host(hostname: str, no_proxy: list) -> bool: 

105 if not no_proxy: 

106 v = os.environ.get("no_proxy", os.environ.get("NO_PROXY", "")).replace(" ", "") 

107 if v: 

108 no_proxy = v.split(",") 

109 if not no_proxy: 

110 no_proxy = DEFAULT_NO_PROXY_HOST 

111 

112 if '*' in no_proxy: 

113 return True 

114 if hostname in no_proxy: 

115 return True 

116 if _is_ip_address(hostname): 

117 return any([_is_address_in_network(hostname, subnet) for subnet in no_proxy if _is_subnet_address(subnet)]) 

118 for domain in [domain for domain in no_proxy if domain.startswith('.')]: 

119 if hostname.endswith(domain): 

120 return True 

121 return False 

122 

123 

124def get_proxy_info( 

125 hostname: str, is_secure: bool, proxy_host: str = None, proxy_port: int = 0, proxy_auth: tuple = None, 

126 no_proxy: list = None, proxy_type: str = 'http') -> tuple: 

127 """ 

128 Try to retrieve proxy host and port from environment 

129 if not provided in options. 

130 Result is (proxy_host, proxy_port, proxy_auth). 

131 proxy_auth is tuple of username and password 

132 of proxy authentication information. 

133 

134 Parameters 

135 ---------- 

136 hostname: str 

137 Websocket server name. 

138 is_secure: bool 

139 Is the connection secure? (wss) looks for "https_proxy" in env 

140 before falling back to "http_proxy" 

141 proxy_host: str 

142 http proxy host name. 

143 http_proxy_port: str or int 

144 http proxy port. 

145 http_no_proxy: list 

146 Whitelisted host names that don't use the proxy. 

147 http_proxy_auth: tuple 

148 HTTP proxy auth information. Tuple of username and password. Default is None. 

149 proxy_type: str 

150 Specify the proxy protocol (http, socks4, socks4a, socks5, socks5h). Default is "http". 

151 Use socks4a or socks5h if you want to send DNS requests through the proxy. 

152 """ 

153 if _is_no_proxy_host(hostname, no_proxy): 

154 return None, 0, None 

155 

156 if proxy_host: 

157 port = proxy_port 

158 auth = proxy_auth 

159 return proxy_host, port, auth 

160 

161 env_keys = ["http_proxy"] 

162 if is_secure: 

163 env_keys.insert(0, "https_proxy") 

164 

165 for key in env_keys: 

166 value = os.environ.get(key, os.environ.get(key.upper(), "")).replace(" ", "") 

167 if value: 

168 proxy = urlparse(value) 

169 auth = (unquote(proxy.username), unquote(proxy.password)) if proxy.username else None 

170 return proxy.hostname, proxy.port, auth 

171 

172 return None, 0, None