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

1import email.utils 

2import http.cookies 

3import ipaddress 

4import math 

5import time 

6from typing import Optional, Union 

7from http.cookies import CookieError 

8from ._logging import debug, warning 

9 

10""" 

11_cookiejar.py 

12websocket - WebSocket client library for Python 

13 

14Copyright 2026 engn33r 

15 

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

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

18You may obtain a copy of the License at 

19 

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

21 

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

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

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

25See the License for the specific language governing permissions and 

26limitations under the License. 

27""" 

28 

29 

30def _has_control_char(text: str) -> bool: 

31 return any(ord(c) < 0x20 or ord(c) == 0x7F for c in text) 

32 

33 

34class SimpleCookieJar: 

35 def __init__(self) -> None: 

36 self.jar: dict = {} 

37 # (domain key, morsel key) -> epoch after which the morsel is expired 

38 self._expiry: dict = {} 

39 

40 def add( 

41 self, 

42 set_cookie: Optional[Union[str, list]], 

43 host: Optional[str] = None, 

44 ) -> None: 

45 if not set_cookie: 

46 return 

47 cookie_headers = set_cookie if isinstance(set_cookie, list) else [set_cookie] 

48 for cookie_header in cookie_headers: 

49 try: 

50 simple_cookie = http.cookies.SimpleCookie(cookie_header) 

51 except CookieError: 

52 warning("Ignoring invalid Set-Cookie header") 

53 continue 

54 

55 # Store each morsel only under its own domain: morsels that 

56 # carry a domain attribute never leak into a sibling's domain 

57 # entry, and morsels without a domain are dropped (consistent 

58 # with the no-domain rule above). 

59 for v in simple_cookie.values(): 

60 if domain := v.get("domain"): 

61 domain = self._domain_key(domain) 

62 if host is not None and not self._domain_match(host, domain): 

63 debug("Ignoring Set-Cookie for another domain") 

64 continue 

65 if _has_control_char(v.value) or _has_control_char(v.coded_value): 

66 # RFC 6265 4.1.1: cookie values exclude CTLs. A quoted 

67 # value can smuggle them in as octal escapes (\015\012) 

68 # that SimpleCookie decodes into raw CR/LF. 

69 debug("Ignoring Set-Cookie with control characters") 

70 continue 

71 deadline = self._deadline(v) 

72 if deadline is not None and deadline <= time.time(): 

73 # RFC 6265 5.3 step 12: an already-expired cookie 

74 # (Max-Age=0 is a deletion) removes any stored value 

75 cookie = self.jar.get(domain) 

76 if cookie is not None: 

77 cookie.pop(v.key, None) 

78 self._expiry.pop((domain, v.key), None) 

79 if not cookie: 

80 del self.jar[domain] 

81 continue 

82 cookie = self.jar.get(domain) 

83 if cookie is None: 

84 cookie = http.cookies.SimpleCookie() 

85 cookie[v.key] = v 

86 self.jar[domain] = cookie 

87 if deadline is None: 

88 self._expiry.pop((domain, v.key), None) 

89 else: 

90 self._expiry[(domain, v.key)] = deadline 

91 

92 def set(self, set_cookie: str) -> None: 

93 """Replace-style counterpart to add(): clears the existing entry 

94 for each cookie's domain before storing (last write wins), with 

95 the same per-morsel storage rule as add().""" 

96 if set_cookie: 

97 simple_cookie = http.cookies.SimpleCookie(set_cookie) 

98 for v in simple_cookie.values(): 

99 if domain := v.get("domain"): 

100 self._forget(self._domain_key(domain)) 

101 self.add(set_cookie) 

102 

103 def get(self, host: str, secure: bool = False) -> str: 

104 """Cookie header value for host. Morsels with the Secure attribute 

105 are only returned when secure is True (RFC 6265 5.4 step 1).""" 

106 if not host: 

107 return "" 

108 

109 cookies = [] 

110 for domain in list(self.jar): 

111 if not self._host_in_domain(host, domain[1:]): 

112 continue 

113 cookie = self.jar[domain] 

114 now = time.time() 

115 for k in [k for k, _ in cookie.items() if self._expired(domain, k, now)]: 

116 del cookie[k] 

117 self._expiry.pop((domain, k), None) 

118 if not cookie: 

119 del self.jar[domain] 

120 continue 

121 cookies.append(cookie) 

122 

123 return "; ".join( 

124 filter( 

125 None, 

126 sorted( 

127 [ 

128 f"{k}={v.coded_value}" 

129 for cookie in filter(None, cookies) 

130 for k, v in cookie.items() 

131 if secure or not v["secure"] 

132 ] 

133 ), 

134 ) 

135 ) 

136 

137 @staticmethod 

138 def _domain_key(domain: str) -> str: 

139 """Jar key for a Domain attribute: lowercase, one leading dot, no 

140 trailing dot (``Domain=Example.COM.`` and ``.example.com`` are the 

141 same domain).""" 

142 return "." + domain.lower().strip(".") 

143 

144 @staticmethod 

145 def _is_ip(host: str) -> bool: 

146 try: 

147 ipaddress.ip_address(host.strip("[]")) 

148 except ValueError: 

149 return False 

150 return True 

151 

152 @classmethod 

153 def _host_in_domain(cls, host: str, labels: str) -> bool: 

154 """RFC 6265 5.1.3 domain-match of host against a domain without its 

155 leading dot. An IP address only matches itself: the suffix rule 

156 applies to host names, otherwise Domain=0.1 would cover 10.0.0.1 

157 and 192.168.0.1 alike.""" 

158 host = host.lower().rstrip(".") 

159 if cls._is_ip(host) or cls._is_ip(labels): 

160 # an IP host, or a cookie whose Domain is an IP, is host-only 

161 return host.strip("[]") == labels.strip("[]") 

162 return host == labels or host.endswith("." + labels) 

163 

164 @classmethod 

165 def _domain_match(cls, host: str, domain: str) -> bool: 

166 # RFC 6265 5.3 step 6: ignore cookies whose Domain does not 

167 # match the host that set them, so one server cannot plant 

168 # cookies for another. The two-label floor keeps public-suffix 

169 # style values like Domain=com from matching every host under 

170 # them; trailing dots are removed first, so Domain=com. is also a 

171 # single label. An IP address passes only by exact match, so the 

172 # floor does not apply to it (no public-suffix list: a co.uk-style 

173 # suffix would still pass; add a PSL if that matters). 

174 labels = domain.lower().strip(".") 

175 if not cls._host_in_domain(host, labels): 

176 return False 

177 return "." in labels or cls._is_ip(labels) 

178 

179 def _forget(self, domain: str) -> None: 

180 """Drop a domain entry and the expiry records under it.""" 

181 self.jar.pop(domain, None) 

182 for key in [k for k in self._expiry if k[0] == domain]: 

183 del self._expiry[key] 

184 

185 @staticmethod 

186 def _deadline(v: http.cookies.Morsel) -> Optional[float]: 

187 """RFC 6265 5.2: Max-Age overrides Expires; an invalid attribute 

188 value is ignored. Returns the epoch after which the morsel is 

189 expired, or None for a session cookie.""" 

190 max_age = v.get("max-age", "") 

191 if max_age: 

192 try: 

193 delta = int(max_age) 

194 except ValueError: 

195 pass # invalid Max-Age: ignore it, Expires may still apply 

196 else: 

197 try: 

198 return time.time() + delta 

199 except OverflowError: 

200 # RFC 6265 5.2.2: too large to represent means the latest 

201 # (or, when negative, the earliest) representable time 

202 return math.inf if delta > 0 else -math.inf 

203 expires = v.get("expires", "") 

204 if expires: 

205 parsed = email.utils.parsedate_tz(expires) 

206 if parsed is not None: 

207 try: 

208 return email.utils.mktime_tz(parsed) 

209 except (OverflowError, ValueError): 

210 # A date the platform cannot convert (e.g. a year past 

211 # 9999) is an invalid attribute, ignored per RFC 6265 5.2.1 

212 pass 

213 return None 

214 

215 def _expired(self, domain: str, key: str, now: float) -> bool: 

216 deadline = self._expiry.get((domain, key)) 

217 return deadline is not None and deadline <= now