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

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 2026 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 # An entry that is empty after removing leading dots ("a,,b", a leading or 

116 # trailing comma, ".") names no domain. Kept, it would make the suffix check 

117 # below hostname.endswith("."), so every hostname ending in a dot would 

118 # silently bypass the proxy. urllib skips empty entries the same way. 

119 no_proxy = [entry for entry in no_proxy or [] if entry.lstrip(".")] 

120 

121 if "*" in no_proxy: 

122 return True 

123 if hostname in no_proxy: 

124 return True 

125 if _is_ip_address(hostname): 

126 return any( 

127 [ 

128 _is_address_in_network(hostname, subnet) 

129 for subnet in no_proxy 

130 if _is_subnet_address(subnet) 

131 ] 

132 ) 

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

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

135 for domain in no_proxy: 

136 if domain.startswith("."): 

137 # Handle .domain.com format - match the domain itself and its 

138 # subdomains, but not unrelated hosts that merely share the suffix 

139 # (e.g. ".example.com" must not match "notexample.com"). 

140 stripped_domain = domain.lstrip(".") 

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

142 return True 

143 else: 

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

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

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

147 return True 

148 return False 

149 

150 

151def get_proxy_info( 

152 hostname: str, 

153 is_secure: bool, 

154 proxy_host: Optional[str] = None, 

155 proxy_port: int = 0, 

156 proxy_auth: Optional[tuple] = None, 

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

158 proxy_type: str = "http", 

159) -> tuple: 

160 """ 

161 Try to retrieve proxy host and port from environment 

162 if not provided in options. 

163 Result is (proxy_host, proxy_port, proxy_auth). 

164 proxy_auth is tuple of username and password 

165 of proxy authentication information. 

166 

167 Parameters 

168 ---------- 

169 hostname: str 

170 Websocket server name. 

171 is_secure: bool 

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

173 instead of "http_proxy" 

174 proxy_host: str 

175 http proxy host name. 

176 proxy_port: str or int 

177 http proxy port. Required when proxy_host is set. 

178 no_proxy: list 

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

180 proxy_auth: tuple 

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

182 proxy_type: str 

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

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

185 """ 

186 if _is_no_proxy_host(hostname, no_proxy): 

187 return None, 0, None 

188 

189 if proxy_host: 

190 if not proxy_port: 

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

192 port = proxy_port 

193 auth = proxy_auth 

194 return proxy_host, port, auth 

195 

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

197 # httpoxy (CVE-2016-1000110). This only matters when running inside a CGI 

198 # script (REQUEST_METHOD is set): there the web server turns the client's 

199 # "Proxy:" request header into the HTTP_PROXY environment variable, so it is 

200 # ignored, as urllib does. Outside CGI (plain scripts, WSGI, ASGI) request 

201 # headers never reach os.environ and HTTP_PROXY is honored as usual. The 

202 # lowercase http_proxy cannot come from a request header and is kept, but 

203 # it must be matched by exact key case: on Windows os.environ is 

204 # case-insensitive, so os.environ.get("http_proxy") would return HTTP_PROXY. 

205 if not is_secure and "REQUEST_METHOD" in os.environ: 

206 value = next((v for k, v in os.environ.items() if k == env_key), None) 

207 else: 

208 value = os.environ.get(env_key) 

209 if value is None: 

210 value = os.environ.get(env_key.upper()) 

211 value = (value or "").replace(" ", "") 

212 if value: 

213 proxy = urlparse(value) 

214 # The environment proxy is always spoken to in plaintext HTTP CONNECT, 

215 # so a TLS or SOCKS proxy would receive the request and any 

216 # Proxy-Authorization credentials in cleartext. 

217 if proxy.scheme == "https" or proxy.scheme.startswith("socks"): 

218 raise WebSocketProxyException( 

219 f"{env_key} uses a {proxy.scheme}:// proxy, which is not supported " 

220 "from the environment; use an http:// proxy, or pass http_proxy_host " 

221 "and proxy_type for a SOCKS proxy" 

222 ) 

223 auth = ( 

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

225 if proxy.username 

226 else None 

227 ) 

228 return proxy.hostname, proxy.port, auth 

229 

230 return None, 0, None