Coverage for app/venv/lib/python3.14/site-packages/weblate/accounts/tasks.py: 65%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7import logging 

8from datetime import timedelta 

9from email.mime.image import MIMEImage 

10from smtplib import SMTP, SMTPConnectError 

11from types import MethodType 

12from typing import TYPE_CHECKING, TypedDict 

13 

14import sentry_sdk 

15from celery.schedules import crontab 

16from django.conf import settings 

17from django.core.mail import EmailMultiAlternatives, get_connection 

18from django.core.mail.backends.smtp import EmailBackend as DjangoSMTPEmailBackend 

19from django.db import transaction 

20from django.utils.timezone import now 

21from social_django.models import Code, Partial 

22 

23from weblate.utils.celery import app 

24from weblate.utils.errors import report_error 

25from weblate.utils.html import HTML2Text 

26from weblate.utils.icons import load_icon 

27 

28if TYPE_CHECKING: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 from collections.abc import Generator 

30 

31 from weblate.accounts.notifications import Notification 

32 

33LOGGER = logging.getLogger("weblate.smtp") 

34 

35# The batch size needs to be below exim's default connection_max_messages = 500. 

36EMAIL_BATCH_SIZE = 200 

37 

38 

39class OutgoingEmail(TypedDict): 

40 address: str 

41 subject: str 

42 body: str 

43 headers: dict[str, str] 

44 

45 

46@app.task(trail=False) 

47def cleanup_social_auth() -> None: 

48 """Cleanup expired partial social authentications.""" 

49 age = now() - timedelta(seconds=settings.AUTH_TOKEN_VALID) 

50 # Delete old not verified codes 

51 Code.objects.filter(verified=False, timestamp__lt=age).delete() 

52 

53 # Delete old partial data 

54 Partial.objects.filter(timestamp__lt=age).delete() 

55 

56 

57@app.task(trail=False) 

58def cleanup_auditlog() -> None: 

59 """Cleanup old auditlog entries.""" 

60 from weblate.accounts.models import AuditLog 

61 

62 timestamp = now() 

63 

64 # Cleanup old entries 

65 AuditLog.objects.filter( 

66 timestamp__lt=timestamp - timedelta(days=settings.AUDITLOG_EXPIRY) 

67 ).delete() 

68 

69 # Finalize pending two-factor entries, these happen due to 

70 # WebAuthn keys being added in two stages. Mature entries older than 5 minutes 

71 # but look only two hours into past for performance reasons 

72 for audit in AuditLog.objects.filter( 72 ↛ 79line 72 didn't jump to line 79 because the loop on line 72 never started

73 timestamp__range=( 

74 timestamp - timedelta(hours=2), 

75 timestamp - timedelta(minutes=5), 

76 ), 

77 activity="twofactor-add", 

78 ): 

79 if "skip_notify" in audit.params: 

80 del audit.params["skip_notify"] 

81 audit.save(update_fields=["params"]) 

82 

83 

84class NotificationFactory: 

85 def __init__(self) -> None: 

86 self.outgoing: list[OutgoingEmail] = [] 

87 self.instances: dict[str, Notification] = {} 

88 

89 def for_action(self, action: int) -> Generator[Notification]: 

90 from weblate.accounts.notifications import NOTIFICATIONS_ACTIONS 

91 

92 if action not in NOTIFICATIONS_ACTIONS: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 return 

94 for notification_cls in NOTIFICATIONS_ACTIONS[action]: 

95 name = notification_cls.get_name() 

96 try: 

97 yield self.instances[name] 

98 except KeyError: 

99 result = self.instances[name] = notification_cls(self.outgoing) 

100 yield result 

101 

102 def send_queued(self) -> None: 

103 if self.outgoing: 

104 queue_mails(self.outgoing) 

105 self.outgoing.clear() 

106 

107 

108@app.task(trail=False) 

109@transaction.atomic 

110def notify_changes(change_ids: list[int]) -> None: 

111 from weblate.trans.models import Change 

112 

113 changes = Change.objects.prefetch_for_render().filter(pk__in=change_ids) 

114 factory = NotificationFactory() 

115 

116 for change in changes.iterator(chunk_size=200): 

117 change.fill_in_prefetched() 

118 for notification in factory.for_action(change.action): 

119 notification.notify_immediate(change) 

120 factory.send_queued() 

121 

122 

123@transaction.atomic 

124def notify_digest(method) -> None: 

125 from weblate.accounts.notifications import NOTIFICATIONS 

126 

127 outgoing: list[OutgoingEmail] = [] 

128 for notification_cls in NOTIFICATIONS: 

129 notification = notification_cls(outgoing) 

130 getattr(notification, method)() 

131 if outgoing: 

132 queue_mails(outgoing) 

133 

134 

135@app.task(trail=False) 

136def notify_daily() -> None: 

137 notify_digest("notify_daily") 

138 

139 

140@app.task(trail=False) 

141def notify_weekly() -> None: 

142 notify_digest("notify_weekly") 

143 

144 

145@app.task(trail=False) 

146def notify_monthly() -> None: 

147 notify_digest("notify_monthly") 

148 

149 

150@app.task(trail=False) 

151def notify_auditlog(log_id: int, email: str) -> None: 

152 from weblate.accounts.models import AuditLog 

153 from weblate.accounts.notifications import send_notification_email 

154 

155 audit = AuditLog.objects.get(pk=log_id) 

156 send_notification_email( 

157 audit.user.profile.language if audit.user else "en", 

158 [email], 

159 "account_activity", 

160 context={ 

161 "message": audit.get_message, 

162 "extra_message": audit.get_extra_message, 

163 "address": audit.shortened_address, 

164 "user_agent": audit.user_agent, 

165 }, 

166 info=f"{audit.activity} from {audit.address}", 

167 ) 

168 

169 

170SMTP_DATA_PATCH = "_weblate_patched_data" 

171 

172 

173def weblate_logging_smtp_data(self, msg): 

174 (code, msg) = getattr(self, SMTP_DATA_PATCH)(msg) 

175 if code == 250: 

176 LOGGER.debug("SMTP completed (%s): %s", code, msg.decode()) 

177 else: 

178 LOGGER.error("SMTP failed (%s): %s", code, msg.decode()) 

179 return (code, msg) 

180 

181 

182def monkey_patch_smtp_logging(connection): 

183 if isinstance(connection, DjangoSMTPEmailBackend): 

184 # Ensure the connection is open 

185 connection.open() 

186 

187 # Monkey patch smtplib.SMTP or smtplib.SMTP_SSL 

188 backend = connection.connection 

189 if isinstance(backend, SMTP) and not hasattr(backend, SMTP_DATA_PATCH): 

190 setattr(backend, SMTP_DATA_PATCH, backend.data) 

191 backend.data = MethodType(weblate_logging_smtp_data, backend) # type: ignore[method-assign] 

192 

193 return connection 

194 

195 

196def queue_mails(mails: list[OutgoingEmail]) -> None: 

197 """Enqueue e-mails for delivery in reasonable batches.""" 

198 for offset in range(0, len(mails), EMAIL_BATCH_SIZE): 

199 send_mails.delay(mails[offset : offset + EMAIL_BATCH_SIZE]) 

200 

201 

202@app.task( 

203 trail=False, 

204 autoretry_for=(SMTPConnectError, OSError), 

205 retry_backoff=600, 

206 retry_backoff_max=3600, 

207) 

208def send_mails(mails: list[OutgoingEmail]) -> None: 

209 """Send multiple mails in single connection.""" 

210 images = [] 

211 with sentry_sdk.start_span(op="email.images"): 

212 for name in ("email-logo.png", "email-logo-footer.png"): 

213 image = MIMEImage(load_icon(name, auto_prefix=False)) 

214 image.add_header("Content-ID", f"<{name}@cid.weblate.org>") 

215 image.add_header("Content-Disposition", "inline", filename=name) 

216 images.append(image) 

217 

218 with sentry_sdk.start_span(op="email.connect"): 

219 connection = get_connection() 

220 try: 

221 connection.open() 

222 except Exception: 

223 LOGGER.exception("Could not initialize e-mail backend") 

224 report_error("Could not send notifications") 

225 connection.close() 

226 return 

227 connection = monkey_patch_smtp_logging(connection) 

228 

229 html2text = HTML2Text() 

230 

231 try: 

232 for mail in mails: 

233 with sentry_sdk.start_span(op="email.text"): 

234 text = html2text.handle(mail["body"]) 

235 email = EmailMultiAlternatives( 

236 settings.EMAIL_SUBJECT_PREFIX + mail["subject"], 

237 text, 

238 to=[mail["address"]], 

239 headers=mail["headers"], 

240 connection=connection, 

241 ) 

242 email.mixed_subtype = "related" 

243 for image in images: 

244 email.attach(image) 

245 email.attach_alternative(mail["body"], "text/html") 

246 with sentry_sdk.start_span(op="email.send"): 

247 LOGGER.debug("sending e-mail to %s", mail["address"]) 

248 email.send() 

249 finally: 

250 connection.close() 

251 

252 

253@app.on_after_finalize.connect 

254def setup_periodic_tasks(sender, **kwargs) -> None: 

255 sender.add_periodic_task(3600, cleanup_social_auth.s(), name="social-auth-cleanup") 

256 sender.add_periodic_task(3600, cleanup_auditlog.s(), name="auditlog-cleanup") 

257 sender.add_periodic_task( 

258 crontab(hour=1, minute=0), notify_daily.s(), name="notify-daily" 

259 ) 

260 sender.add_periodic_task( 

261 crontab(hour=2, minute=0, day_of_week="mon"), 

262 notify_weekly.s(), 

263 name="notify-weekly", 

264 ) 

265 sender.add_periodic_task( 

266 crontab(hour=3, minute=0, day_of_month=1), 

267 notify_monthly.s(), 

268 name="notify-monthly", 

269 )