Coverage for ivatar/ivataraccount/models.py: 85%

1""" 

2Our models for ivatar.ivataraccount 

3""" 

4 

5import base64 

6import contextlib 

7import hashlib 

8import time 

9from io import BytesIO 

10from os import urandom 

11from typing import List 

12from urllib.error import HTTPError, URLError 

13from ivatar.utils import urlopen, Bluesky 

14from urllib.parse import urlsplit, urlunsplit, quote 

15import logging 

16 

17from PIL import Image 

18from django.contrib.auth.models import User 

19from django.contrib import messages 

20from django.db import models 

21from django.utils import timezone 

22from django.http import HttpResponseRedirect 

23from django.urls import reverse_lazy, reverse 

24from django.utils.translation import gettext_lazy as _ 

25from django.core.cache import cache 

26from django.core.exceptions import ObjectDoesNotExist 

27from django.core.mail import send_mail 

28from django.template.loader import render_to_string 

29from openid.association import Association as OIDAssociation 

30from openid.store import nonce as oidnonce 

31from openid.store.interface import OpenIDStore 

32from django.db.models.signals import post_save, post_delete, pre_delete 

33from django.dispatch import receiver 

34 

35from libravatar import libravatar_url 

36 

37from django.conf import settings 

38from ivatar.photo_cache import ( 

39 invalidate_photo_cache, 

40 invalidate_digest_cache, 

41 pregenerate_photo_thumbnails, 

42) 

43from ivatar.utils import openid_variations 

44from .gravatar import get_photo as get_gravatar_photo 

45 

46# Initialize logger 

47logger = logging.getLogger("ivatar") 

48 

49# Access settings through django.conf.settings 

50MAX_LENGTH_EMAIL = getattr(settings, "MAX_LENGTH_EMAIL", 254) 

51MAX_PIXELS = getattr(settings, "MAX_PIXELS", 7000) 

52AVATAR_MAX_SIZE = getattr(settings, "AVATAR_MAX_SIZE", 512) 

53JPEG_QUALITY = getattr(settings, "JPEG_QUALITY", 85) 

54MAX_LENGTH_URL = getattr(settings, "MAX_LENGTH_URL", 255) 

55SECURE_BASE_URL = getattr(settings, "SECURE_BASE_URL", "https://avatars.linux-kernel.at/avatar/") 

56SITE_NAME = getattr(settings, "SITE_NAME", "libravatar") 

57DEFAULT_FROM_EMAIL = getattr(settings, "DEFAULT_FROM_EMAIL", "ivatar@mg.linux-kernel.at") 

58 

59 

60def file_format(image_type): 

61 """ 

62 Helper method returning a short image type 

63 """ 

64 if image_type in ("JPEG", "MPO"): 

65 return "jpg" 

66 elif image_type == "PNG": 

67 return "png" 

68 elif image_type == "GIF": 

69 return "gif" 

70 elif image_type == "WEBP": 

71 return "webp" 

72 return None 

73 

74 

75def pil_format(image_type): 

76 """ 

77 Helper method returning the 'encoder name' for PIL 

78 """ 

79 if image_type in ("jpg", "jpeg", "mpo"): 

80 return "JPEG" 

81 elif image_type == "png": 

82 return "PNG" 

83 elif image_type == "gif": 

84 return "GIF" 

85 elif image_type == "webp": 

86 return "WEBP" 

87 

88 logger.info("Unsupported file format: %s", image_type) 

89 return None 

90 

91 

92class UserPreference(models.Model): 

93 """ 

94 Holds the user users preferences 

95 """ 

96 

97 THEMES = ( 

98 ("default", "Default theme"), 

99 ("clime", "climes theme"), 

100 ("green", "green theme"), 

101 ("red", "red theme"), 

102 ) 

103 

104 theme = models.CharField( 

105 max_length=10, 

106 choices=THEMES, 

107 default="default", 

108 ) 

109 

110 user = models.OneToOneField( 

111 User, 

112 on_delete=models.deletion.CASCADE, 

113 primary_key=True, 

114 ) 

115 

116 def __str__(self): 

117 return "Preference (%i) for %s" % (self.pk, self.user) 

118 

119 

120class BaseAccountModel(models.Model): 

121 """ 

122 Base, abstract model, holding fields we use in all cases 

123 """ 

124 

125 user = models.ForeignKey( 

126 User, 

127 on_delete=models.deletion.CASCADE, 

128 ) 

129 ip_address = models.GenericIPAddressField(unpack_ipv4=True, null=True) 

130 add_date = models.DateTimeField(default=timezone.now) 

131 

132 class Meta: # pylint: disable=too-few-public-methods 

133 """ 

134 Class attributes 

135 """ 

136 

137 abstract = True 

138 

139 

140class AccessStat(models.Model): 

141 """ 

142 Abstract base model for per-object access statistics. 

143 

144 Stores the access_count in a separate, narrow table so that frequent 

145 counter updates do not generate dead tuples on the wide parent tables 

146 and remain eligible for PostgreSQL HOT (Heap Only Tuple) updates. 

147 Subclasses add a OneToOneField (used as primary key) to the parent model. 

148 """ 

149 

150 access_count = models.BigIntegerField(default=0, editable=False) 

151 # No index on access_count — keeps HOT updates possible on these rows. 

152 

153 class Meta: # pylint: disable=too-few-public-methods 

154 """ 

155 Class attributes 

156 """ 

157 

158 abstract = True 

159 

160 

161class Photo(BaseAccountModel): 

162 """ 

163 Model holding the photos and information about them 

164 """ 

165 

166 ip_address = models.GenericIPAddressField(unpack_ipv4=True) 

167 data = models.BinaryField() 

168 format = models.CharField(max_length=4) 

169 class Meta: # pylint: disable=too-few-public-methods 

170 """ 

171 Class attributes 

172 """ 

173 

174 verbose_name = _("photo") 

175 verbose_name_plural = _("photos") 

176 

177 def import_image(self, service_name, email_address): 

178 """ 

179 Allow to import image from other (eg. Gravatar) service 

180 """ 

181 image_url = False 

182 

183 if service_name == "Gravatar": 

184 if gravatar := get_gravatar_photo(email_address): 

185 image_url = gravatar["image_url"] 

186 

187 if service_name == "Libravatar": 

188 image_url = libravatar_url(email_address, size=AVATAR_MAX_SIZE) 

189 

190 if not image_url: 

191 return False # pragma: no cover 

192 try: 

193 image = urlopen(image_url) 

194 except HTTPError as exc: 

195 logger.warning( 

196 f"{service_name} import failed with an HTTP error: {exc.code}" 

197 ) 

198 return False 

199 except URLError as exc: 

200 logger.warning(f"{service_name} import failed: {exc.reason}") 

201 return False 

202 data = image.read() 

203 

204 try: 

205 img = Image.open(BytesIO(data)) 

206 # How am I supposed to test this? 

207 except ValueError: # pragma: no cover 

208 return False # pragma: no cover 

209 

210 self.format = file_format(img.format) 

211 if not self.format: 

212 logger.warning(f"Unable to determine format: {img}") 

213 return False # pragma: no cover 

214 self.data = data 

215 super().save() 

216 return True 

217 

218 def save( 

219 self, force_insert=False, force_update=False, using=None, update_fields=None 

220 ): 

221 """ 

222 Override save from parent, taking care about the image 

223 """ 

224 # Use PIL to read the file format 

225 try: 

226 img = Image.open(BytesIO(self.data)) 

227 except Exception as exc: # pylint: disable=broad-except 

228 # For debugging only 

229 logger.error(f"Exception caught in Photo.save(): {exc}") 

230 return False 

231 self.format = file_format(img.format) 

232 if not self.format: 

233 logger.error("Format not recognized") 

234 return False 

235 result = super().save( 

236 force_insert=force_insert, 

237 force_update=force_update, 

238 using=using, 

239 update_fields=update_fields, 

240 ) 

241 return result 

242 

243 def delete(self, using=None, keep_parents=False): 

244 """ 

245 Override delete from parent, invalidating cached photo variants 

246 """ 

247 photo_id = self.pk 

248 result = super().delete(using=using, keep_parents=keep_parents) 

249 if photo_id: 

250 invalidate_photo_cache(photo_id) 

251 return result 

252 

253 def perform_crop(self, request, dimensions, email, openid): 

254 """ 

255 Helper to crop the image 

256 """ 

257 if request.user.photo_set.count() == 1: 

258 # This is the first photo, assign to all confirmed addresses 

259 for addr in request.user.confirmedemail_set.all(): 

260 addr.photo = self 

261 addr.save() 

262 

263 for addr in request.user.confirmedopenid_set.all(): 

264 addr.photo = self 

265 addr.save() 

266 

267 if email: 

268 # Explicitly asked 

269 email.photo = self 

270 email.save() 

271 

272 if openid: 

273 # Explicitly asked 

274 openid.photo = self 

275 openid.save() 

276 

277 # Do the real work cropping 

278 img = Image.open(BytesIO(self.data)) 

279 

280 # This should be anyway checked during save... 

281 dimensions["a"], dimensions["b"] = img.size # pylint: disable=invalid-name 

282 if dimensions["a"] > MAX_PIXELS or dimensions["b"] > MAX_PIXELS: 

283 messages.error( 

284 request, 

285 _( 

286 "Image dimensions are too big (max: %(max_pixels)s x %(max_pixels)s" 

287 % { 

288 "max_pixels": MAX_PIXELS, 

289 } 

290 ), 

291 ) 

292 return HttpResponseRedirect(reverse_lazy("profile")) 

293 

294 if dimensions["w"] == 0 and dimensions["h"] == 0: 

295 dimensions["w"], dimensions["h"] = dimensions["a"], dimensions["b"] 

296 min_from_w_h = min(dimensions["w"], dimensions["h"]) 

297 dimensions["w"], dimensions["h"] = min_from_w_h, min_from_w_h 

298 elif ( 

299 (dimensions["w"] < 0) 

300 or ((dimensions["x"] + dimensions["w"]) > dimensions["a"]) 

301 or (dimensions["h"] < 0) 

302 or ((dimensions["y"] + dimensions["h"]) > dimensions["b"]) 

303 ): 

304 messages.error(request, _("Crop outside of original image bounding box")) 

305 return HttpResponseRedirect(reverse_lazy("profile")) 

306 

307 cropped = img.crop( 

308 ( 

309 dimensions["x"], 

310 dimensions["y"], 

311 dimensions["x"] + dimensions["w"], 

312 dimensions["y"] + dimensions["h"], 

313 ) 

314 ) 

315 # cropped.load() 

316 # Resize the image only if it's larger than the specified max width. 

317 cropped_w, cropped_h = cropped.size 

318 max_w = AVATAR_MAX_SIZE 

319 if cropped_w > max_w or cropped_h > max_w: 

320 cropped = cropped.resize((max_w, max_w), Image.LANCZOS) 

321 

322 data = BytesIO() 

323 cropped.save(data, pil_format(self.format), quality=JPEG_QUALITY) 

324 data.seek(0) 

325 

326 # Overwrite the existing image 

327 self.data = data.read() 

328 self.save() 

329 

330 return HttpResponseRedirect(reverse_lazy("profile")) 

331 

332 @property 

333 def access_count(self): 

334 """ 

335 Property to access access_count from the related stat object for backwards compatibility. 

336 """ 

337 try: 

338 return self.stat.access_count 

339 except ObjectDoesNotExist: 

340 return 0 

341 

342 def __str__(self): 

343 return "%s (%i) from %s" % (self.format, self.pk or 0, self.user) 

344 

345 

346# pylint: disable=too-few-public-methods 

347class ConfirmedEmailManager(models.Manager): 

348 """ 

349 Manager for our confirmed email addresses model 

350 """ 

351 

352 @staticmethod 

353 def create_confirmed_email(user, email_address, is_logged_in): 

354 """ 

355 Helper method to create confirmed email address 

356 """ 

357 confirmed = ConfirmedEmail() 

358 confirmed.user = user 

359 confirmed.ip_address = "0.0.0.0" 

360 confirmed.email = email_address 

361 confirmed.save() 

362 

363 external_photos = [] 

364 if is_logged_in: 

365 if gravatar := get_gravatar_photo(confirmed.email): 

366 external_photos.append(gravatar) 

367 

368 return (confirmed.pk, external_photos) 

369 

370 

371from django.db.models.functions import MD5, SHA256, Lower, Trim 

372 

373# ... (other code) 

374 

375class ConfirmedEmail(BaseAccountModel): 

376 """ 

377 Model holding our confirmed email addresses, as well as the relation 

378 to the assigned photo 

379 """ 

380 

381 email = models.EmailField(unique=True, max_length=MAX_LENGTH_EMAIL) 

382 photo = models.ForeignKey( 

383 Photo, 

384 related_name="emails", 

385 blank=True, 

386 null=True, 

387 on_delete=models.deletion.SET_NULL, 

388 ) 

389 # Alternative assignment - use Bluesky handle 

390 bluesky_handle = models.CharField(max_length=256, null=True, blank=True) 

391 digest = models.GeneratedField( 

392 expression=MD5(Lower(Trim("email"))), 

393 output_field=models.CharField(max_length=32), 

394 db_persist=True, 

395 ) 

396 digest_sha256 = models.GeneratedField( 

397 expression=SHA256(Lower(Trim("email"))), 

398 output_field=models.CharField(max_length=64), 

399 db_persist=True, 

400 ) 

401 objects = ConfirmedEmailManager() 

402 

403 class Meta: # pylint: disable=too-few-public-methods 

404 """ 

405 Class attributes 

406 """ 

407 

408 verbose_name = _("confirmed email") 

409 verbose_name_plural = _("confirmed emails") 

410 

411 indexes = [ 

412 models.Index(fields=["digest"], name="idx_cemail_digest"), 

413 models.Index(fields=["digest_sha256"], name="idx_cemail_digest_sha256"), 

414 models.Index(fields=["bluesky_handle"], name="idx_cemail_bluesky_handle"), 

415 ] 

416 

417 def set_photo(self, photo): 

418 """ 

419 Helper method to set photo 

420 """ 

421 self.photo = photo 

422 self.save() 

423 

424 def set_bluesky_handle(self, handle): 

425 """ 

426 Helper method to set Bluesky handle 

427 """ 

428 

429 bs = Bluesky() 

430 handle = bs.normalize_handle(handle) 

431 avatar = bs.get_profile(handle) 

432 if not avatar: 

433 raise ValueError("Invalid Bluesky handle") 

434 self.bluesky_handle = handle 

435 self.save() 

436 

437 def save( 

438 self, force_insert=False, force_update=False, using=None, update_fields=None 

439 ): 

440 """ 

441 Override save from parent, add digest 

442 """ 

443 # We need to manually expire the page caches 

444 # TODO: Verify this works as expected 

445 # First check if we already have an ID 

446 if self.pk: 

447 cache_url = reverse_lazy( 

448 "assign_photo_email", kwargs={"email_id": int(self.pk)} 

449 ) 

450 

451 cache_key = f"views.decorators.cache.cache_page.{quote(str(cache_url))}" 

452 try: 

453 if cache.has_key(cache_key): 

454 cache.delete(cache_key) 

455 logger.debug("Successfully cleaned up cached page: %s" % cache_key) 

456 except Exception as exc: 

457 logger.warning( 

458 "Failed to clean up cached page {}: {}".format(cache_key, exc) 

459 ) 

460 

461 # Invalidate Bluesky avatar URL cache if bluesky_handle changed 

462 if hasattr(self, "bluesky_handle") and self.bluesky_handle: 

463 try: 

464 cache.delete(self.bluesky_handle) 

465 logger.debug( 

466 "Successfully cleaned up Bluesky avatar cache for handle: %s" 

467 % self.bluesky_handle 

468 ) 

469 except Exception as exc: 

470 logger.warning( 

471 "Failed to clean up Bluesky avatar cache for handle %s: %s" 

472 % (self.bluesky_handle, exc) 

473 ) 

474 

475 return super().save( 

476 force_insert=force_insert, 

477 force_update=force_update, 

478 using=using, 

479 update_fields=update_fields, 

480 ) 

481 

482 @property 

483 def access_count(self): 

484 """ 

485 Property to access access_count from the related stat object for backwards compatibility. 

486 """ 

487 try: 

488 return self.stat.access_count 

489 except ObjectDoesNotExist: 

490 return 0 

491 

492 def __str__(self): 

493 return "%s (%i) from %s" % (self.email, self.pk, self.user) 

494 

495 

496class UnconfirmedEmail(BaseAccountModel): 

497 """ 

498 Model holding unconfirmed email addresses as well as the verification key 

499 """ 

500 

501 email = models.EmailField(max_length=MAX_LENGTH_EMAIL) 

502 verification_key = models.CharField(max_length=64) 

503 last_send_date = models.DateTimeField(null=True, blank=True) 

504 last_status = models.TextField(max_length=2047, null=True, blank=True) 

505 

506 class Meta: # pylint: disable=too-few-public-methods 

507 """ 

508 Class attributes 

509 """ 

510 

511 verbose_name = _("unconfirmed email") 

512 verbose_name_plural = _("unconfirmed emails") 

513 

514 def save( 

515 self, force_insert=False, force_update=False, using=None, update_fields=None 

516 ): 

517 if not self.verification_key: 

518 hash_object = hashlib.new("sha256") 

519 hash_object.update( 

520 urandom(1024) 

521 + self.user.username.encode("utf-8") # pylint: disable=no-member 

522 ) # pylint: disable=no-member 

523 self.verification_key = hash_object.hexdigest() 

524 super().save( 

525 force_insert=force_insert, 

526 force_update=force_update, 

527 using=using, 

528 update_fields=update_fields, 

529 ) 

530 

531 def send_confirmation_mail(self, url=SECURE_BASE_URL): 

532 """ 

533 Send confirmation mail to that mail address 

534 """ 

535 link = url + reverse( 

536 "confirm_email", kwargs={"verification_key": self.verification_key} 

537 ) 

538 email_subject = _("Confirm your email address on %s") % SITE_NAME 

539 email_body = render_to_string( 

540 "email_confirmation.txt", 

541 { 

542 "verification_link": link, 

543 "site_name": SITE_NAME, 

544 }, 

545 ) 

546 self.last_send_date = timezone.now() 

547 self.last_status = "OK" 

548 # if settings.DEBUG: 

549 # print('DEBUG: %s' % link) 

550 try: 

551 send_mail(email_subject, email_body, DEFAULT_FROM_EMAIL, [self.email]) 

552 except Exception as e: 

553 self.last_status = f"{e}" 

554 self.save() 

555 return True 

556 

557 def __str__(self): 

558 return "%s (%i) from %s" % (self.email, self.pk, self.user) 

559 

560 

561class UnconfirmedOpenId(BaseAccountModel): 

562 """ 

563 Model holding unconfirmed OpenIDs 

564 """ 

565 

566 openid = models.URLField(unique=False, max_length=MAX_LENGTH_URL) 

567 

568 class Meta: # pylint: disable=too-few-public-methods 

569 """ 

570 Meta class 

571 """ 

572 

573 verbose_name = _("unconfirmed OpenID") 

574 verbose_name_plural = "unconfirmed_OpenIDs" 

575 

576 def __str__(self): 

577 return "%s (%i) from %s" % (self.openid, self.pk, self.user) 

578 

579 

580class ConfirmedOpenId(BaseAccountModel): 

581 """ 

582 Model holding confirmed OpenIDs, as well as the relation to 

583 the assigned photo 

584 """ 

585 

586 openid = models.URLField(unique=True, max_length=MAX_LENGTH_URL) 

587 photo = models.ForeignKey( 

588 Photo, 

589 related_name="openids", 

590 blank=True, 

591 null=True, 

592 on_delete=models.deletion.SET_NULL, 

593 ) 

594 # http://<id>/ base version - http w/ trailing slash 

595 digest = models.CharField(max_length=64, db_index=True) 

596 # http://<id> - http w/o trailing slash 

597 alt_digest1 = models.CharField(max_length=64, null=True, blank=True, default=None, db_index=True) 

598 # https://<id>/ - https w/ trailing slash 

599 alt_digest2 = models.CharField(max_length=64, null=True, blank=True, default=None, db_index=True) 

600 # https://<id> - https w/o trailing slash 

601 alt_digest3 = models.CharField(max_length=64, null=True, blank=True, default=None, db_index=True) 

602 # Alternative assignment - use Bluesky handle 

603 bluesky_handle = models.CharField(max_length=256, null=True, blank=True) 

604 

605 class Meta: # pylint: disable=too-few-public-methods 

606 """ 

607 Meta class 

608 """ 

609 

610 verbose_name = _("confirmed OpenID") 

611 verbose_name_plural = _("confirmed OpenIDs") 

612 

613 def set_photo(self, photo): 

614 """ 

615 Helper method to save photo 

616 """ 

617 self.photo = photo 

618 self.save() 

619 

620 def set_bluesky_handle(self, handle): 

621 """ 

622 Helper method to set Bluesky handle 

623 """ 

624 bs = Bluesky() 

625 handle = bs.normalize_handle(handle) 

626 avatar = bs.get_profile(handle) 

627 if not avatar: 

628 raise ValueError("Invalid Bluesky handle") 

629 self.bluesky_handle = handle 

630 self.save() 

631 

632 def save( 

633 self, force_insert=False, force_update=False, using=None, update_fields=None 

634 ): 

635 url = urlsplit(self.openid) 

636 if url.username: # pragma: no cover 

637 password = url.password or "" 

638 netloc = f"{url.username}:{password}@{url.hostname}" 

639 else: 

640 netloc = url.hostname 

641 lowercase_url = urlunsplit( 

642 (url.scheme.lower(), netloc, url.path, url.query, url.fragment) 

643 ) 

644 self.openid = lowercase_url 

645 

646 self.digest = hashlib.sha256( 

647 openid_variations(lowercase_url)[0].encode("utf-8") 

648 ).hexdigest() 

649 self.alt_digest1 = hashlib.sha256( 

650 openid_variations(lowercase_url)[1].encode("utf-8") 

651 ).hexdigest() 

652 self.alt_digest2 = hashlib.sha256( 

653 openid_variations(lowercase_url)[2].encode("utf-8") 

654 ).hexdigest() 

655 self.alt_digest3 = hashlib.sha256( 

656 openid_variations(lowercase_url)[3].encode("utf-8") 

657 ).hexdigest() 

658 

659 # Invalidate page caches and Bluesky avatar cache 

660 if self.pk: 

661 # Invalidate assign_photo_openid page cache 

662 cache_url = reverse_lazy( 

663 "assign_photo_openid", kwargs={"openid_id": int(self.pk)} 

664 ) 

665 cache_key = f"views.decorators.cache.cache_page.{quote(str(cache_url))}" 

666 try: 

667 if cache.has_key(cache_key): 

668 cache.delete(cache_key) 

669 logger.debug("Successfully cleaned up cached page: %s" % cache_key) 

670 except Exception as exc: 

671 logger.warning( 

672 "Failed to clean up cached page {}: {}".format(cache_key, exc) 

673 ) 

674 

675 # Invalidate Bluesky avatar URL cache if bluesky_handle exists 

676 if hasattr(self, "bluesky_handle") and self.bluesky_handle: 

677 try: 

678 cache.delete(self.bluesky_handle) 

679 logger.debug( 

680 "Successfully cleaned up Bluesky avatar cache for handle: %s" 

681 % self.bluesky_handle 

682 ) 

683 except Exception as exc: 

684 logger.warning( 

685 "Failed to clean up Bluesky avatar cache for handle %s: %s" 

686 % (self.bluesky_handle, exc) 

687 ) 

688 

689 return super().save( 

690 force_insert=force_insert, 

691 force_update=force_update, 

692 using=using, 

693 update_fields=update_fields, 

694 ) 

695 

696 @property 

697 def access_count(self): 

698 """ 

699 Property to access access_count from the related stat object for backwards compatibility. 

700 """ 

701 try: 

702 return self.stat.access_count 

703 except ObjectDoesNotExist: 

704 return 0 

705 

706 def __str__(self): 

707 return "%s (%i) (%s)" % (self.openid, self.pk, self.user) 

708 

709 

710class PhotoAccessStat(AccessStat): 

711 """ 

712 Access statistics for Photo objects. 

713 """ 

714 

715 photo = models.OneToOneField( 

716 Photo, 

717 on_delete=models.CASCADE, 

718 primary_key=True, 

719 related_name="stat", 

720 ) 

721 

722 class Meta: # pylint: disable=too-few-public-methods 

723 """ 

724 Class attributes 

725 """ 

726 

727 verbose_name = _("photo access stat") 

728 verbose_name_plural = _("photo access stats") 

729 

730 

731@receiver(post_save, sender=Photo) 

732def create_photo_stat(sender, instance, created, **kwargs): 

733 """ 

734 Automatically create a PhotoAccessStat row when a new Photo is created. 

735 """ 

736 if created: 

737 PhotoAccessStat.objects.get_or_create(photo=instance) 

738 

739 

740def _get_email_digests(instance) -> List[str]: 

741 """Helper to extract MD5 and SHA256 digests for a ConfirmedEmail instance.""" 

742 digests = [] 

743 if getattr(instance, "digest", None): 

744 digests.append(instance.digest) 

745 if getattr(instance, "digest_sha256", None): 

746 digests.append(instance.digest_sha256) 

747 if not digests and getattr(instance, "email", None): 

748 normalized = instance.email.strip().lower().encode("utf-8") 

749 digests.append(hashlib.md5(normalized).hexdigest()) 

750 digests.append(hashlib.sha256(normalized).hexdigest()) 

751 return digests 

752 

753 

754def _get_openid_digests(instance) -> List[str]: 

755 """Helper to extract all OpenID SHA256 variation digests for a ConfirmedOpenId instance.""" 

756 digests = [] 

757 for attr in ("digest", "alt_digest1", "alt_digest2", "alt_digest3"): 

758 val = getattr(instance, attr, None) 

759 if val: 

760 digests.append(val) 

761 if not digests and getattr(instance, "openid", None): 

762 with contextlib.suppress(Exception): 

763 url = urlsplit(instance.openid) 

764 if url.username: 

765 password = url.password or "" 

766 netloc = f"{url.username}:{password}@{url.hostname}" 

767 else: 

768 netloc = url.hostname 

769 lowercase_url = urlunsplit( 

770 (url.scheme.lower(), netloc, url.path, url.query, url.fragment) 

771 ) 

772 for var in openid_variations(lowercase_url): 

773 digests.append(hashlib.sha256(var.encode("utf-8")).hexdigest()) 

774 return digests 

775 

776 

777@receiver(pre_delete, sender=User) 

778def invalidate_cache_on_user_delete(sender, instance, **kwargs): 

779 """ 

780 Invalidate all photo caches and digest response caches when a User is deleted. 

781 """ 

782 try: 

783 digests = [] 

784 for email in instance.confirmedemail_set.all(): 

785 digests.extend(_get_email_digests(email)) 

786 for openid in instance.confirmedopenid_set.all(): 

787 digests.extend(_get_openid_digests(openid)) 

788 for photo in instance.photo_set.all(): 

789 invalidate_photo_cache(photo.pk) 

790 if digests: 

791 invalidate_digest_cache(digests) 

792 except Exception as exc: # pylint: disable=broad-exception-caught 

793 logger.warning("Error invalidating cache on user delete: %s", exc) 

794 

795 

796@receiver(post_save, sender=Photo) 

797def invalidate_photo_cache_on_save(sender, instance, **kwargs): 

798 """ 

799 Invalidate cached resized variants when a photo is saved or updated, 

800 and pre-generate standard thumbnail sizes. 

801 """ 

802 if instance.pk: 

803 invalidate_photo_cache(instance.pk) 

804 # Also invalidate digest response cache for linked emails/openids 

805 try: 

806 digests = [] 

807 for email in instance.emails.all(): 

808 digests.extend(_get_email_digests(email)) 

809 for openid in instance.openids.all(): 

810 digests.extend(_get_openid_digests(openid)) 

811 if digests: 

812 invalidate_digest_cache(digests) 

813 except Exception: # pylint: disable=broad-exception-caught 

814 pass 

815 if instance.data and getattr(settings, "PREGENERATE_THUMBNAILS_ON_SAVE", True): 

816 pregenerate_photo_thumbnails(instance) 

817 

818 

819@receiver(pre_delete, sender=Photo) 

820def invalidate_photo_digests_on_pre_delete(sender, instance, **kwargs): 

821 """ 

822 Invalidate response cache for emails and openids associated with this photo before deletion. 

823 """ 

824 try: 

825 digests = [] 

826 for email in instance.emails.all(): 

827 digests.extend(_get_email_digests(email)) 

828 for openid in instance.openids.all(): 

829 digests.extend(_get_openid_digests(openid)) 

830 if digests: 

831 invalidate_digest_cache(digests) 

832 except Exception: # pylint: disable=broad-exception-caught 

833 pass 

834 

835 

836@receiver(post_delete, sender=Photo)