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

1import ipaddress 

2import os 

3from typing import Optional 

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 return ipaddress.ip_network(ip).subnet_of(ipaddress.ip_network(net)) 

98 except TypeError: 

99 return False 

100 

101 

102def _is_no_proxy_host(hostname: str, no_proxy: Optional[list[str]]) -> 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[str]] = 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 or ""), unquote(proxy.password or "")) 

184 if proxy.username 

185 else None 

186 ) 

187 return proxy.hostname, proxy.port, auth 

188 

189 return None, 0, None