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

1import os 

2import socket 

3import struct 

4 

5from typing import Optional 

6from urllib.parse import unquote, urlparse 

7 

8""" 

9_url.py 

10websocket - WebSocket client library for Python 

11 

12Copyright 2023 engn33r 

13 

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

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

16You may obtain a copy of the License at 

17 

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

19 

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

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

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

23See the License for the specific language governing permissions and 

24limitations under the License. 

25""" 

26 

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

28 

29 

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

31 """ 

32 parse url and the result is tuple of 

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

34 

35 Parameters 

36 ---------- 

37 url: str 

38 url string. 

39 """ 

40 if ":" not in url: 

41 raise ValueError("url is invalid") 

42 

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

44 

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

46 if parsed.hostname: 

47 hostname = parsed.hostname 

48 else: 

49 raise ValueError("hostname is invalid") 

50 port = 0 

51 if parsed.port: 

52 port = parsed.port 

53 

54 is_secure = False 

55 if scheme == "ws": 

56 if not port: 

57 port = 80 

58 elif scheme == "wss": 

59 is_secure = True 

60 if not port: 

61 port = 443 

62 else: 

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

64 

65 if parsed.path: 

66 resource = parsed.path 

67 else: 

68 resource = "/" 

69 

70 if parsed.query: 

71 resource += "?" + parsed.query 

72 

73 return hostname, port, resource, is_secure 

74 

75 

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

77 

78 

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

80 try: 

81 socket.inet_aton(addr) 

82 except socket.error: 

83 return False 

84 else: 

85 return True 

86 

87 

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

89 try: 

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

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

92 except ValueError: 

93 return False 

94 

95 

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

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

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

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

100 

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

102 return ipaddr & netmask == netaddr 

103 

104 

105def _is_no_proxy_host(hostname: str, no_proxy: Optional[list]) -> bool: 

106 if not no_proxy: 

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

108 if v: 

109 no_proxy = v.split(",") 

110 if not no_proxy: 

111 no_proxy = DEFAULT_NO_PROXY_HOST 

112 

113 if '*' in no_proxy: 

114 return True 

115 if hostname in no_proxy: 

116 return True 

117 if _is_ip_address(hostname): 

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

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

120 if hostname.endswith(domain): 

121 return True 

122 return False 

123 

124 

125def get_proxy_info( 

126 hostname: str, is_secure: bool, proxy_host: Optional[str] = None, proxy_port: int = 0, proxy_auth: Optional[tuple] = None, 

127 no_proxy: Optional[list] = None, proxy_type: str = 'http') -> tuple: 

128 """ 

129 Try to retrieve proxy host and port from environment 

130 if not provided in options. 

131 Result is (proxy_host, proxy_port, proxy_auth). 

132 proxy_auth is tuple of username and password 

133 of proxy authentication information. 

134 

135 Parameters 

136 ---------- 

137 hostname: str 

138 Websocket server name. 

139 is_secure: bool 

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

141 instead of "http_proxy" 

142 proxy_host: str 

143 http proxy host name. 

144 proxy_port: str or int 

145 http proxy port. 

146 no_proxy: list 

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

148 proxy_auth: tuple 

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

150 proxy_type: str 

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

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

153 """ 

154 if _is_no_proxy_host(hostname, no_proxy): 

155 return None, 0, None 

156 

157 if proxy_host: 

158 port = proxy_port 

159 auth = proxy_auth 

160 return proxy_host, port, auth 

161 

162 env_key = "https_proxy" if is_secure else "http_proxy" 

163 value = os.environ.get(env_key, os.environ.get(env_key.upper(), "")).replace(" ", "") 

164 if value: 

165 proxy = urlparse(value) 

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

167 return proxy.hostname, proxy.port, auth 

168 

169 return None, 0, None