Coverage for an_website/contact/contact.py: 64.545%

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"""A page that allows users to contact the website operator.""" 

15 

16from __future__ import annotations 

17 

18import asyncio 

19import logging 

20import smtplib 

21import ssl 

22import sys 

23import time 

24from collections.abc import Iterable, Mapping 

25from configparser import ConfigParser 

26from datetime import datetime, timezone 

27from email import utils as email_utils 

28from email.message import Message 

29from math import pi 

30from typing import Any, Final, cast 

31from urllib.parse import quote, urlencode 

32 

33import typed_stream 

34from tornado.web import Application, HTTPError, MissingArgumentError 

35 

36from .. import NAME 

37from ..utils.request_handler import HTMLRequestHandler 

38from ..utils.utils import ModuleInfo 

39 

40LOGGER: Final = logging.getLogger(__name__) 

41 

42 

43def get_module_info() -> ModuleInfo: 

44 """Create and return the ModuleInfo for this module.""" 

45 return ModuleInfo( 

46 handlers=((r"/kontakt", ContactPage),), 

47 name="Kontakt-Formular", 

48 description="Nehme mit dem Betreiber der Webseite Kontakt auf.", 

49 path="/kontakt", 

50 keywords=("Kontakt", "Formular"), 

51 aliases=("/contact",), 

52 hidden=True, 

53 ) 

54 

55 

56def apply_contact_stuff_to_app(app: Application, config: ConfigParser) -> None: 

57 """Apply contact stuff to the app.""" 

58 if "/kontakt" not in typed_stream.Stream( 

59 cast(Iterable[ModuleInfo], app.settings.get("MODULE_INFOS", ())) 

60 ).map(lambda m: m.path): 

61 return 

62 

63 contact_address = config.get( 

64 "CONTACT", 

65 "CONTACT_ADDRESS", 

66 fallback=( 

67 f"{NAME.removesuffix('-dev')}@restmail.net" 

68 if app.settings.get("debug") 

69 else "" 

70 ), 

71 ) 

72 sender_address = config.get( 

73 "CONTACT", 

74 "SENDER_ADDRESS", 

75 fallback=( 

76 "Marcell D'Avis <davis@1und1.de>" 

77 if contact_address.endswith("@restmail.net") 

78 else None 

79 ), 

80 ) 

81 

82 if not sender_address: 

83 # the contact form in other mode if no sender address is set 

84 app.settings.update( 

85 CONTACT_ADDRESS=contact_address, 

86 CONTACT_USE_FORM=1, 

87 ) 

88 return 

89 app.settings["CONTACT_ADDRESS"] = None 

90 

91 if not (contact_address and sender_address): 

92 return 

93 

94 app.settings.update( 

95 CONTACT_USE_FORM=2, 

96 CONTACT_RECIPIENTS={ 

97 address.strip() for address in contact_address.split(",") 

98 }, 

99 CONTACT_SMTP_SERVER=config.get( 

100 "CONTACT", 

101 "SMTP_SERVER", 

102 fallback=( 

103 "restmail.net" 

104 if contact_address.endswith("@restmail.net") 

105 else "localhost" 

106 ), 

107 ), 

108 CONTACT_SMTP_PORT=config.getint( 

109 "CONTACT", 

110 "SMTP_PORT", 

111 fallback=25 if contact_address.endswith("@restmail.net") else 587, 

112 ), 

113 CONTACT_SMTP_STARTTLS=config.getboolean( 

114 "CONTACT", 

115 "SMTP_STARTTLS", 

116 fallback=None, 

117 ), 

118 CONTACT_SENDER_ADDRESS=sender_address, 

119 CONTACT_SENDER_USERNAME=config.get( 

120 "CONTACT", "SENDER_USERNAME", fallback=None 

121 ), 

122 CONTACT_SENDER_PASSWORD=config.get( 

123 "CONTACT", "SENDER_PASSWORD", fallback=None 

124 ), 

125 ) 

126 

127 

128def send_message( # pylint: disable=too-many-arguments 

129 message: Message, 

130 from_address: str, 

131 recipients: Iterable[str], 

132 server: str = "localhost", 

133 sender: None | str = None, 

134 username: None | str = None, 

135 password: None | str = None, 

136 starttls: None | bool = None, 

137 port: int = 587, 

138 *, 

139 date: None | datetime = None, 

140) -> dict[str, tuple[int, bytes]]: 

141 """Send an email.""" 

142 recipients = list(recipients) 

143 for spam, eggs in enumerate(recipients): 

144 if eggs.startswith("@"): 

145 recipients[spam] = "contact" + eggs 

146 

147 message["Date"] = email_utils.format_datetime( 

148 date or datetime.now(tz=timezone.utc) 

149 ) 

150 

151 if sender: 

152 message["Sender"] = sender 

153 message["From"] = from_address 

154 message["To"] = ", ".join(recipients) 

155 

156 with smtplib.SMTP(server, port) as smtp: 

157 smtp.set_debuglevel(sys.flags.dev_mode * 2) 

158 smtp.ehlo_or_helo_if_needed() 

159 if starttls is None: 

160 starttls = smtp.has_extn("starttls") 

161 if starttls: 

162 smtp.starttls(context=ssl.create_default_context()) 

163 if username and password: 

164 smtp.login(username, password) 

165 return smtp.send_message(message) 

166 

167 

168def add_geoip_info_to_message( 

169 message: Message, 

170 geoip_info: Mapping[str, Any], 

171 header_prefix: str = "X-GeoIP", 

172) -> None: 

173 """Add GeoIP information to the message.""" 

174 for spam, eggs in geoip_info.items(): 

175 header = f"{header_prefix}-{spam.replace('_', '-')}" 

176 if isinstance(eggs, dict): 

177 add_geoip_info_to_message(message, eggs, header) 

178 else: 

179 message[header] = str(eggs) 

180 

181 

182class ContactPage(HTMLRequestHandler): 

183 """The request handler for the contact page.""" 

184 

185 RATELIMIT_POST_LIMIT = 5 

186 RATELIMIT_POST_COUNT_PER_PERIOD = 1 

187 RATELIMIT_POST_PERIOD = 120 

188 

189 ACCESS_LOG: dict[str, float] = {} 

190 

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

192 """Handle GET requests to the contact page.""" 

193 if not self.settings.get("CONTACT_USE_FORM"): 

194 raise HTTPError(503) 

195 if head: 

196 return 

197 

198 current_time = time.monotonic() 

199 

200 # clean up old items from the access log 

201 for key, value in tuple(self.ACCESS_LOG.items()): 

202 if current_time - value > 10: 

203 del self.ACCESS_LOG[key] 

204 

205 self.ACCESS_LOG[str(self.request.remote_ip)] = current_time 

206 

207 await self.render( 

208 "pages/contact.html", 

209 subject=self.get_argument("subject", ""), 

210 message=self.get_argument("message", ""), 

211 ) 

212 

213 async def post(self) -> None: 

214 """Handle POST requests to the contact page.""" 

215 if not self.settings.get("CONTACT_USE_FORM"): 

216 raise HTTPError(503) 

217 

218 if atime := self.ACCESS_LOG.get(str(self.request.remote_ip)): 

219 del self.ACCESS_LOG[str(self.request.remote_ip)] 

220 if time.monotonic() - atime < pi: 

221 LOGGER.info("Rejected message because of timing") 

222 await self.render( 

223 "pages/empty.html", 

224 text="Nicht gesendet, da du zu schnell warst.", 

225 ) 

226 return 

227 

228 text = self.get_argument("nachricht") 

229 if not text: 

230 raise MissingArgumentError("nachricht") # raise on empty message 

231 

232 name = self.get_argument("name", "") 

233 address_arg = self.get_argument("addresse", "") 

234 address = ( 

235 f"{name} <{address_arg or 'anonymous@foo.bar'}>" 

236 if name 

237 else address_arg or "anonymous@foo.bar" 

238 ) 

239 

240 message = Message() 

241 host = self.request.host_name 

242 name = name or address_arg or "Jemand" 

243 message["Subject"] = str( 

244 self.get_argument("subjekt", "") 

245 or f"{name} will etwas über {host} schreiben." 

246 ) 

247 message.set_payload(text, "UTF-8") 

248 

249 if honeypot := self.get_argument("message", ""): # 🍯 

250 LOGGER.info( 

251 "rejected message: %s", 

252 { 

253 "Subject": message["Subject"], 

254 "message": message, 

255 "address": address, 

256 "geoip": await self.geoip(), 

257 "🍯": honeypot, 

258 }, 

259 ) 

260 await self.render("pages/empty.html", text="Erfolgreich gesendet.") 

261 return 

262 

263 if self.settings.get("CONTACT_USE_FORM") == 1: 

264 query = urlencode( 

265 { 

266 "subject": message["Subject"], 

267 "body": f"{text}\n\nVon: {address}", 

268 }, 

269 quote_via=quote, 

270 ) 

271 self.redirect( 

272 f"mailto:{self.settings.get('CONTACT_ADDRESS')}?{query}" 

273 ) 

274 return 

275 

276 geoip = await self.geoip() 

277 if geoip: 

278 add_geoip_info_to_message(message, geoip) 

279 

280 await asyncio.to_thread( 

281 send_message, 

282 message=message, 

283 from_address=address, 

284 server=self.settings.get("CONTACT_SMTP_SERVER", "localhost"), 

285 sender=self.settings.get("CONTACT_SENDER_ADDRESS"), 

286 recipients=self.settings.get("CONTACT_RECIPIENTS", ""), 

287 username=self.settings.get("CONTACT_SENDER_USERNAME"), 

288 password=self.settings.get("CONTACT_SENDER_PASSWORD"), 

289 starttls=self.settings.get("CONTACT_SMTP_STARTTLS"), 

290 port=self.settings.get("CONTACT_SMTP_PORT", 587), 

291 date=await self.get_time(), 

292 ) 

293 

294 await self.render("pages/empty.html", text="Erfolgreich gesendet.")