Coverage for an_website/patches/__init__.py: 91.791%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13# pylint: disable=protected-access 

14 

15"""Patches that improve everything.""" 

16 

17import http.client 

18import json as stdlib_json # pylint: disable=preferred-module 

19from collections.abc import Callable 

20from configparser import RawConfigParser 

21from contextlib import suppress 

22from pathlib import Path 

23from threading import Thread 

24from types import MethodType 

25from typing import Any 

26from urllib.parse import urlsplit 

27 

28import certifi 

29import defusedxml 

30import jsonpickle # type: ignore[import-untyped] 

31import multiprocessing_importlib_resources 

32import orjson 

33import pycurl 

34import tornado.httputil 

35import yaml 

36from emoji import EMOJI_DATA 

37from pillow_jxl import JpegXLImagePlugin # noqa: F401 

38from setproctitle import setthreadtitle 

39from tornado.httpclient import AsyncHTTPClient, HTTPRequest 

40from tornado.httputil import ( 

41 HTTPFile, 

42 HTTPHeaders, 

43 HTTPInputError, 

44 HTTPServerRequest, 

45 ParseBodyConfig, 

46) 

47from tornado.log import gen_log 

48from tornado.web import GZipContentEncoding, RequestHandler 

49 

50from .. import CA_BUNDLE_PATH, MEDIA_TYPES 

51from . import braille, json # noqa: F401 # pylint: disable=reimported 

52 

53 

54def apply() -> None: 

55 """Improve.""" 

56 multiprocessing_importlib_resources._patch() 

57 patch_certifi() 

58 patch_configparser() 

59 patch_emoji() 

60 patch_http() 

61 patch_json() 

62 patch_jsonpickle() 

63 patch_threading() 

64 patch_xml() 

65 

66 patch_tornado_arguments() 

67 patch_tornado_gzip() 

68 patch_tornado_httpclient() 

69 patch_tornado_logs() 

70 

71 

72def patch_certifi() -> None: 

73 """Make everything use our CA bundle.""" 

74 certifi.where = lambda: CA_BUNDLE_PATH 

75 certifi.contents = lambda: Path(certifi.where()).read_text("ASCII") 

76 

77 

78def patch_configparser() -> None: 

79 """Make configparser funky.""" 

80 RawConfigParser.BOOLEAN_STATES.update( # type: ignore[attr-defined] 

81 { 

82 "sure": True, 

83 "nope": False, 

84 "accept": True, 

85 "reject": False, 

86 "enabled": True, 

87 "disabled": False, 

88 } 

89 ) 

90 

91 

92def patch_emoji() -> None: 

93 """Add cool new emoji.""" 

94 EMOJI_DATA["🐱\u200D💻"] = { 

95 "de": ":hacker_katze:", 

96 "en": ":hacker_cat:", 

97 "status": 2, 

98 "E": 1, 

99 } 

100 for de_name, en_name, rect in ( 

101 ("rot", "red", "🟥"), 

102 ("blau", "blue", "🟦"), 

103 ("orang", "orange", "🟧"), 

104 ("gelb", "yellow", "🟨"), 

105 ("grün", "green", "🟩"), 

106 ("lilan", "purple", "🟪"), 

107 ("braun", "brown", "🟫"), 

108 ): 

109 EMOJI_DATA[f"🫙\u200D{rect}"] = { 

110 "de": f":{de_name}es_glas:", 

111 "en": f":{en_name}_jar:", 

112 "status": 2, 

113 "E": 14, 

114 } 

115 EMOJI_DATA[f"🏳\uFE0F\u200D{rect}"] = { 

116 "de": f":{de_name}e_flagge:", 

117 "en": f":{en_name}_flag:", 

118 "status": 2, 

119 "E": 11, 

120 } 

121 EMOJI_DATA[f"\u2691\uFE0F\u200D{rect}"] = { 

122 "de": f":tief{de_name}e_flagge:", 

123 "en": f":deep_{en_name}_flag:", 

124 "status": 2, 

125 "E": 11, 

126 } 

127 

128 

129def patch_http() -> None: 

130 """Add response code 420.""" 

131 http.client.responses[420] = "Enhance Your Calm" 

132 

133 

134def patch_json() -> None: 

135 """Replace json with orjson.""" 

136 if getattr(stdlib_json, "_omegajson", False): 

137 return 

138 stdlib_json.dumps = json.dumps 

139 stdlib_json.dump = json.dump # type: ignore[assignment] 

140 stdlib_json.loads = json.loads # type: ignore[assignment] 

141 stdlib_json.load = json.load 

142 

143 

144def patch_jsonpickle() -> None: 

145 """Make jsonpickle return bytes.""" 

146 jsonpickle.load_backend("orjson") 

147 jsonpickle.set_preferred_backend("orjson") 

148 jsonpickle.enable_fallthrough(False) 

149 

150 

151def patch_threading() -> None: 

152 """Set thread names.""" 

153 _bootstrap = Thread._bootstrap # type: ignore[attr-defined] 

154 

155 def bootstrap(self: Thread) -> None: 

156 with suppress(Exception): 

157 setthreadtitle(self.name) 

158 _bootstrap(self) 

159 

160 Thread._bootstrap = bootstrap # type: ignore[attr-defined] 

161 

162 

163def patch_tornado_arguments() -> None: # noqa: C901 

164 """Improve argument parsing.""" 

165 # pylint: disable=too-complex 

166 

167 def ensure_bytes(value: Any) -> bytes: 

168 """Return the value as bytes.""" 

169 if isinstance(value, bool): 

170 return b"true" if value else b"false" 

171 if isinstance(value, bytes): 

172 return value 

173 return str(value).encode("UTF-8") 

174 

175 def parse_body_arguments( # pylint: disable=too-many-arguments 

176 content_type: str, 

177 body: bytes, 

178 arguments: dict[str, list[bytes]], 

179 files: dict[str, list[HTTPFile]], 

180 headers: None | HTTPHeaders = None, 

181 *, 

182 config: ParseBodyConfig | None = None, 

183 _: Callable[..., None] = tornado.httputil.parse_body_arguments, 

184 ) -> None: 

185 # pylint: disable=too-many-branches 

186 if content_type.startswith("application/json"): 

187 if headers and "Content-Encoding" in headers: 

188 gen_log.warning( 

189 "Unsupported Content-Encoding: %s", 

190 headers["Content-Encoding"], 

191 ) 

192 return 

193 try: 

194 spam = orjson.loads(body) 

195 except Exception as exc: 

196 raise HTTPInputError(f"Invalid JSON body: {exc}") from exc 

197 if not isinstance(spam, dict): 

198 return 

199 for key, value in spam.items(): 

200 if value is not None: 

201 arguments.setdefault(key, []).append(ensure_bytes(value)) 

202 elif content_type.startswith("application/yaml"): 

203 if headers and "Content-Encoding" in headers: 

204 gen_log.warning( 

205 "Unsupported Content-Encoding: %s", 

206 headers["Content-Encoding"], 

207 ) 

208 return 

209 try: 

210 spam = yaml.safe_load(body) 

211 except Exception as exc: 

212 raise HTTPInputError(f"Invalid YAML body: {exc}") from exc 

213 if not isinstance(spam, dict): 

214 return 

215 for key, value in spam.items(): 

216 if value is not None: 

217 arguments.setdefault(key, []).append(ensure_bytes(value)) 

218 else: 

219 _(content_type, body, arguments, files, headers, config=config) 

220 

221 parse_body_arguments.__doc__ = tornado.httputil.parse_body_arguments.__doc__ 

222 

223 tornado.httputil.parse_body_arguments = parse_body_arguments 

224 

225 

226def patch_tornado_gzip() -> None: 

227 """Use gzip for more content types.""" 

228 GZipContentEncoding.CONTENT_TYPES = { 

229 type for type, data in MEDIA_TYPES.items() if data.get("compressible") 

230 } 

231 

232 

233def patch_tornado_httpclient() -> None: # fmt: off 

234 """Make requests quick.""" 

235 BACON = 0x75800 # noqa: N806 # pylint: disable=invalid-name 

236 EGGS = 1 << 25 # noqa: N806 # pylint: disable=invalid-name 

237 

238 AsyncHTTPClient.configure("tornado.curl_httpclient.CurlAsyncHTTPClient") 

239 

240 def prepare_curl_callback(self: HTTPRequest, curl: pycurl.Curl) -> None: 

241 # pylint: disable=c-extension-no-member, useless-suppression 

242 if urlsplit(self.url).scheme == "https": # noqa: SIM102 

243 if (ver := pycurl.version_info())[2] >= BACON and ver[4] & EGGS: 

244 curl.setopt(pycurl.HTTP_VERSION, pycurl.CURL_HTTP_VERSION_3) 

245 

246 original_request_init = HTTPRequest.__init__ 

247 

248 def request_init(self: HTTPRequest, *args: Any, **kwargs: Any) -> None: 

249 if len(args) < 18: # there are too many positional arguments here 

250 prepare_curl_method = MethodType(prepare_curl_callback, self) 

251 kwargs.setdefault("prepare_curl_callback", prepare_curl_method) 

252 original_request_init(self, *args, **kwargs) 

253 

254 request_init.__doc__ = HTTPRequest.__init__.__doc__ 

255 

256 HTTPRequest.__init__ = request_init # type: ignore[method-assign] 

257 

258 

259def patch_tornado_logs() -> None: 

260 """Anonymize Tornado logs.""" 

261 # pylint: disable=import-outside-toplevel 

262 from ..utils.utils import SUS_PATHS, anonymize_ip 

263 

264 RequestHandler._request_summary = ( # type: ignore[method-assign] 

265 lambda self: "%s %s (%s)" # pylint: disable=consider-using-f-string 

266 % ( 

267 self.request.method, 

268 self.request.uri, 

269 ( 

270 self.request.remote_ip 

271 if self.request.path == "/robots.txt" 

272 or self.request.path.lower() in SUS_PATHS 

273 else anonymize_ip(self.request.remote_ip, ignore_invalid=True) 

274 ), 

275 ) 

276 ) 

277 

278 HTTPServerRequest.__repr__ = ( # type: ignore[method-assign] 

279 lambda self: "%s(%s)" # pylint: disable=consider-using-f-string 

280 % ( 

281 self.__class__.__name__, 

282 ", ".join( 

283 [ 

284 "%s=%r" # pylint: disable=consider-using-f-string 

285 % ( 

286 n, 

287 getattr(self, n), 

288 ) 

289 for n in ("protocol", "host", "method", "uri", "version") 

290 ] 

291 ), 

292 ) 

293 ) 

294 

295 

296def patch_xml() -> None: 

297 """Make XML safer.""" 

298 defusedxml.defuse_stdlib() # type: ignore[attr-defined] 

299 defusedxml.xmlrpc.monkey_patch() # type: ignore[attr-defined]