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

1""" 

2_http.py 

3websocket - WebSocket client library for Python 

4 

5Copyright 2025 engn33r 

6 

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

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

9You may obtain a copy of the License at 

10 

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

12 

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

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

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

16See the License for the specific language governing permissions and 

17limitations under the License. 

18""" 

19 

20import errno 

21import os 

22import socket 

23from base64 import encodebytes as base64encode 

24 

25from ._exceptions import ( 

26 WebSocketAddressException, 

27 WebSocketException, 

28 WebSocketProxyException, 

29) 

30from ._logging import debug, dump, trace 

31from ._socket import DEFAULT_SOCKET_OPTION, recv_line, send 

32from ._ssl_compat import HAVE_SSL, ssl 

33from ._url import get_proxy_info, parse_url 

34 

35__all__ = ["proxy_info", "connect", "read_headers"] 

36 

37try: 

38 from python_socks._errors import ProxyConnectionError, ProxyError, ProxyTimeoutError 

39 from python_socks._types import ProxyType 

40 from python_socks.sync import Proxy 

41 

42 HAVE_PYTHON_SOCKS = True 

43except: 

44 HAVE_PYTHON_SOCKS = False 

45 

46 class ProxyError(Exception): 

47 pass 

48 

49 class ProxyTimeoutError(Exception): 

50 pass 

51 

52 class ProxyConnectionError(Exception): 

53 pass 

54 

55 

56class proxy_info: 

57 def __init__(self, **options): 

58 self.proxy_host = options.get("http_proxy_host", None) 

59 if self.proxy_host: 

60 self.proxy_port = options.get("http_proxy_port", 0) 

61 self.auth = options.get("http_proxy_auth", None) 

62 self.no_proxy = options.get("http_no_proxy", None) 

63 self.proxy_protocol = options.get("proxy_type", "http") 

64 # Note: If timeout not specified, default python-socks timeout is 60 seconds 

65 self.proxy_timeout = options.get("http_proxy_timeout", None) 

66 if self.proxy_protocol not in [ 

67 "http", 

68 "socks4", 

69 "socks4a", 

70 "socks5", 

71 "socks5h", 

72 ]: 

73 raise ProxyError( 

74 "Only http, socks4, socks5 proxy protocols are supported" 

75 ) 

76 else: 

77 self.proxy_port = 0 

78 self.auth = None 

79 self.no_proxy = None 

80 self.proxy_protocol = "http" 

81 

82 

83def _start_proxied_socket(url: str, options, proxy) -> tuple: 

84 if not HAVE_PYTHON_SOCKS: 

85 raise WebSocketException( 

86 "Python Socks is needed for SOCKS proxying but is not available" 

87 ) 

88 

89 hostname, port, resource, is_secure = parse_url(url) 

90 

91 if proxy.proxy_protocol == "socks4": 

92 rdns = False 

93 proxy_type = ProxyType.SOCKS4 

94 # socks4a sends DNS through proxy 

95 elif proxy.proxy_protocol == "socks4a": 

96 rdns = True 

97 proxy_type = ProxyType.SOCKS4 

98 elif proxy.proxy_protocol == "socks5": 

99 rdns = False 

100 proxy_type = ProxyType.SOCKS5 

101 # socks5h sends DNS through proxy 

102 elif proxy.proxy_protocol == "socks5h": 

103 rdns = True 

104 proxy_type = ProxyType.SOCKS5 

105 

106 ws_proxy = Proxy.create( 

107 proxy_type=proxy_type, 

108 host=proxy.proxy_host, 

109 port=int(proxy.proxy_port), 

110 username=proxy.auth[0] if proxy.auth else None, 

111 password=proxy.auth[1] if proxy.auth else None, 

112 rdns=rdns, 

113 ) 

114 

115 sock = ws_proxy.connect(hostname, port, timeout=proxy.proxy_timeout) 

116 

117 if is_secure: 

118 if HAVE_SSL: 

119 sock = _ssl_socket(sock, options.sslopt, hostname) 

120 else: 

121 raise WebSocketException("SSL not available.") 

122 

123 return sock, (hostname, port, resource) 

124 

125 

126def connect(url: str, options, proxy, socket): 

127 # Use _start_proxied_socket() only for socks4 or socks5 proxy 

128 # Use _tunnel() for http proxy 

129 # TODO: Use python-socks for http protocol also, to standardize flow 

130 if proxy.proxy_host and not socket and proxy.proxy_protocol != "http": 

131 return _start_proxied_socket(url, options, proxy) 

132 

133 hostname, port_from_url, resource, is_secure = parse_url(url) 

134 

135 if socket: 

136 return socket, (hostname, port_from_url, resource) 

137 

138 addrinfo_list, need_tunnel, auth = _get_addrinfo_list( 

139 hostname, port_from_url, is_secure, proxy 

140 ) 

141 if not addrinfo_list: 

142 raise WebSocketException(f"Host not found.: {hostname}:{port_from_url}") 

143 

144 sock = None 

145 try: 

146 sock = _open_socket(addrinfo_list, options.sockopt, options.timeout) 

147 if need_tunnel: 

148 sock = _tunnel(sock, hostname, port_from_url, auth) 

149 

150 if is_secure: 

151 if HAVE_SSL: 

152 sock = _ssl_socket(sock, options.sslopt, hostname) 

153 else: 

154 raise WebSocketException("SSL not available.") 

155 

156 return sock, (hostname, port_from_url, resource) 

157 except: 

158 if sock: 

159 sock.close() 

160 raise 

161 

162 

163def _get_addrinfo_list(hostname, port: int, is_secure: bool, proxy) -> tuple: 

164 phost, pport, pauth = get_proxy_info( 

165 hostname, 

166 is_secure, 

167 proxy.proxy_host, 

168 proxy.proxy_port, 

169 proxy.auth, 

170 proxy.no_proxy, 

171 ) 

172 try: 

173 # when running on windows 10, getaddrinfo without socktype returns a socktype 0. 

174 # This generates an error exception: `_on_error: exception Socket type must be stream or datagram, not 0` 

175 # or `OSError: [Errno 22] Invalid argument` when creating socket. Force the socket type to SOCK_STREAM. 

176 if not phost: 

177 addrinfo_list = socket.getaddrinfo( 

178 hostname, port, 0, socket.SOCK_STREAM, socket.SOL_TCP 

179 ) 

180 return addrinfo_list, False, None 

181 else: 

182 pport = pport and pport or 80 

183 # when running on windows 10, the getaddrinfo used above 

184 # returns a socktype 0. This generates an error exception: 

185 # _on_error: exception Socket type must be stream or datagram, not 0 

186 # Force the socket type to SOCK_STREAM 

187 addrinfo_list = socket.getaddrinfo( 

188 phost, pport, 0, socket.SOCK_STREAM, socket.SOL_TCP 

189 ) 

190 return addrinfo_list, True, pauth 

191 except socket.gaierror as e: 

192 raise WebSocketAddressException(e) 

193 

194 

195def _open_socket(addrinfo_list, sockopt, timeout): 

196 err = None 

197 for addrinfo in addrinfo_list: 

198 family, socktype, proto = addrinfo[:3] 

199 sock = socket.socket(family, socktype, proto) 

200 sock.settimeout(timeout) 

201 for opts in DEFAULT_SOCKET_OPTION: 

202 sock.setsockopt(*opts) 

203 for opts in sockopt: 

204 sock.setsockopt(*opts) 

205 

206 address = addrinfo[4] 

207 err = None 

208 while not err: 

209 try: 

210 sock.connect(address) 

211 except socket.error as error: 

212 sock.close() 

213 error.remote_ip = str(address[0]) 

214 try: 

215 eConnRefused = ( 

216 errno.ECONNREFUSED, 

217 errno.WSAECONNREFUSED, 

218 errno.ENETUNREACH, 

219 ) 

220 except AttributeError: 

221 eConnRefused = (errno.ECONNREFUSED, errno.ENETUNREACH) 

222 if error.errno not in eConnRefused: 

223 raise error 

224 err = error 

225 continue 

226 else: 

227 break 

228 else: 

229 continue 

230 break 

231 else: 

232 if err: 

233 raise err 

234 

235 return sock 

236 

237 

238def _wrap_sni_socket(sock: socket.socket, sslopt: dict, hostname, check_hostname): 

239 context = sslopt.get("context", None) 

240 if not context: 

241 context = ssl.SSLContext(sslopt.get("ssl_version", ssl.PROTOCOL_TLS_CLIENT)) 

242 # Non default context need to manually enable SSLKEYLOGFILE support by setting the keylog_filename attribute. 

243 # For more details see also: 

244 # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#context-creation 

245 # * https://docs.python.org/3.8/library/ssl.html?highlight=sslkeylogfile#ssl.SSLContext.keylog_filename 

246 keylog_file = os.environ.get("SSLKEYLOGFILE") 

247 if keylog_file is not None: 

248 context.keylog_filename = keylog_file 

249 

250 if sslopt.get("cert_reqs", ssl.CERT_NONE) != ssl.CERT_NONE: 

251 cafile = sslopt.get("ca_certs", None) 

252 capath = sslopt.get("ca_cert_path", None) 

253 if cafile or capath: 

254 try: 

255 context.load_verify_locations(cafile=cafile, capath=capath) 

256 except (FileNotFoundError, ssl.SSLError, ValueError) as e: 

257 raise WebSocketException(f"SSL CA certificate loading failed: {e}") 

258 elif hasattr(context, "load_default_certs"): 

259 try: 

260 context.load_default_certs(ssl.Purpose.SERVER_AUTH) 

261 except ssl.SSLError as e: 

262 raise WebSocketException( 

263 f"SSL default certificate loading failed: {e}" 

264 ) 

265 if sslopt.get("certfile", None): 

266 try: 

267 context.load_cert_chain( 

268 sslopt["certfile"], 

269 sslopt.get("keyfile", None), 

270 sslopt.get("password", None), 

271 ) 

272 except (FileNotFoundError, ValueError) as e: 

273 raise WebSocketException(f"SSL client certificate loading failed: {e}") 

274 except ssl.SSLError as e: 

275 raise WebSocketException(f"SSL client certificate loading failed: {e}") 

276 

277 # Python 3.10 switch to PROTOCOL_TLS_CLIENT defaults to "cert_reqs = ssl.CERT_REQUIRED" and "check_hostname = True" 

278 # If both disabled, set check_hostname before verify_mode 

279 # see https://github.com/liris/websocket-client/commit/b96a2e8fa765753e82eea531adb19716b52ca3ca#commitcomment-10803153 

280 if sslopt.get("cert_reqs", ssl.CERT_NONE) == ssl.CERT_NONE and not sslopt.get( 

281 "check_hostname", False 

282 ): 

283 context.check_hostname = False 

284 context.verify_mode = ssl.CERT_NONE 

285 else: 

286 context.check_hostname = sslopt.get("check_hostname", True) 

287 context.verify_mode = sslopt.get("cert_reqs", ssl.CERT_REQUIRED) 

288 

289 if "ciphers" in sslopt: 

290 try: 

291 context.set_ciphers(sslopt["ciphers"]) 

292 except ssl.SSLError as e: 

293 raise WebSocketException(f"SSL cipher configuration failed: {e}") 

294 if "cert_chain" in sslopt: 

295 try: 

296 cert_chain = sslopt["cert_chain"] 

297 if not isinstance(cert_chain, (tuple, list)) or len(cert_chain) != 3: 

298 raise ValueError( 

299 "cert_chain must be a tuple/list of (certfile, keyfile, password)" 

300 ) 

301 certfile, keyfile, password = cert_chain 

302 context.load_cert_chain(certfile, keyfile, password) 

303 except ValueError: 

304 raise 

305 except (FileNotFoundError, ssl.SSLError) as e: 

306 raise WebSocketException( 

307 f"SSL client certificate configuration failed: {e}" 

308 ) 

309 if "ecdh_curve" in sslopt: 

310 try: 

311 context.set_ecdh_curve(sslopt["ecdh_curve"]) 

312 except ValueError as e: 

313 raise WebSocketException(f"SSL ECDH curve configuration failed: {e}") 

314 

315 return context.wrap_socket( 

316 sock, 

317 do_handshake_on_connect=sslopt.get("do_handshake_on_connect", True), 

318 suppress_ragged_eofs=sslopt.get("suppress_ragged_eofs", True), 

319 server_hostname=hostname, 

320 ) 

321 

322 

323def _ssl_socket(sock: socket.socket, user_sslopt: dict, hostname): 

324 sslopt: dict = {"cert_reqs": ssl.CERT_REQUIRED} 

325 sslopt.update(user_sslopt) 

326 

327 cert_path = os.environ.get("WEBSOCKET_CLIENT_CA_BUNDLE") 

328 if ( 

329 cert_path 

330 and os.path.isfile(cert_path) 

331 and user_sslopt.get("ca_certs", None) is None 

332 ): 

333 sslopt["ca_certs"] = cert_path 

334 elif ( 

335 cert_path 

336 and os.path.isdir(cert_path) 

337 and user_sslopt.get("ca_cert_path", None) is None 

338 ): 

339 sslopt["ca_cert_path"] = cert_path 

340 

341 if sslopt.get("server_hostname", None): 

342 hostname = sslopt["server_hostname"] 

343 

344 check_hostname = sslopt.get("check_hostname", True) 

345 sock = _wrap_sni_socket(sock, sslopt, hostname, check_hostname) 

346 

347 return sock 

348 

349 

350def _tunnel(sock: socket.socket, host, port: int, auth) -> socket.socket: 

351 debug("Connecting proxy...") 

352 connect_header = f"CONNECT {host}:{port} HTTP/1.1\r\n" 

353 connect_header += f"Host: {host}:{port}\r\n" 

354 

355 # TODO: support digest auth. 

356 if auth and auth[0]: 

357 auth_str = auth[0] 

358 if auth[1]: 

359 auth_str += f":{auth[1]}" 

360 encoded_str = base64encode(auth_str.encode()).strip().decode().replace("\n", "") 

361 connect_header += f"Proxy-Authorization: Basic {encoded_str}\r\n" 

362 connect_header += "\r\n" 

363 dump("request header", connect_header) 

364 

365 send(sock, connect_header) 

366 

367 try: 

368 status, _, _ = read_headers(sock) 

369 except (socket.error, WebSocketException) as e: 

370 raise WebSocketProxyException(str(e)) 

371 

372 if status != 200: 

373 raise WebSocketProxyException(f"failed CONNECT via proxy status: {status}") 

374 

375 return sock 

376 

377 

378def read_headers(sock: socket.socket) -> tuple: 

379 status = None 

380 status_message = None 

381 headers: dict = {} 

382 trace("--- response header ---") 

383 

384 while True: 

385 line = recv_line(sock) 

386 line = line.decode("utf-8").strip() 

387 if not line: 

388 break 

389 trace(line) 

390 if not status: 

391 status_info = line.split(" ", 2) 

392 status = int(status_info[1]) 

393 if len(status_info) > 2: 

394 status_message = status_info[2] 

395 else: 

396 kv = line.split(":", 1) 

397 if len(kv) != 2: 

398 raise WebSocketException("Invalid header") 

399 key, value = kv 

400 if key.lower() == "set-cookie" and headers.get("set-cookie"): 

401 existing_cookie = headers.get("set-cookie") 

402 if existing_cookie is not None: 

403 headers["set-cookie"] = existing_cookie + "; " + value.strip() 

404 else: 

405 headers["set-cookie"] = value.strip() 

406 else: 

407 headers[key.lower()] = value.strip() 

408 

409 trace("-----------------------") 

410 

411 return status, headers, status_message