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

1""" 

2_http.py 

3websocket - WebSocket client library for Python 

4 

5Copyright 2026 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 

24from typing import Any, Tuple 

25 

26from ._exceptions import ( 

27 WebSocketAddressException, 

28 WebSocketException, 

29 WebSocketProxyException, 

30) 

31from ._logging import debug, dump, trace 

32from ._socket import DEFAULT_SOCKET_OPTION, recv_line, send 

33from ._ssl_compat import HAVE_SSL, ssl 

34from ._url import get_proxy_info, parse_url 

35 

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

37 

38# Import python_socks if available, otherwise define fallback classes 

39try: 

40 from python_socks._errors import ProxyConnectionError, ProxyError, ProxyTimeoutError 

41 from python_socks._types import ProxyType 

42 from python_socks.sync import Proxy 

43 

44 HAVE_PYTHON_SOCKS = True 

45except ImportError: 

46 HAVE_PYTHON_SOCKS = False 

47 

48 class ProxyError(Exception): # type: ignore[no-redef] 

49 pass 

50 

51 class ProxyTimeoutError(Exception): # type: ignore[no-redef] 

52 pass 

53 

54 class ProxyConnectionError(Exception): # type: ignore[no-redef] 

55 pass 

56 

57 class ProxyType: # type: ignore[no-redef] 

58 pass 

59 

60 

61class proxy_info: 

62 def __init__(self, **options): 

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

64 if self.proxy_host: 

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

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

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

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

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

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

71 if self.proxy_protocol not in [ 

72 "http", 

73 "socks4", 

74 "socks4a", 

75 "socks5", 

76 "socks5h", 

77 ]: 

78 raise ProxyError( 

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

80 ) 

81 else: 

82 self.proxy_port = 0 

83 self.auth = None 

84 self.no_proxy = None 

85 self.proxy_protocol = "http" 

86 self.proxy_timeout = None 

87 

88 

89def _start_proxied_socket( 

90 url: str, options: Any, proxy: Any 

91) -> Tuple[socket.socket, Tuple[str, int, str]]: 

92 if proxy.proxy_host and not proxy.proxy_port: 

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

94 if not HAVE_PYTHON_SOCKS: 

95 raise WebSocketException( 

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

97 ) 

98 

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

100 

101 if proxy.proxy_protocol == "socks4": 

102 rdns = False 

103 proxy_type = ProxyType.SOCKS4 

104 # socks4a sends DNS through proxy 

105 elif proxy.proxy_protocol == "socks4a": 

106 rdns = True 

107 proxy_type = ProxyType.SOCKS4 

108 elif proxy.proxy_protocol == "socks5": 

109 rdns = False 

110 proxy_type = ProxyType.SOCKS5 

111 # socks5h sends DNS through proxy 

112 elif proxy.proxy_protocol == "socks5h": 

113 rdns = True 

114 proxy_type = ProxyType.SOCKS5 

115 

116 ws_proxy = Proxy.create( 

117 proxy_type=proxy_type, 

118 host=proxy.proxy_host, 

119 port=int(proxy.proxy_port), 

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

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

122 rdns=rdns, 

123 ) 

124 

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

126 

127 if is_secure: 

128 if HAVE_SSL: 

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

130 else: 

131 raise WebSocketException("SSL not available.") 

132 

133 return sock, (hostname, port, resource) 

134 

135 

136def connect( 

137 url: str, options: Any, proxy: Any, socket: Any 

138) -> Tuple[socket.socket, Tuple[str, int, str]]: 

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

140 # Use _tunnel() for http proxy 

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

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

143 return _start_proxied_socket(url, options, proxy) 

144 

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

146 

147 if socket: 

148 return socket, (hostname, port_from_url, resource) 

149 

150 addrinfo_list, need_tunnel, auth = _get_addrinfo_list( 

151 hostname, port_from_url, is_secure, proxy 

152 ) 

153 if not addrinfo_list: 

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

155 

156 sock = None 

157 try: 

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

159 if need_tunnel: 

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

161 

162 if is_secure: 

163 if HAVE_SSL: 

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

165 else: 

166 raise WebSocketException("SSL not available.") 

167 

168 return sock, (hostname, port_from_url, resource) 

169 except: 

170 if sock: 

171 sock.close() 

172 raise 

173 

174 

175def _get_addrinfo_list( 

176 hostname: str, port: int, is_secure: bool, proxy: Any 

177) -> Tuple[list, bool, Any]: 

178 try: 

179 phost, pport, pauth = get_proxy_info( 

180 hostname, 

181 is_secure, 

182 proxy.proxy_host, 

183 proxy.proxy_port, 

184 proxy.auth, 

185 proxy.no_proxy, 

186 ) 

187 except TypeError as e: 

188 raise WebSocketAddressException(e) 

189 try: 

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

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

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

193 if not phost: 

194 addrinfo_list = socket.getaddrinfo( 

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

196 ) 

197 return addrinfo_list, False, None 

198 else: 

199 pport = pport and pport or 80 

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

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

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

203 # Force the socket type to SOCK_STREAM 

204 addrinfo_list = socket.getaddrinfo( 

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

206 ) 

207 return addrinfo_list, True, pauth 

208 except (socket.gaierror, TypeError) as e: 

209 raise WebSocketAddressException(e) 

210 

211 

212def _open_socket(addrinfo_list, sockopt, timeout): 

213 err = None 

214 for addrinfo in addrinfo_list: 

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

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

217 sock.settimeout(timeout) 

218 for opts in DEFAULT_SOCKET_OPTION: 

219 sock.setsockopt(*opts) 

220 for opts in sockopt: 

221 sock.setsockopt(*opts) 

222 

223 address = addrinfo[4] 

224 err = None 

225 while not err: 

226 try: 

227 sock.connect(address) 

228 except socket.error as error: 

229 sock.close() 

230 error.remote_ip = str(address[0]) # type: ignore[attr-defined] 

231 eConnRefused = ( 

232 errno.ECONNREFUSED, 

233 getattr(errno, "WSAECONNREFUSED", errno.ECONNREFUSED), 

234 errno.ENETUNREACH, 

235 ) 

236 if error.errno not in eConnRefused: 

237 raise error 

238 err = error 

239 continue 

240 else: 

241 break 

242 else: 

243 continue 

244 break 

245 else: 

246 if err: 

247 raise err 

248 

249 return sock 

250 

251 

252def _wrap_sni_socket( 

253 sock: socket.socket, sslopt: dict, hostname: str, check_hostname: bool 

254) -> Any: 

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

256 if not context: 

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

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

259 # For more details see also: 

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

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

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

263 if keylog_file is not None: 

264 context.keylog_filename = keylog_file 

265 

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

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

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

269 if cafile or capath: 

270 try: 

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

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

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

274 elif hasattr(context, "load_default_certs"): 

275 try: 

276 context.load_default_certs(ssl.Purpose.SERVER_AUTH) 

277 except ssl.SSLError as e: 

278 raise WebSocketException( 

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

280 ) 

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

282 try: 

283 context.load_cert_chain( 

284 sslopt["certfile"], 

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

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

287 ) 

288 except (FileNotFoundError, ValueError) as e: 

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

290 except ssl.SSLError as e: 

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

292 

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

294 # If both disabled, set check_hostname before verify_mode 

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

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

297 "check_hostname", False 

298 ): 

299 context.check_hostname = False 

300 context.verify_mode = ssl.CERT_NONE 

301 else: 

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

303 cert_reqs = sslopt.get("cert_reqs", ssl.CERT_REQUIRED) 

304 if check_hostname and cert_reqs == ssl.CERT_NONE: 

305 raise WebSocketException( 

306 "SSL certificate verification configuration failed: " 

307 "check_hostname cannot be enabled when cert_reqs is CERT_NONE" 

308 ) 

309 context.check_hostname = check_hostname 

310 context.verify_mode = cert_reqs 

311 

312 if "ciphers" in sslopt: 

313 try: 

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

315 except ssl.SSLError as e: 

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

317 if "cert_chain" in sslopt: 

318 try: 

319 cert_chain = sslopt["cert_chain"] 

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

321 raise ValueError( 

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

323 ) 

324 certfile, keyfile, password = cert_chain 

325 context.load_cert_chain(certfile, keyfile, password) 

326 except ValueError: 

327 raise 

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

329 raise WebSocketException( 

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

331 ) 

332 if "ecdh_curve" in sslopt: 

333 try: 

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

335 except ValueError as e: 

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

337 

338 return context.wrap_socket( 

339 sock, 

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

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

342 server_hostname=hostname, 

343 ) 

344 

345 

346def _ssl_socket(sock: socket.socket, user_sslopt: dict, hostname: str) -> Any: 

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

348 sslopt.update(user_sslopt) 

349 

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

351 if ( 

352 cert_path 

353 and os.path.isfile(cert_path) 

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

355 ): 

356 sslopt["ca_certs"] = cert_path 

357 elif ( 

358 cert_path 

359 and os.path.isdir(cert_path) 

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

361 ): 

362 sslopt["ca_cert_path"] = cert_path 

363 

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

365 hostname = sslopt["server_hostname"] 

366 

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

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

369 

370 return sock 

371 

372 

373def _tunnel(sock: socket.socket, host: str, port: int, auth: Any) -> socket.socket: 

374 debug("Connecting proxy...") 

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

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

377 

378 # TODO: support digest auth. 

379 if auth and auth[0]: 

380 auth_str = auth[0] 

381 if auth[1]: 

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

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

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

385 connect_header += "\r\n" 

386 dump("request header", connect_header) 

387 

388 send(sock, connect_header) 

389 

390 try: 

391 status, _, _ = read_headers(sock) 

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

393 raise WebSocketProxyException(str(e)) 

394 

395 if status != 200: 

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

397 

398 return sock 

399 

400 

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

402 status = None 

403 status_message = None 

404 headers: dict = {} 

405 trace("--- response header ---") 

406 

407 while True: 

408 line = recv_line(sock) 

409 try: 

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

411 except UnicodeDecodeError as e: 

412 raise WebSocketException( 

413 f"Invalid header line, not valid UTF-8 at byte {e.start}: {line!r}" 

414 ) 

415 if not line: 

416 break 

417 trace(line) 

418 if not status: 

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

420 try: 

421 status = int(status_info[1]) 

422 except (IndexError, ValueError): 

423 raise WebSocketException(f"Invalid status line: {line!r}") 

424 if len(status_info) > 2: 

425 status_message = status_info[2] 

426 else: 

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

428 if len(kv) != 2: 

429 raise WebSocketException("Invalid header") 

430 key, value = kv 

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

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

433 if existing_cookie is not None: 

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

435 else: 

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

437 else: 

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

439 

440 trace("-----------------------") 

441 

442 return status, headers, status_message