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

1import ipaddress 

2import os 

3from typing import Optional, Union 

4from urllib.parse import unquote, urlparse 

5from ._exceptions import WebSocketProxyException 

6 

7""" 

8_url.py 

9websocket - WebSocket client library for Python 

10 

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

71 

72 return hostname, port, resource, is_secure 

73 

74 

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

76 if not isinstance(addr, str): 

77 raise TypeError("_is_ip_address() argument 1 must be str") 

78 try: 

79 ipaddress.ip_address(addr) 

80 except ValueError: 

81 return False 

82 else: 

83 return True 

84 

85 

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

87 try: 

88 ipaddress.ip_network(hostname) 

89 except ValueError: 

90 return False 

91 else: 

92 return True 

93 

94 

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

96 try: 

97 ip_net: Union[ipaddress.IPv4Network, ipaddress.IPv6Network] = ( 

98 ipaddress.ip_network(ip) 

99 ) 

100 target_net: Union[ipaddress.IPv4Network, ipaddress.IPv6Network] = ( 

101 ipaddress.ip_network(net) 

102 ) 

103 return ip_net.subnet_of(target_net) # type: ignore[arg-type] 

104 except (TypeError, ValueError): 

105 return False 

106 

107 

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

109 if not no_proxy: 

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

111 " ", "" 

112 ): 

113 no_proxy = v.split(",") 

114 

115 if not no_proxy: 

116 no_proxy = [] 

117 

118 if "*" in no_proxy: 

119 return True 

120 if hostname in no_proxy: 

121 return True 

122 if _is_ip_address(hostname): 

123 return any( 

124 [ 

125 _is_address_in_network(hostname, subnet) 

126 for subnet in no_proxy 

127 if _is_subnet_address(subnet) 

128 ] 

129 ) 

130 # Check domain suffix matching - handle both .domain.com and domain.com formats 

131 # This makes behavior consistent with urllib and other Python libraries 

132 for domain in no_proxy: 

133 if domain.startswith("."): 

134 # Handle .domain.com format 

135 stripped_domain = domain.lstrip(".") 

136 if hostname.endswith(stripped_domain): 

137 return True 

138 else: 

139 # Handle domain.com format (should match subdomains too) 

140 # E.g., "example.com" should match "sub.example.com" 

141 if hostname == domain or hostname.endswith("." + domain): 

142 return True 

143 return False 

144 

145 

146def get_proxy_info( 

147 hostname: str, 

148 is_secure: bool, 

149 proxy_host: Optional[str] = None, 

150 proxy_port: int = 0, 

151 proxy_auth: Optional[tuple] = None, 

152 no_proxy: Optional[list[str]] = None, 

153 proxy_type: str = "http", 

154) -> tuple: 

155 """ 

156 Try to retrieve proxy host and port from environment 

157 if not provided in options. 

158 Result is (proxy_host, proxy_port, proxy_auth). 

159 proxy_auth is tuple of username and password 

160 of proxy authentication information. 

161 

162 Parameters 

163 ---------- 

164 hostname: str 

165 Websocket server name. 

166 is_secure: bool 

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

168 instead of "http_proxy" 

169 proxy_host: str 

170 http proxy host name. 

171 proxy_port: str or int 

172 http proxy port. 

173 no_proxy: list 

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

175 proxy_auth: tuple 

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

177 proxy_type: str 

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

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

180 """ 

181 if _is_no_proxy_host(hostname, no_proxy): 

182 return None, 0, None 

183 

184 if proxy_host: 

185 if not proxy_port: 

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

187 port = proxy_port 

188 auth = proxy_auth 

189 return proxy_host, port, auth 

190 

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

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

193 " ", "" 

194 ) 

195 if value: 

196 proxy = urlparse(value) 

197 auth = ( 

198 (unquote(proxy.username or ""), unquote(proxy.password or "")) 

199 if proxy.username 

200 else None 

201 ) 

202 return proxy.hostname, proxy.port, auth 

203 

204 return None, 0, None