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

1import os 

2import socket 

3import struct 

4from typing import Optional 

5from urllib.parse import unquote, urlparse 

6from ._exceptions import WebSocketProxyException 

7 

8""" 

9_url.py 

10websocket - WebSocket client library for Python 

11 

12Copyright 2024 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 += f"?{parsed.query}" 

72 

73 return hostname, port, resource, is_secure 

74 

75 

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

77 try: 

78 socket.inet_aton(addr) 

79 except socket.error: 

80 return False 

81 else: 

82 return True 

83 

84 

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

86 try: 

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

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

89 except ValueError: 

90 return False 

91 

92 

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

94 ipaddr: int = struct.unpack("!I", socket.inet_aton(ip))[0] 

95 netaddr, netmask = net.split("/") 

96 netaddr: int = struct.unpack("!I", socket.inet_aton(netaddr))[0] 

97 

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

99 return ipaddr & netmask == netaddr 

100 

101 

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

103 if not no_proxy: 

104 if v := os.environ.get("no_proxy", os.environ.get("NO_PROXY", "")).replace( 

105 " ", "" 

106 ): 

107 no_proxy = v.split(",") 

108 

109 if not no_proxy: 

110 no_proxy = [] 

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( 

118 [ 

119 _is_address_in_network(hostname, subnet) 

120 for subnet in no_proxy 

121 if _is_subnet_address(subnet) 

122 ] 

123 ) 

124 for domain in [domain for domain in no_proxy if domain.startswith(".")]: 

125 endDomain = domain.lstrip('.') 

126 if hostname.endswith(endDomain): 

127 return True 

128 return False 

129 

130 

131def get_proxy_info( 

132 hostname: str, 

133 is_secure: bool, 

134 proxy_host: Optional[str] = None, 

135 proxy_port: int = 0, 

136 proxy_auth: Optional[tuple] = None, 

137 no_proxy: Optional[list] = None, 

138 proxy_type: str = "http", 

139) -> tuple: 

140 """ 

141 Try to retrieve proxy host and port from environment 

142 if not provided in options. 

143 Result is (proxy_host, proxy_port, proxy_auth). 

144 proxy_auth is tuple of username and password 

145 of proxy authentication information. 

146 

147 Parameters 

148 ---------- 

149 hostname: str 

150 Websocket server name. 

151 is_secure: bool 

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

153 instead of "http_proxy" 

154 proxy_host: str 

155 http proxy host name. 

156 proxy_port: str or int 

157 http proxy port. 

158 no_proxy: list 

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

160 proxy_auth: tuple 

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

162 proxy_type: str 

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

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

165 """ 

166 if _is_no_proxy_host(hostname, no_proxy): 

167 return None, 0, None 

168 

169 if proxy_host: 

170 if not proxy_port: 

171 raise WebSocketProxyException("Cannot use port 0 when proxy_host specified") 

172 port = proxy_port 

173 auth = proxy_auth 

174 return proxy_host, port, auth 

175 

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

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

178 " ", "" 

179 ) 

180 if value: 

181 proxy = urlparse(value) 

182 auth = ( 

183 (unquote(proxy.username), unquote(proxy.password)) 

184 if proxy.username 

185 else None 

186 ) 

187 return proxy.hostname, proxy.port, auth 

188 

189 return None, 0, None