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

1import errno 

2import selectors 

3import math 

4import socket 

5from typing import Optional, Union, Any 

6 

7from ._exceptions import ( 

8 WebSocketConnectionClosedException, 

9 WebSocketException, 

10 WebSocketTimeoutException, 

11) 

12from ._ssl_compat import SSLError, SSLEOFError, SSLWantReadError, SSLWantWriteError 

13from ._utils import extract_error_code, extract_err_message 

14 

15""" 

16_socket.py 

17websocket - WebSocket client library for Python 

18 

19Copyright 2026 engn33r 

20 

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

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

23You may obtain a copy of the License at 

24 

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

26 

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

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

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

30See the License for the specific language governing permissions and 

31limitations under the License. 

32""" 

33 

34DEFAULT_SOCKET_OPTION = [(socket.SOL_TCP, socket.TCP_NODELAY, 1)] 

35if hasattr(socket, "SO_KEEPALIVE"): 

36 DEFAULT_SOCKET_OPTION.append((socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)) 

37if hasattr(socket, "TCP_KEEPIDLE"): 

38 DEFAULT_SOCKET_OPTION.append((socket.SOL_TCP, socket.TCP_KEEPIDLE, 30)) 

39if hasattr(socket, "TCP_KEEPINTVL"): 

40 DEFAULT_SOCKET_OPTION.append((socket.SOL_TCP, socket.TCP_KEEPINTVL, 10)) 

41if hasattr(socket, "TCP_KEEPCNT"): 

42 DEFAULT_SOCKET_OPTION.append((socket.SOL_TCP, socket.TCP_KEEPCNT, 3)) 

43MAX_LINE_BYTES = 65536 

44 

45# Raised by a retried recv()/send() whose socket the selector reported ready 

46# but which still cannot make progress. BlockingIOError covers EAGAIN and 

47# EWOULDBLOCK; a TLS socket may want either direction on either call. 

48_WOULD_BLOCK = (BlockingIOError, SSLWantReadError, SSLWantWriteError) 

49 

50_default_timeout = None 

51 

52__all__ = [ 

53 "DEFAULT_SOCKET_OPTION", 

54 "sock_opt", 

55 "setdefaulttimeout", 

56 "getdefaulttimeout", 

57 "recv", 

58 "recv_line", 

59 "send", 

60] 

61 

62 

63class sock_opt: 

64 def __init__( 

65 self, sockopt: Optional[list[tuple]], sslopt: Optional[dict[str, Any]] 

66 ) -> None: 

67 if sockopt is None: 

68 sockopt = [] 

69 if sslopt is None: 

70 sslopt = {} 

71 self.sockopt = sockopt 

72 self.sslopt = sslopt 

73 self.timeout: Optional[Union[int, float]] = None 

74 

75 

76def _validate_timeout(timeout: Optional[Union[int, float]]) -> None: 

77 """Reject timeouts that the socket layer would misread or reject late.""" 

78 if timeout is None: 

79 return 

80 # bool subclasses int; True would become 1.0s on the socket 

81 if isinstance(timeout, bool) or not isinstance(timeout, (int, float)): 

82 raise TypeError( 

83 "timeout must be an int or float, not %r" % (type(timeout).__name__,) 

84 ) 

85 if not math.isfinite(timeout): 

86 raise ValueError("timeout must be a finite number, got %r" % (timeout,)) 

87 

88 

89def setdefaulttimeout(timeout: Optional[Union[int, float]]) -> None: 

90 """ 

91 Set the global timeout setting to connect. 

92 

93 Parameters 

94 ---------- 

95 timeout: int or float 

96 default socket timeout time (in seconds). 

97 None clears the default timeout. 

98 """ 

99 global _default_timeout 

100 _validate_timeout(timeout) 

101 _default_timeout = timeout 

102 

103 

104def getdefaulttimeout() -> Optional[Union[int, float]]: 

105 """ 

106 Get default timeout 

107 

108 Returns 

109 ---------- 

110 _default_timeout: int or float 

111 Return the global timeout setting (in seconds) to connect. 

112 """ 

113 return _default_timeout 

114 

115 

116def recv(sock: socket.socket, bufsize: int) -> bytes: 

117 if not sock: 

118 raise WebSocketConnectionClosedException("socket is already closed.") 

119 

120 def _recv(): 

121 try: 

122 return sock.recv(bufsize) 

123 except SSLWantReadError: 

124 # Don't return None implicitly - fall through to retry logic 

125 pass 

126 except socket.error as exc: 

127 error_code = extract_error_code(exc) 

128 if error_code not in [errno.EAGAIN, errno.EWOULDBLOCK]: 

129 raise 

130 # Don't return None implicitly - fall through to retry logic 

131 

132 # Retry logic using selector for both SSLWantReadError and EAGAIN/EWOULDBLOCK 

133 sel = selectors.DefaultSelector() 

134 try: 

135 sel.register(sock, selectors.EVENT_READ) 

136 r = sel.select(sock.gettimeout()) 

137 finally: 

138 sel.close() 

139 

140 if r: 

141 try: 

142 return sock.recv(bufsize) 

143 except _WOULD_BLOCK as e: 

144 # Ready but still nothing to read (e.g. a partial TLS record on 

145 # a non-blocking socket): a timeout, not a raw socket error. 

146 raise WebSocketTimeoutException( 

147 "Connection timed out waiting for data" 

148 ) from e 

149 else: 

150 # Selector timeout should raise WebSocketTimeoutException 

151 # not return None which gets misclassified as connection closed 

152 raise WebSocketTimeoutException("Connection timed out waiting for data") 

153 

154 try: 

155 bytes_ = _recv() 

156 except TimeoutError: 

157 raise WebSocketTimeoutException("Connection timed out") 

158 except socket.timeout as e: 

159 message = extract_err_message(e) 

160 raise WebSocketTimeoutException(message) 

161 except SSLError as e: 

162 message = extract_err_message(e) 

163 if isinstance(message, str) and "timed out" in message: 

164 raise WebSocketTimeoutException(message) 

165 else: 

166 raise 

167 

168 if bytes_ is None: 

169 raise WebSocketConnectionClosedException("Connection to remote host was lost.") 

170 if not bytes_: 

171 raise WebSocketConnectionClosedException("Connection to remote host was lost.") 

172 

173 return bytes_ 

174 

175 

176def recv_line(sock: socket.socket) -> bytes: 

177 line = [] 

178 while True: 

179 c = recv(sock, 1) 

180 line.append(c) 

181 if c == b"\n": 

182 break 

183 if len(line) > MAX_LINE_BYTES: 

184 raise WebSocketException(f"Header line exceeds {MAX_LINE_BYTES} bytes") 

185 return b"".join(line) 

186 

187 

188def send(sock: socket.socket, data: Union[bytes, str]) -> int: 

189 if isinstance(data, str): 

190 data = data.encode("utf-8") 

191 

192 if not sock: 

193 raise WebSocketConnectionClosedException("socket is already closed.") 

194 

195 def _send() -> int: 

196 try: 

197 return sock.send(data) 

198 except SSLEOFError: 

199 raise WebSocketConnectionClosedException("socket is already closed.") 

200 except SSLWantWriteError: 

201 pass 

202 except socket.error as exc: 

203 error_code = extract_error_code(exc) 

204 if error_code is None: 

205 raise 

206 if error_code not in [errno.EAGAIN, errno.EWOULDBLOCK]: 

207 raise 

208 

209 sel = selectors.DefaultSelector() 

210 try: 

211 sel.register(sock, selectors.EVENT_WRITE) 

212 w = sel.select(sock.gettimeout()) 

213 finally: 

214 sel.close() 

215 

216 if w: 

217 try: 

218 return sock.send(data) 

219 except _WOULD_BLOCK as e: 

220 raise WebSocketTimeoutException( 

221 "Connection timed out waiting to send data" 

222 ) from e 

223 raise WebSocketTimeoutException("Connection timed out waiting to send data") 

224 

225 try: 

226 return _send() 

227 except socket.timeout as e: 

228 message = extract_err_message(e) 

229 raise WebSocketTimeoutException(message) 

230 except (OSError, SSLError) as e: 

231 message = extract_err_message(e) 

232 if isinstance(message, str) and "timed out" in message: 

233 raise WebSocketTimeoutException(message) 

234 else: 

235 raise