Coverage for an_website / utils / request_handler.py: 92.500%

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 

14""" 

15Useful request handlers used by other modules. 

16 

17This should only contain request handlers and the get_module_info function. 

18""" 

19 

20import logging 

21from collections.abc import Mapping 

22from http.client import responses 

23from typing import Any, ClassVar, Final, override 

24from urllib.parse import unquote 

25 

26import regex 

27from tornado.web import HTTPError 

28 

29from .base_request_handler import BaseRequestHandler 

30from .utils import ( 

31 SUS_PATHS, 

32 get_close_matches, 

33 remove_suffix_ignore_case, 

34 replace_umlauts, 

35) 

36 

37LOGGER: Final = logging.getLogger(__name__) 

38 

39 

40class HTMLRequestHandler(BaseRequestHandler): 

41 """A request handler that serves HTML.""" 

42 

43 POSSIBLE_CONTENT_TYPES: ClassVar[tuple[str, ...]] = ( 

44 "text/html", 

45 "text/plain", 

46 "text/markdown", 

47 "application/vnd.asozial.dynload+json", 

48 ) 

49 

50 

51class APIRequestHandler(BaseRequestHandler): 

52 """The base API request handler.""" 

53 

54 POSSIBLE_CONTENT_TYPES: ClassVar[tuple[str, ...]] = ( 

55 "application/json", 

56 "application/yaml", 

57 ) 

58 

59 

60class NotFoundHandler(BaseRequestHandler): 

61 """Show a 404 page if no other RequestHandler is used.""" 

62 

63 @override 

64 def initialize(self, *args: Any, **kwargs: Any) -> None: 

65 """Do nothing to have default title and description.""" 

66 if "module_info" not in kwargs: 

67 kwargs["module_info"] = None 

68 super().initialize(*args, **kwargs) 

69 

70 @override 

71 async def prepare(self) -> None: 

72 """Throw a 404 HTTP error or redirect to another page.""" 

73 self.now = await self.get_time() 

74 

75 if self.request.method not in {"GET", "HEAD"}: 

76 raise HTTPError(404) 

77 

78 new_path = regex.sub(r"/+", "/", self.request.path.rstrip("/")).replace( 

79 "_", "-" 

80 ) 

81 

82 for ext in (".html", ".htm", ".php"): 

83 new_path = remove_suffix_ignore_case(new_path, f"/index{ext}") 

84 new_path = remove_suffix_ignore_case(new_path, ext) 

85 

86 new_path = replace_umlauts(new_path) 

87 

88 if new_path.lower() in SUS_PATHS: 

89 self.set_status(469, reason="Nice Try") 

90 return self.write_error(469) 

91 

92 if new_path and new_path != self.request.path: 

93 return self.redirect(self.fix_url(new_path=new_path), True) 

94 

95 this_path_normalized = unquote(new_path).strip("/").lower() 

96 

97 paths: Mapping[str, str] = self.settings.get("NORMED_PATHS") or {} 

98 

99 if p := paths.get(this_path_normalized): 

100 return self.redirect(self.fix_url(new_path=p), False) 

101 

102 if len(this_path_normalized) <= 1 and self.request.path != "/": 

103 return self.redirect(self.fix_url(new_path="/")) 

104 

105 prefixes = tuple( 

106 (p, repl) 

107 for p, repl in paths.items() 

108 if this_path_normalized.startswith(f"{p}/") 

109 if f"/{p}" != repl.lower() 

110 if p != "api" # api should not be a prefix 

111 ) 

112 

113 if len(prefixes) == 1: 

114 ((prefix, replacement),) = prefixes 

115 return self.redirect( 

116 self.fix_url( 

117 new_path=f"{replacement.strip('/')}" 

118 f"{this_path_normalized.removeprefix(prefix)}" 

119 ), 

120 False, 

121 ) 

122 if prefixes: 

123 LOGGER.error( 

124 "Too many prefixes %r for path %s", prefixes, self.request.path 

125 ) 

126 

127 matches = get_close_matches(this_path_normalized, paths, count=1) 

128 if matches: 

129 return self.redirect( 

130 self.fix_url(new_path=paths[matches[0]]), False 

131 ) 

132 

133 self.set_status(404) 

134 self.write_error(404) 

135 

136 

137class ErrorPage(HTMLRequestHandler): 

138 """A request handler that shows the error page.""" 

139 

140 _success_status: int = 200 

141 """The status code that is expected to be returned.""" 

142 

143 @override 

144 def clear(self) -> None: 

145 """Reset all headers and content for this response.""" 

146 super().clear() 

147 self._success_status = 200 

148 

149 @override 

150 async def get(self, code: str, *, head: bool = False) -> None: 

151 """Show the error page.""" 

152 # pylint: disable=unused-argument 

153 status_code = int(code) 

154 reason = ( 

155 "Nice Try" if status_code == 469 else responses.get(status_code, "") 

156 ) 

157 # set the status code if it is allowed 

158 if status_code not in (204, 304) and not 100 <= status_code < 200: 

159 self.set_status(status_code, reason) 

160 self._success_status = status_code 

161 return await self.render( 

162 "error.html", 

163 status=status_code, 

164 reason=reason, 

165 description=self.get_error_page_description(status_code), 

166 is_traceback=False, 

167 ) 

168 

169 @override 

170 def get_status(self) -> int: 

171 """Hack the status code. 

172 

173 This hacks the status code to be 200 if the status code is expected. 

174 This avoids sending error logs to APM or Webhooks in case of success. 

175 

176 This depends on the fact that Tornado internally uses self._status_code 

177 to set the status code in the response and self.get_status() when 

178 deciding how to log the request. 

179 """ 

180 status = super().get_status() 

181 if status == self._success_status: 

182 return 200 

183 return status 

184 

185 

186class ZeroDivision(BaseRequestHandler): 

187 """A request handler that raises an error.""" 

188 

189 @override 

190 async def prepare(self) -> None: 

191 """Divide by zero and raise an error.""" 

192 self.now = await self.get_time() 

193 self.handle_accept_header(self.POSSIBLE_CONTENT_TYPES) 

194 if self.request.method != "OPTIONS": 

195 420 / 0 # pylint: disable=pointless-statement