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

1""" 

2Our models for ivatar.ivataraccount 

3""" 

4 

5import base64 

6import hashlib 

7import time 

8from io import BytesIO 

9from os import urandom 

10from urllib.error import HTTPError, URLError 

11from ivatar.utils import urlopen, Bluesky 

12from urllib.parse import urlsplit, urlunsplit, quote 

13import logging 

14 

15from PIL import Image 

16from django.contrib.auth.models import User 

17from django.contrib import messages 

18from django.db import models 

19from django.utils import timezone 

20from django.http import HttpResponseRedirect 

21from django.urls import reverse_lazy, reverse 

22from django.utils.translation import gettext_lazy as _ 

23from django.core.cache import cache 

24from django.core.exceptions import ObjectDoesNotExist 

25from django.core.mail import send_mail 

26from django.template.loader import render_to_string 

27from openid.association import Association as OIDAssociation 

28from openid.store import nonce as oidnonce 

29from openid.store.interface import OpenIDStore 

30from django.db.models.signals import post_save 

31from django.dispatch import receiver 

32 

33from libravatar import libravatar_url 

34 

35from django.conf import settings 

36from ivatar.utils import openid_variations 

37from .gravatar import get_photo as get_gravatar_photo 

38 

39# Initialize logger 

40logger = logging.getLogger("ivatar") 

41 

42# Access settings through django.conf.settings 

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

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

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

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

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

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

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

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

51 

52 

53def file_format(image_type): 

54 """ 

55 Helper method returning a short image type 

56 """ 

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

58 return "jpg" 

59 elif image_type == "PNG": 

60 return "png" 

61 elif image_type == "GIF": 

62 return "gif" 

63 elif image_type == "WEBP": 

64 return "webp" 

65 return None 

66 

67 

68def pil_format(image_type): 

69 """ 

70 Helper method returning the 'encoder name' for PIL 

71 """ 

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

73 return "JPEG" 

74 elif image_type == "png": 

75 return "PNG" 

76 elif image_type == "gif": 

77 return "GIF" 

78 elif image_type == "webp": 

79 return "WEBP" 

80 

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

82 return None 

83 

84 

85class UserPreference(models.Model): 

86 """ 

87 Holds the user users preferences 

88 """ 

89 

90 THEMES = ( 

91 ("default", "Default theme"), 

92 ("clime", "climes theme"), 

93 ("green", "green theme"), 

94 ("red", "red theme"), 

95 ) 

96 

97 theme = models.CharField( 

98 max_length=10, 

99 choices=THEMES, 

100 default="default", 

101 ) 

102 

103 user = models.OneToOneField( 

104 User, 

105 on_delete=models.deletion.CASCADE, 

106 primary_key=True, 

107 ) 

108 

109 def __str__(self): 

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

111 

112 

113class BaseAccountModel(models.Model): 

114 """ 

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

116 """ 

117 

118 user = models.ForeignKey( 

119 User, 

120 on_delete=models.deletion.CASCADE, 

121 ) 

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

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

124 

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

126 """ 

127 Class attributes 

128 """ 

129 

130 abstract = True 

131 

132 

133class AccessStat(models.Model): 

134 """ 

135 Abstract base model for per-object access statistics. 

136 

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

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

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

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

141 """ 

142 

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

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

145 

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

147 """ 

148 Class attributes 

149 """ 

150 

151 abstract = True 

152 

153 

154class Photo(BaseAccountModel): 

155 """ 

156 Model holding the photos and information about them 

157 """ 

158 

159 ip_address = models.GenericIPAddressField(unpack_ipv4=True) 

160 data = models.BinaryField() 

161 format = models.CharField(max_length=4) 

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

163 """ 

164 Class attributes 

165 """ 

166 

167 verbose_name = _("photo") 

168 verbose_name_plural = _("photos") 

169 

170 def import_image(self, service_name, email_address): 

171 """ 

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

173 """ 

174 image_url = False 

175 

176 if service_name == "Gravatar": 

177 if gravatar := get_gravatar_photo(email_address): 

178 image_url = gravatar["image_url"] 

179 

180 if service_name == "Libravatar": 

181 image_url = libravatar_url(email_address, size=AVATAR_MAX_SIZE) 

182 

183 if not image_url: 

184 return False # pragma: no cover 

185 try: 

186 image = urlopen(image_url) 

187 except HTTPError as exc: 

188 logger.warning( 

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

190 ) 

191 return False 

192 except URLError as exc: 

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

194 return False 

195 data = image.read() 

196 

197 try: 

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

199 # How am I supposed to test this? 

200 except ValueError: # pragma: no cover 

201 return False # pragma: no cover 

202 

203 self.format = file_format(img.format) 

204 if not self.format: 

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

206 return False # pragma: no cover 

207 self.data = data 

208 super().save() 

209 return True 

210 

211 def save( 

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

213 ): 

214 """ 

215 Override save from parent, taking care about the image 

216 """ 

217 # Use PIL to read the file format 

218 try: 

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

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

221 # For debugging only 

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

223 return False 

224 self.format = file_format(img.format) 

225 if not self.format: 

226 logger.error("Format not recognized") 

227 return False 

228 return super().save( 

229 force_insert=force_insert, 

230 force_update=force_update, 

231 using=using, 

232 update_fields=update_fields, 

233 ) 

234 

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

236 """ 

237 Helper to crop the image 

238 """ 

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

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

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

242 addr.photo = self 

243 addr.save() 

244 

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

246 addr.photo = self 

247 addr.save() 

248 

249 if email: 

250 # Explicitly asked 

251 email.photo = self 

252 email.save() 

253 

254 if openid: 

255 # Explicitly asked 

256 openid.photo = self 

257 openid.save() 

258 

259 # Do the real work cropping 

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

261 

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

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

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

265 messages.error( 

266 request, 

267 _( 

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

269 % { 

270 "max_pixels": MAX_PIXELS, 

271 } 

272 ), 

273 ) 

274 return HttpResponseRedirect(reverse_lazy("profile")) 

275 

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

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

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

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

280 elif ( 

281 (dimensions["w"] < 0) 

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

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

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

285 ): 

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

287 return HttpResponseRedirect(reverse_lazy("profile")) 

288 

289 cropped = img.crop( 

290 ( 

291 dimensions["x"], 

292 dimensions["y"], 

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

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

295 ) 

296 ) 

297 # cropped.load() 

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

299 cropped_w, cropped_h = cropped.size 

300 max_w = AVATAR_MAX_SIZE 

301 if cropped_w > max_w or cropped_h > max_w: 

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

303 

304 data = BytesIO() 

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

306 data.seek(0) 

307 

308 # Overwrite the existing image 

309 self.data = data.read() 

310 self.save() 

311 

312 return HttpResponseRedirect(reverse_lazy("profile")) 

313 

314 @property 

315 def access_count(self): 

316 """ 

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

318 """ 

319 try: 

320 return self.stat.access_count 

321 except ObjectDoesNotExist: 

322 return 0 

323 

324 def __str__(self): 

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

326 

327 

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

329class ConfirmedEmailManager(models.Manager): 

330 """ 

331 Manager for our confirmed email addresses model 

332 """ 

333 

334 @staticmethod 

335 def create_confirmed_email(user, email_address, is_logged_in): 

336 """ 

337 Helper method to create confirmed email address 

338 """ 

339 confirmed = ConfirmedEmail() 

340 confirmed.user = user 

341 confirmed.ip_address = "0.0.0.0" 

342 confirmed.email = email_address 

343 confirmed.save() 

344 

345 external_photos = [] 

346 if is_logged_in: 

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

348 external_photos.append(gravatar) 

349 

350 return (confirmed.pk, external_photos) 

351 

352 

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

354 

355# ... (other code) 

356 

357class ConfirmedEmail(BaseAccountModel): 

358 """ 

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

360 to the assigned photo 

361 """ 

362 

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

364 photo = models.ForeignKey( 

365 Photo, 

366 related_name="emails", 

367 blank=True, 

368 null=True, 

369 on_delete=models.deletion.SET_NULL, 

370 ) 

371 # Alternative assignment - use Bluesky handle 

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

373 digest = models.GeneratedField( 

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

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

376 db_persist=True, 

377 ) 

378 digest_sha256 = models.GeneratedField( 

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

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

381 db_persist=True, 

382 ) 

383 objects = ConfirmedEmailManager() 

384 

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

386 """ 

387 Class attributes 

388 """ 

389 

390 verbose_name = _("confirmed email") 

391 verbose_name_plural = _("confirmed emails") 

392 

393 indexes = [ 

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

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

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

397 ] 

398 

399 def set_photo(self, photo): 

400 """ 

401 Helper method to set photo 

402 """ 

403 self.photo = photo 

404 self.save() 

405 

406 def set_bluesky_handle(self, handle): 

407 """ 

408 Helper method to set Bluesky handle 

409 """ 

410 

411 bs = Bluesky() 

412 handle = bs.normalize_handle(handle) 

413 avatar = bs.get_profile(handle) 

414 if not avatar: 

415 raise ValueError("Invalid Bluesky handle") 

416 self.bluesky_handle = handle 

417 self.save() 

418 

419 def save( 

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

421 ): 

422 """ 

423 Override save from parent, add digest 

424 """ 

425 # We need to manually expire the page caches 

426 # TODO: Verify this works as expected 

427 # First check if we already have an ID 

428 if self.pk: 

429 cache_url = reverse_lazy( 

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

431 ) 

432 

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

434 try: 

435 if cache.has_key(cache_key): 

436 cache.delete(cache_key) 

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

438 except Exception as exc: 

439 logger.warning( 

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

441 ) 

442 

443 # Invalidate Bluesky avatar URL cache if bluesky_handle changed 

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

445 try: 

446 cache.delete(self.bluesky_handle) 

447 logger.debug( 

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

449 % self.bluesky_handle 

450 ) 

451 except Exception as exc: 

452 logger.warning( 

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

454 % (self.bluesky_handle, exc) 

455 ) 

456 

457 return super().save( 

458 force_insert=force_insert, 

459 force_update=force_update, 

460 using=using, 

461 update_fields=update_fields, 

462 ) 

463 

464 @property 

465 def access_count(self): 

466 """ 

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

468 """ 

469 try: 

470 return self.stat.access_count 

471 except ObjectDoesNotExist: 

472 return 0 

473 

474 def __str__(self): 

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

476 

477 

478class UnconfirmedEmail(BaseAccountModel): 

479 """ 

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

481 """ 

482 

483 email = models.EmailField(max_length=MAX_LENGTH_EMAIL) 

484 verification_key = models.CharField(max_length=64) 

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

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

487 

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

489 """ 

490 Class attributes 

491 """ 

492 

493 verbose_name = _("unconfirmed email") 

494 verbose_name_plural = _("unconfirmed emails") 

495 

496 def save( 

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

498 ): 

499 if not self.verification_key: 

500 hash_object = hashlib.new("sha256") 

501 hash_object.update( 

502 urandom(1024) 

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

504 ) # pylint: disable=no-member 

505 self.verification_key = hash_object.hexdigest() 

506 super().save( 

507 force_insert=force_insert, 

508 force_update=force_update, 

509 using=using, 

510 update_fields=update_fields, 

511 ) 

512 

513 def send_confirmation_mail(self, url=SECURE_BASE_URL): 

514 """ 

515 Send confirmation mail to that mail address 

516 """ 

517 link = url + reverse( 

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

519 ) 

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

521 email_body = render_to_string( 

522 "email_confirmation.txt", 

523 { 

524 "verification_link": link, 

525 "site_name": SITE_NAME, 

526 }, 

527 ) 

528 self.last_send_date = timezone.now() 

529 self.last_status = "OK" 

530 # if settings.DEBUG: 

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

532 try: 

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

534 except Exception as e: 

535 self.last_status = f"{e}" 

536 self.save() 

537 return True 

538 

539 def __str__(self): 

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

541 

542 

543class UnconfirmedOpenId(BaseAccountModel): 

544 """ 

545 Model holding unconfirmed OpenIDs 

546 """ 

547 

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

549 

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

551 """ 

552 Meta class 

553 """ 

554 

555 verbose_name = _("unconfirmed OpenID") 

556 verbose_name_plural = "unconfirmed_OpenIDs" 

557 

558 def __str__(self): 

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

560 

561 

562class ConfirmedOpenId(BaseAccountModel): 

563 """ 

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

565 the assigned photo 

566 """ 

567 

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

569 photo = models.ForeignKey( 

570 Photo, 

571 related_name="openids", 

572 blank=True, 

573 null=True, 

574 on_delete=models.deletion.SET_NULL, 

575 ) 

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

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

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

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

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

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

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

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

584 # Alternative assignment - use Bluesky handle 

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

586 

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

588 """ 

589 Meta class 

590 """ 

591 

592 verbose_name = _("confirmed OpenID") 

593 verbose_name_plural = _("confirmed OpenIDs") 

594 

595 def set_photo(self, photo): 

596 """ 

597 Helper method to save photo 

598 """ 

599 self.photo = photo 

600 self.save() 

601 

602 def set_bluesky_handle(self, handle): 

603 """ 

604 Helper method to set Bluesky handle 

605 """ 

606 bs = Bluesky() 

607 handle = bs.normalize_handle(handle) 

608 avatar = bs.get_profile(handle) 

609 if not avatar: 

610 raise ValueError("Invalid Bluesky handle") 

611 self.bluesky_handle = handle 

612 self.save() 

613 

614 def save( 

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

616 ): 

617 url = urlsplit(self.openid) 

618 if url.username: # pragma: no cover 

619 password = url.password or "" 

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

621 else: 

622 netloc = url.hostname 

623 lowercase_url = urlunsplit( 

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

625 ) 

626 self.openid = lowercase_url 

627 

628 self.digest = hashlib.sha256( 

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

630 ).hexdigest() 

631 self.alt_digest1 = hashlib.sha256( 

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

633 ).hexdigest() 

634 self.alt_digest2 = hashlib.sha256( 

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

636 ).hexdigest() 

637 self.alt_digest3 = hashlib.sha256( 

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

639 ).hexdigest() 

640 

641 # Invalidate page caches and Bluesky avatar cache 

642 if self.pk: 

643 # Invalidate assign_photo_openid page cache 

644 cache_url = reverse_lazy( 

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

646 ) 

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

648 try: 

649 if cache.has_key(cache_key): 

650 cache.delete(cache_key) 

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

652 except Exception as exc: 

653 logger.warning( 

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

655 ) 

656 

657 # Invalidate Bluesky avatar URL cache if bluesky_handle exists 

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

659 try: 

660 cache.delete(self.bluesky_handle) 

661 logger.debug( 

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

663 % self.bluesky_handle 

664 ) 

665 except Exception as exc: 

666 logger.warning( 

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

668 % (self.bluesky_handle, exc) 

669 ) 

670 

671 return super().save( 

672 force_insert=force_insert, 

673 force_update=force_update, 

674 using=using, 

675 update_fields=update_fields, 

676 ) 

677 

678 @property 

679 def access_count(self): 

680 """ 

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

682 """ 

683 try: 

684 return self.stat.access_count 

685 except ObjectDoesNotExist: 

686 return 0 

687 

688 def __str__(self): 

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

690 

691 

692class PhotoAccessStat(AccessStat): 

693 """ 

694 Access statistics for Photo objects. 

695 """ 

696 

697 photo = models.OneToOneField( 

698 Photo, 

699 on_delete=models.CASCADE, 

700 primary_key=True, 

701 related_name="stat", 

702 ) 

703 

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

705 """ 

706 Class attributes 

707 """ 

708 

709 verbose_name = _("photo access stat") 

710 verbose_name_plural = _("photo access stats") 

711 

712 

713@receiver(post_save, sender=Photo) 

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

715 """ 

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

717 """ 

718 if created: 

719 PhotoAccessStat.objects.get_or_create(photo=instance) 

720 

721 

722class ConfirmedEmailAccessStat(AccessStat): 

723 """ 

724 Access statistics for ConfirmedEmail objects. 

725 """ 

726 

727 email = models.OneToOneField( 

728 ConfirmedEmail, 

729 on_delete=models.CASCADE, 

730 primary_key=True, 

731 related_name="stat", 

732 ) 

733 

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

735 """ 

736 Class attributes 

737 """ 

738 

739 verbose_name = _("confirmed email access stat") 

740 verbose_name_plural = _("confirmed email access stats") 

741 

742 

743@receiver(post_save, sender=ConfirmedEmail) 

744def create_confirmed_email_stat(sender, instance, created, **kwargs): 

745 """ 

746 Automatically create a ConfirmedEmailAccessStat row when a new ConfirmedEmail is created. 

747 """ 

748 if created: 

749 ConfirmedEmailAccessStat.objects.get_or_create(email=instance) 

750 

751 

752class ConfirmedOpenIdAccessStat(AccessStat): 

753 """ 

754 Access statistics for ConfirmedOpenId objects. 

755 """ 

756 

757 openid = models.OneToOneField( 

758 ConfirmedOpenId, 

759 on_delete=models.CASCADE, 

760 primary_key=True, 

761 related_name="stat", 

762 ) 

763 

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

765 """ 

766 Class attributes 

767 """ 

768 

769 verbose_name = _("confirmed openid access stat") 

770 verbose_name_plural = _("confirmed openid access stats") 

771 

772 

773@receiver(post_save, sender=ConfirmedOpenId) 

774def create_confirmed_openid_stat(sender, instance, created, **kwargs): 

775 """ 

776 Automatically create a ConfirmedOpenIdAccessStat row when a new ConfirmedOpenId is created. 

777 """ 

778 if created: 

779 ConfirmedOpenIdAccessStat.objects.get_or_create(openid=instance) 

780 

781 

782class OpenIDNonce(models.Model): 

783 """ 

784 Model holding OpenID Nonces 

785 See also: https://github.com/edx/django-openid-auth/ 

786 """ 

787 

788 server_url = models.CharField(max_length=255) 

789 timestamp = models.IntegerField() 

790 salt = models.CharField(max_length=128) 

791 

792 def __str__(self): 

793 return "%s (%i) (timestamp: %i)" % (self.server_url, self.pk, self.timestamp) 

794 

795 

796class OpenIDAssociation(models.Model): 

797 """ 

798 Model holding the relation/association about OpenIDs 

799 """ 

800 

801 server_url = models.TextField(max_length=2047) 

802 handle = models.CharField(max_length=255) 

803 secret = models.TextField(max_length=255) # stored base64 encoded 

804 issued = models.IntegerField() 

805 lifetime = models.IntegerField() 

806 assoc_type = models.TextField(max_length=64) 

807 

808 def __str__(self): 

809 return "%s (%i) (%s, lifetime: %i)" % ( 

810 self.server_url, 

811 self.pk, 

812 self.assoc_type, 

813 self.lifetime, 

814 ) 

815 

816 

817class DjangoOpenIDStore(OpenIDStore): 

818 """ 

819 The Python openid library needs an OpenIDStore subclass to persist data 

820 related to OpenID authentications. This one uses our Django models. 

821 """ 

822 

823 @staticmethod 

824 def storeAssociation(server_url, association): # pragma: no cover 

825 """ 

826 Helper method to store associations 

827 """ 

828 assoc = OpenIDAssociation( 

829 server_url=server_url, 

830 handle=association.handle, 

831 secret=base64.encodebytes(association.secret), 

832 issued=association.issued, 

833 lifetime=association.issued, 

834 assoc_type=association.assoc_type, 

835 ) 

836 assoc.save() 

837 

838 def getAssociation(self, server_url, handle=None): # pragma: no cover 

839 """ 

840 Helper method to get associations 

841 """ 

842 assocs = [] 

843 if handle is not None: 

844 assocs = OpenIDAssociation.objects.filter( # pylint: disable=no-member 

845 server_url=server_url, handle=handle 

846 ) 

847 else: 

848 assocs = OpenIDAssociation.objects.filter( # pylint: disable=no-member 

849 server_url=server_url 

850 ) 

851 if not assocs: 

852 return None 

853 associations = [] 

854 for assoc in assocs: 

855 if isinstance(assoc.secret, str): 

856 assoc.secret = assoc.secret.split("b'")[1].split("'")[0] 

857 assoc.secret = bytes(assoc.secret, "utf-8") 

858 association = OIDAssociation( 

859 assoc.handle, 

860 base64.decodebytes(assoc.secret), 

861 assoc.issued,