Coverage for ivatar/ivataraccount/views.py: 77%
699 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 11:51 +0000
1"""
2View classes for ivatar/ivataraccount/
3"""
5from io import BytesIO
6from ivatar.utils import urlopen, Bluesky
7import base64
8import binascii
9import contextlib
10from xml.sax import saxutils
11import gzip
12import logging
14from PIL import Image
16from django.db.models import ProtectedError
17from django.core.exceptions import ObjectDoesNotExist
18from django.contrib.auth.decorators import login_required
19from django.contrib.auth.models import User
20from django.utils.decorators import method_decorator
21from django.contrib.messages.views import SuccessMessageMixin
22from django.contrib import messages
23from django.views.generic.edit import FormView, UpdateView
24from django.views.generic.base import View, TemplateView
25from django.views.generic.detail import DetailView
26from django.contrib.auth import authenticate, login
27from django.contrib.auth.forms import UserCreationForm, SetPasswordForm
28from django.contrib.auth.views import LoginView
29from django.contrib.auth.views import (
30 PasswordResetView as PasswordResetViewOriginal,
31)
32from django.utils.crypto import get_random_string
33from django.utils.translation import gettext_lazy as _
34from django.http import HttpResponseRedirect, HttpResponse
35from django.urls import reverse_lazy, reverse
36from django.shortcuts import render
37from django_openid_auth.models import UserOpenID
39from openid import oidutil
40from openid.consumer import consumer
42from ipware import get_client_ip
44from email_validator import validate_email
46from libravatar import libravatar_url
47from django.conf import settings
48from .gravatar import get_photo as get_gravatar_photo
50# Access settings through django.conf.settings
51MAX_NUM_PHOTOS = getattr(settings, "MAX_NUM_PHOTOS", 5)
52MAX_PHOTO_SIZE = getattr(settings, "MAX_PHOTO_SIZE", 10485760)
53JPEG_QUALITY = getattr(settings, "JPEG_QUALITY", 85)
54AVATAR_MAX_SIZE = getattr(settings, "AVATAR_MAX_SIZE", 512)
55SOCIAL_AUTH_FEDORA_KEY = getattr(settings, "SOCIAL_AUTH_FEDORA_KEY", None)
56SITE_NAME = getattr(settings, "SITE_NAME", "libravatar")
57IVATAR_VERSION = getattr(settings, "IVATAR_VERSION", "2.0")
58MAX_NUM_UNCONFIRMED_EMAILS = getattr(settings, "MAX_NUM_UNCONFIRMED_EMAILS", 5)
59SECURE_BASE_URL = getattr(settings, "SECURE_BASE_URL", "https://avatars.linux-kernel.at/avatar/")
60BASE_URL = getattr(settings, "BASE_URL", "http://avatars.linux-kernel.at/avatar/")
61IVATAR_MAILGUN_API_KEY = getattr(settings, "IVATAR_MAILGUN_API_KEY", None)
62IVATAR_MAILGUN_SENDER_DOMAIN = getattr(settings, "IVATAR_MAILGUN_SENDER_DOMAIN", None)
64from .forms import AddEmailForm, UploadPhotoForm, AddOpenIDForm
65from .forms import UpdatePreferenceForm, UploadLibravatarExportForm
66from .forms import DeleteAccountForm
67from .models import UnconfirmedEmail, ConfirmedEmail, Photo
68from .models import UnconfirmedOpenId, ConfirmedOpenId, DjangoOpenIDStore
69from .models import UserPreference
70from .models import file_format
71from .read_libravatar_export import read_gzdata as libravatar_read_gzdata
73# Initialize loggers
74logger = logging.getLogger("ivatar")
75security_logger = logging.getLogger("ivatar.security")
77# Import OpenTelemetry with graceful degradation
78from ..telemetry_utils import (
79 trace_file_upload,
80 trace_authentication,
81 get_telemetry_metrics,
82)
84avatar_metrics = get_telemetry_metrics()
87def openid_logging(message, level=0):
88 """
89 Helper method for openid logging
90 """
91 # Normal messages are not that important
92 # No need for coverage here
93 if level > 0: # pragma: no cover
94 logger.debug(message)
97class CreateView(SuccessMessageMixin, FormView):
98 """
99 View class for creating a new user
100 """
102 template_name = "new.html"
103 form_class = UserCreationForm
105 @trace_authentication("user_registration")
106 def form_valid(self, form):
107 form.save()
108 user = authenticate(
109 username=form.cleaned_data["username"],
110 password=form.cleaned_data["password1"],
111 )
112 if user is not None:
113 # If the username looks like a mail address, automagically
114 # add it as unconfirmed mail and set it also as user's
115 # email address
116 with contextlib.suppress(Exception):
117 self._extracted_from_form_valid_(form, user)
118 login(self.request, user)
119 pref = UserPreference.objects.create(
120 user_id=user.pk
121 ) # pylint: disable=no-member
122 pref.save()
123 return HttpResponseRedirect(reverse_lazy("profile"))
124 return HttpResponseRedirect(reverse_lazy("login")) # pragma: no cover
126 def _extracted_from_form_valid_(self, form, user):
127 # This will error out if it's not a valid address
128 valid = validate_email(form.cleaned_data["username"])
129 user.email = valid.email
130 user.save()
131 # The following will also error out if it already exists
132 unconfirmed = UnconfirmedEmail()
133 unconfirmed.email = valid.email
134 unconfirmed.user = user
135 unconfirmed.save()
136 unconfirmed.send_confirmation_mail(
137 url=self.request.build_absolute_uri("/")[:-1]
138 )
140 def get(self, request, *args, **kwargs):
141 """
142 Handle get for create view
143 """
144 if request.user and request.user.is_authenticated:
145 return HttpResponseRedirect(reverse_lazy("profile"))
146 return super().get(self, request, args, kwargs)
149@method_decorator(login_required, name="dispatch")
150class PasswordSetView(SuccessMessageMixin, FormView):
151 """
152 View class for changing the password
153 """
155 template_name = "password_change.html"
156 form_class = SetPasswordForm
157 success_message = _("password changed successfully - please login again")
158 success_url = reverse_lazy("profile")
160 def get_form_kwargs(self):
161 kwargs = super().get_form_kwargs()
162 kwargs["user"] = self.request.user
163 return kwargs
165 def form_valid(self, form):
166 form.save()
167 super().form_valid(form)
168 return HttpResponseRedirect(reverse_lazy("login"))
171@method_decorator(login_required, name="dispatch")
172class AddEmailView(SuccessMessageMixin, FormView):
173 """
174 View class for adding email addresses
175 """
177 template_name = "add_email.html"
178 form_class = AddEmailForm
179 success_url = reverse_lazy("profile")
181 def form_valid(self, form):
182 if not form.save(self.request):
183 return render(self.request, self.template_name, {"form": form})
185 messages.success(self.request, _("Address added successfully"))
186 return super().form_valid(form)
189@method_decorator(login_required, name="dispatch")
190class RemoveUnconfirmedEmailView(SuccessMessageMixin, View):
191 """
192 View class for removing a unconfirmed email address
193 """
195 @staticmethod
196 def post(request, *args, **kwargs): # pylint: disable=unused-argument
197 """
198 Handle post request - removing unconfirmed email
199 """
200 try:
201 email = UnconfirmedEmail.objects.get( # pylint: disable=no-member
202 user=request.user, id=kwargs["email_id"]
203 )
204 email.delete()
205 messages.success(request, _("Address removed"))
206 except UnconfirmedEmail.DoesNotExist: # pylint: disable=no-member
207 messages.error(request, _("Address does not exist"))
208 return HttpResponseRedirect(reverse_lazy("profile"))
211class ConfirmEmailView(SuccessMessageMixin, TemplateView):
212 """
213 View class for confirming an unconfirmed email address
214 """
216 template_name = "email_confirmed.html"
218 def get(self, request, *args, **kwargs):
219 # be tolerant of extra crap added by mail clients
220 key = kwargs["verification_key"].replace(" ", "")
222 if len(key) != 64:
223 messages.error(request, _("Verification key incorrect"))
224 return HttpResponseRedirect(reverse_lazy("profile"))
226 try:
227 unconfirmed = UnconfirmedEmail.objects.get(
228 verification_key=key
229 ) # pylint: disable=no-member
230 except UnconfirmedEmail.DoesNotExist: # pylint: disable=no-member
231 messages.error(request, _("Verification key does not exist"))
232 return HttpResponseRedirect(reverse_lazy("profile"))
234 if ConfirmedEmail.objects.filter(email=unconfirmed.email).count() > 0:
235 messages.error(
236 request,
237 _("This mail address has been taken already and cannot be confirmed"),
238 )
239 return HttpResponseRedirect(reverse_lazy("profile"))
241 # TODO: Check for a reasonable expiration time in unconfirmed email
243 (confirmed_id, external_photos) = ConfirmedEmail.objects.create_confirmed_email(
244 unconfirmed.user, unconfirmed.email, not request.user.is_anonymous
245 )
247 unconfirmed.delete()
249 # if there's a single image in this user's profile,
250 # assign it to the new email
251 confirmed = ConfirmedEmail.objects.get(id=confirmed_id)
252 if confirmed.user.photo_set.count() == 1:
253 confirmed.set_photo(confirmed.user.photo_set.first())
254 kwargs["photos"] = external_photos
255 kwargs["email_id"] = confirmed_id
256 return super().get(request, *args, **kwargs)
259@method_decorator(login_required, name="dispatch")
260class RemoveConfirmedEmailView(SuccessMessageMixin, View):
261 """
262 View class for removing a confirmed email address
263 """
265 @staticmethod
266 def post(request, *args, **kwargs): # pylint: disable=unused-argument
267 """
268 Handle post request - removing confirmed email
269 """
270 try:
271 email = ConfirmedEmail.objects.get(user=request.user, id=kwargs["email_id"])
272 email.delete()
273 messages.success(request, _("Address removed"))
274 except ConfirmedEmail.DoesNotExist: # pylint: disable=no-member
275 messages.error(request, _("Address does not exist"))
276 return HttpResponseRedirect(reverse_lazy("profile"))
279@method_decorator(login_required, name="dispatch")
280class AssignPhotoEmailView(SuccessMessageMixin, TemplateView):
281 """
282 View class for assigning a photo to an email address
283 """
285 model = Photo
286 template_name = "assign_photo_email.html"
288 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
289 """
290 Handle post request - assign photo to email
291 """
292 photo = None
294 try:
295 email = ConfirmedEmail.objects.get(user=request.user, id=kwargs["email_id"])
296 except ConfirmedEmail.DoesNotExist: # pylint: disable=no-member
297 messages.error(request, _("Invalid request"))
298 return HttpResponseRedirect(reverse_lazy("profile"))
300 if "photoNone" in request.POST:
301 email.photo = None
302 email.bluesky_handle = None
303 elif "photoBluesky" in request.POST:
304 # Keep the existing Bluesky handle, clear the photo
305 email.photo = None
306 # Don't clear bluesky_handle - keep it as is
307 else:
308 if "photo_id" not in request.POST:
309 messages.error(request, _("Invalid request [photo_id] missing"))
310 return HttpResponseRedirect(reverse_lazy("profile"))
312 if request.POST["photo_id"] == "bluesky":
313 # Handle Bluesky photo selection
314 email.photo = None
315 # Don't clear bluesky_handle - keep it as is
316 else:
317 try:
318 photo = self.model.objects.get( # pylint: disable=no-member
319 id=request.POST["photo_id"], user=request.user
320 )
321 except self.model.DoesNotExist: # pylint: disable=no-member
322 messages.error(request, _("Photo does not exist"))
323 return HttpResponseRedirect(reverse_lazy("profile"))
324 email.photo = photo
325 email.bluesky_handle = None
326 email.save()
328 messages.success(request, _("Successfully changed photo"))
329 return HttpResponseRedirect(reverse_lazy("profile"))
331 def get_context_data(self, **kwargs):
332 data = super().get_context_data(**kwargs)
333 data["email"] = ConfirmedEmail.objects.get(pk=kwargs["email_id"])
334 return data
337@method_decorator(login_required, name="dispatch")
338class AssignPhotoOpenIDView(SuccessMessageMixin, TemplateView):
339 """
340 View class for assigning a photo to an openid address
341 """
343 model = Photo
344 template_name = "assign_photo_openid.html"
346 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
347 """
348 Handle post - assign photo to openid
349 """
350 photo = None
352 try:
353 openid = ConfirmedOpenId.objects.get( # pylint: disable=no-member
354 user=request.user, id=kwargs["openid_id"]
355 )
356 except ConfirmedOpenId.DoesNotExist: # pylint: disable=no-member
357 messages.error(request, _("Invalid request"))
358 return HttpResponseRedirect(reverse_lazy("profile"))
360 if "photoNone" in request.POST:
361 openid.photo = None
362 else:
363 if "photo_id" not in request.POST:
364 messages.error(request, _("Invalid request [photo_id] missing"))
365 return HttpResponseRedirect(reverse_lazy("profile"))
367 try:
368 photo = self.model.objects.get( # pylint: disable=no-member
369 id=request.POST["photo_id"], user=request.user
370 )
371 except self.model.DoesNotExist: # pylint: disable=no-member
372 messages.error(request, _("Photo does not exist"))
373 return HttpResponseRedirect(reverse_lazy("profile"))
374 openid.photo = photo
375 openid.bluesky_handle = None
376 openid.save()
378 messages.success(request, _("Successfully changed photo"))
379 return HttpResponseRedirect(reverse_lazy("profile"))
381 def get_context_data(self, **kwargs):
382 data = super().get_context_data(**kwargs)
383 data["openid"] = ConfirmedOpenId.objects.get(
384 pk=kwargs["openid_id"]
385 ) # pylint: disable=no-member
386 return data
389@method_decorator(login_required, name="dispatch")
390class AssignBlueskyHandleToEmailView(SuccessMessageMixin, TemplateView):
391 """
392 View class for assigning a Bluesky handle to an email address
393 """
395 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
396 """
397 Handle post request - assign bluesky handle to email
398 """
400 try:
401 email = ConfirmedEmail.objects.get(user=request.user, id=kwargs["email_id"])
402 except ConfirmedEmail.DoesNotExist: # pylint: disable=no-member
403 messages.error(request, _("Invalid request"))
404 return HttpResponseRedirect(reverse_lazy("profile"))
406 if "bluesky_handle" not in request.POST:
407 messages.error(request, _("Invalid request [bluesky_handle] missing"))
408 return HttpResponseRedirect(reverse_lazy("profile"))
409 bluesky_handle = request.POST["bluesky_handle"]
411 try:
412 bs = Bluesky()
414 bs.get_avatar(bluesky_handle)
415 except Exception as e:
416 messages.error(request, _(f"Handle '{bluesky_handle}' not found: {e}"))
417 return HttpResponseRedirect(
418 reverse_lazy(
419 "assign_photo_email", kwargs={"email_id": int(kwargs["email_id"])}
420 )
421 )
422 try:
423 email.set_bluesky_handle(bluesky_handle)
424 except Exception as e:
425 messages.error(request, _(f"Error: {e}"))
426 return HttpResponseRedirect(
427 reverse_lazy(
428 "assign_photo_email", kwargs={"email_id": int(kwargs["email_id"])}
429 )
430 )
431 email.photo = None
432 email.save()
434 messages.success(request, _("Successfully assigned Bluesky handle"))
435 return HttpResponseRedirect(reverse_lazy("profile"))
437 def get_context_data(self, **kwargs):
438 data = super().get_context_data(**kwargs)
439 data["email"] = ConfirmedEmail.objects.get(pk=kwargs["email_id"])
440 return data
443@method_decorator(login_required, name="dispatch")
444class AssignBlueskyHandleToOpenIdView(SuccessMessageMixin, TemplateView):
445 """
446 View class for assigning a Bluesky handle to an email address
447 """
449 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
450 """
451 Handle post request - assign bluesky handle to email
452 """
454 try:
455 openid = ConfirmedOpenId.objects.get(
456 user=request.user, id=kwargs["open_id"]
457 )
458 except ConfirmedOpenId.DoesNotExist: # pylint: disable=no-member
459 messages.error(request, _("Invalid request"))
460 return HttpResponseRedirect(reverse_lazy("profile"))
462 if "bluesky_handle" not in request.POST:
463 messages.error(request, _("Invalid request [bluesky_handle] missing"))
464 return HttpResponseRedirect(reverse_lazy("profile"))
465 bluesky_handle = request.POST["bluesky_handle"]
467 try:
468 bs = Bluesky()
470 bs.get_avatar(bluesky_handle)
471 except Exception as e:
472 messages.error(request, _(f"Handle '{bluesky_handle}' not found: {e}"))
473 return HttpResponseRedirect(
474 reverse_lazy(
475 "assign_photo_openid", kwargs={"openid_id": int(kwargs["open_id"])}
476 )
477 )
478 try:
479 openid.set_bluesky_handle(bluesky_handle)
480 except Exception as e:
481 messages.error(request, _(f"Error: {e}"))
482 return HttpResponseRedirect(
483 reverse_lazy(
484 "assign_photo_openid", kwargs={"openid_id": int(kwargs["open_id"])}
485 )
486 )
487 openid.photo = None
488 openid.save()
490 messages.success(request, _("Successfully assigned Bluesky handle"))
491 return HttpResponseRedirect(reverse_lazy("profile"))
493 def get_context_data(self, **kwargs):
494 data = super().get_context_data(**kwargs)
495 data["openid"] = ConfirmedOpenId.objects.get(pk=kwargs["open_id"])
496 return data
499@method_decorator(login_required, name="dispatch")
500class ImportPhotoView(SuccessMessageMixin, TemplateView):
501 """
502 View class to import a photo from another service
503 Currently only Gravatar is supported
504 """
506 template_name = "import_photo.html"
508 def get_context_data(self, **kwargs):
509 context = super().get_context_data(**kwargs)
510 context["photos"] = []
511 addr = None
512 if "email_id" in kwargs:
513 try:
514 addr = ConfirmedEmail.objects.get(pk=kwargs["email_id"]).email
515 except ConfirmedEmail.ObjectDoesNotExist: # pylint: disable=no-member
516 messages.error(self.request, _("Address does not exist"))
517 return context
519 if addr := kwargs.get("email_addr", None):
520 if gravatar := get_gravatar_photo(addr):
521 context["photos"].append(gravatar)
523 if libravatar_service_url := libravatar_url(
524 email=addr,
525 default=404,
526 size=AVATAR_MAX_SIZE,
527 ):
528 try:
529 urlopen(libravatar_service_url)
530 except OSError as exc:
531 logger.warning(f"Exception caught during photo import: {exc}")
532 else:
533 context["photos"].append(
534 {
535 "service_url": libravatar_service_url,
536 "thumbnail_url": f"{libravatar_service_url}&s=80",
537 "image_url": f"{libravatar_service_url}&s=512",
538 "width": 80,
539 "height": 80,
540 "service_name": "Libravatar",
541 }
542 )
544 return context
546 def post(
547 self, request, *args, **kwargs
548 ): # pylint: disable=no-self-use,unused-argument,too-many-branches,line-too-long
549 """
550 Handle post to photo import
551 """
553 imported = None
555 email_id = kwargs.get("email_id", request.POST.get("email_id", None))
556 addr = kwargs.get("email", request.POST.get("email_addr", None))
558 if email_id:
559 email = ConfirmedEmail.objects.filter(id=email_id, user=request.user)
560 if email.exists():
561 addr = email.first().email
562 else:
563 messages.error(request, _("Address does not exist"))
564 return HttpResponseRedirect(reverse_lazy("profile"))
566 if "photo_Gravatar" in request.POST:
567 photo = Photo()
568 photo.user = request.user
569 photo.ip_address = get_client_ip(request)[0]
570 if photo.import_image("Gravatar", addr):
571 messages.success(request, _("Gravatar image successfully imported"))
572 else:
573 # Honestly, I'm not sure how to test this...
574 messages.error(
575 request, _("Gravatar image import not successful")
576 ) # pragma: no cover
577 imported = True
579 if "photo_Libravatar" in request.POST:
580 photo = Photo()
581 photo.user = request.user
582 photo.ip_address = get_client_ip(request)[0]
583 if photo.import_image("Libravatar", addr):
584 messages.success(request, _("Libravatar image successfully imported"))
585 else:
586 # Honestly, I'm not sure how to test this...
587 messages.error(
588 request, _("Libravatar image import not successful")
589 ) # pragma: no cover
590 imported = True
591 if not imported:
592 messages.warning(request, _("Nothing importable"))
593 return HttpResponseRedirect(reverse_lazy("profile"))
596@method_decorator(login_required, name="dispatch")
597class RawImageView(DetailView):
598 """
599 View to return (binary) raw image data, for use in <img/>-tags
600 """
602 model = Photo
604 def get(self, request, *args, **kwargs):
605 photo = self.model.objects.get(pk=kwargs["pk"]) # pylint: disable=no-member
606 if photo.user.id != request.user.id and not request.user.is_staff:
607 return HttpResponseRedirect(reverse_lazy("home"))
608 return HttpResponse(BytesIO(photo.data), content_type=f"image/{photo.format}")
611@method_decorator(login_required, name="dispatch")
612class DeletePhotoView(SuccessMessageMixin, View):
613 """
614 View class for deleting a photo
615 """
617 model = Photo
619 def get(self, request, *args, **kwargs): # pylint: disable=unused-argument
620 """
621 Handle get - delete photo
622 """
623 try:
624 photo = self.model.objects.get( # pylint: disable=no-member
625 pk=kwargs["pk"], user=request.user
626 )
627 photo.delete()
628 except (self.model.DoesNotExist, ProtectedError): # pylint: disable=no-member
629 messages.error(request, _("No such image or no permission to delete it"))
630 return HttpResponseRedirect(reverse_lazy("profile"))
631 messages.success(request, _("Photo deleted successfully"))
632 return HttpResponseRedirect(reverse_lazy("profile"))
635@method_decorator(login_required, name="dispatch")
636class UploadPhotoView(SuccessMessageMixin, FormView):
637 """
638 View class responsible for photo upload with enhanced security
639 """
641 model = Photo
642 template_name = "upload_photo.html"
643 form_class = UploadPhotoForm
644 success_message = _("Successfully uploaded")
645 success_url = reverse_lazy("profile")
647 def post(self, request, *args, **kwargs):
648 # Check maximum number of photos
649 num_photos = request.user.photo_set.count()
650 if num_photos >= MAX_NUM_PHOTOS:
651 messages.error(
652 request, _("Maximum number of photos (%i) reached" % MAX_NUM_PHOTOS)
653 )
654 return HttpResponseRedirect(reverse_lazy("profile"))
656 return super().post(request, *args, **kwargs)
658 @trace_file_upload("photo_upload")
659 def form_valid(self, form):
660 photo_data = self.request.FILES["photo"]
662 # Additional size check (redundant but good for security)
663 if photo_data.size > MAX_PHOTO_SIZE:
664 messages.error(self.request, _("Image too big"))
665 avatar_metrics.record_file_upload(
666 file_size=photo_data.size,
667 content_type=photo_data.content_type,
668 success=False,
669 )
670 return HttpResponseRedirect(reverse_lazy("profile"))
672 # Enhanced security logging
673 security_logger.info(
674 f"Photo upload attempt by user {self.request.user.id} "
675 f"from IP {get_client_ip(self.request)[0]}, "
676 f"file size: {photo_data.size} bytes"
677 )
679 photo = form.save(self.request, photo_data)
681 if not photo:
682 security_logger.warning(
683 f"Photo upload failed for user {self.request.user.id} - invalid format"
684 )
685 messages.error(self.request, _("Invalid Format"))
686 avatar_metrics.record_file_upload(
687 file_size=photo_data.size,
688 content_type=photo_data.content_type,
689 success=False,
690 )
691 return HttpResponseRedirect(reverse_lazy("profile"))
693 # Log successful upload
694 security_logger.info(
695 f"Photo uploaded successfully by user {self.request.user.id}, "
696 f"photo ID: {photo.pk}"
697 )
699 # Record successful file upload metrics
700 avatar_metrics.record_file_upload(
701 file_size=photo_data.size,
702 content_type=photo_data.content_type,
703 success=True,
704 )
706 # Override success URL -> Redirect to crop page.
707 self.success_url = reverse_lazy("crop_photo", args=[photo.pk])
708 return super().form_valid(form)
711@method_decorator(login_required, name="dispatch")
712class AddOpenIDView(SuccessMessageMixin, FormView):
713 """
714 View class for adding OpenID
715 """
717 template_name = "add_openid.html"
718 form_class = AddOpenIDForm
719 success_url = reverse_lazy("profile")
721 def form_valid(self, form):
722 if openid_id := form.save(self.request.user):
723 # At this point we have an unconfirmed OpenID, but
724 # we do not add the message, that we successfully added it,
725 # since this is misleading
726 return HttpResponseRedirect(
727 reverse_lazy("openid_redirection", args=[openid_id])
728 )
729 else:
730 return render(self.request, self.template_name, {"form": form})
733@method_decorator(login_required, name="dispatch")
734class RemoveUnconfirmedOpenIDView(View):
735 """
736 View class for removing a unconfirmed OpenID
737 """
739 model = UnconfirmedOpenId
741 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
742 """
743 Handle post - remove unconfirmed openid
744 """
745 try:
746 openid = self.model.objects.get( # pylint: disable=no-member
747 user=request.user, id=kwargs["openid_id"]
748 )
749 openid.delete()
750 messages.success(request, _("ID removed"))
751 except (
752 self.model.DoesNotExist
753 ): # pragma: no cover pylint: disable=no-member,line-too-long
754 messages.error(request, _("ID does not exist"))
755 return HttpResponseRedirect(reverse_lazy("profile"))
758@method_decorator(login_required, name="dispatch")
759class RemoveConfirmedOpenIDView(View):
760 """
761 View class for removing a confirmed OpenID
762 """
764 model = ConfirmedOpenId
766 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
767 """
768 Handle post - remove confirmed openid
769 """
770 try:
771 openid = self.model.objects.get( # pylint: disable=no-member
772 user=request.user, id=kwargs["openid_id"]
773 )
774 try:
775 openidobj = (
776 UserOpenID.objects.get( # pylint: disable=no-member,line-too-long
777 user_id=request.user.id, claimed_id=openid.openid
778 )
779 )
780 openidobj.delete()
781 except Exception as exc: # pylint: disable=broad-except
782 # Why it is not there?
783 logger.warning(f"How did we get here: {exc}")
784 openid.delete()
785 messages.success(request, _("ID removed"))
786 except self.model.DoesNotExist: # pylint: disable=no-member
787 messages.error(request, _("ID does not exist"))
788 return HttpResponseRedirect(reverse_lazy("profile"))
791@method_decorator(login_required, name="dispatch")
792class RedirectOpenIDView(View):
793 """
794 Redirect view for OpenID
795 """
797 model = UnconfirmedOpenId
799 def get(self, request, *args, **kwargs): # pylint: disable=unused-argument
800 """
801 Handle get for OpenID redirect view
802 """
803 try:
804 unconfirmed = self.model.objects.get( # pylint: disable=no-member
805 user=request.user, id=kwargs["openid_id"]
806 )
807 except (
808 self.model.DoesNotExist
809 ): # pragma: no cover pylint: disable=no-member,line-too-long
810 messages.error(request, _("ID does not exist"))
811 return HttpResponseRedirect(reverse_lazy("profile"))
813 user_url = unconfirmed.openid
814 session = {"id": request.session.session_key}
816 oidutil.log = openid_logging
817 openid_consumer = consumer.Consumer(session, DjangoOpenIDStore())
819 try:
820 auth_request = openid_consumer.begin(user_url)
821 except consumer.DiscoveryFailure as exc:
822 messages.error(request, _(f"OpenID discovery failed: {exc}"))
823 return HttpResponseRedirect(reverse_lazy("profile"))
824 except UnicodeDecodeError as exc: # pragma: no cover
825 msg = _(
826 "OpenID discovery failed (userid=%(userid)s) for "
827 "%(userurl)s: %(message)s"
828 % {
829 "userid": request.user.id,
830 "userurl": user_url.encode("utf-8"),
831 "message": exc,
832 }
833 )
834 logger.error(f"message: {msg}")
835 messages.error(request, msg)
837 if auth_request is None: # pragma: no cover
838 messages.error(request, _("OpenID discovery failed"))
839 return HttpResponseRedirect(reverse_lazy("profile"))
841 realm = request.build_absolute_uri("/")[:-1] # pragma: no cover
842 return_url = realm + reverse( # pragma: no cover
843 "confirm_openid", args=[kwargs["openid_id"]]
844 )
845 return HttpResponseRedirect( # pragma: no cover
846 auth_request.redirectURL(realm, return_url)
847 )
850@method_decorator(login_required, name="dispatch")
851class ConfirmOpenIDView(View): # pragma: no cover
852 """
853 Confirm OpenID view
854 """
856 model = UnconfirmedOpenId
857 model_confirmed = ConfirmedOpenId
859 def do_request(self, data, *args, **kwargs): # pylint: disable=unused-argument
860 """
861 Handle request, called by get() or post()
862 """
863 session = {"id": self.request.session.session_key}
864 current_url = self.request.build_absolute_uri("/")[:-1] + self.request.path
865 openid_consumer = consumer.Consumer(session, DjangoOpenIDStore())
866 info = openid_consumer.complete(data, current_url)
867 if info.status == consumer.FAILURE:
868 messages.error(
869 self.request, _('Confirmation failed: "') + str(info.message) + '"'
870 )
871 return HttpResponseRedirect(reverse_lazy("profile"))
873 if info.status == consumer.CANCEL:
874 messages.error(self.request, _("Cancelled by user"))
875 return HttpResponseRedirect(reverse_lazy("profile"))
877 if info.status != consumer.SUCCESS:
878 messages.error(self.request, _("Unknown verification error"))
879 return HttpResponseRedirect(reverse_lazy("profile"))
881 try:
882 unconfirmed = self.model.objects.get( # pylint: disable=no-member
883 user=self.request.user, id=kwargs["openid_id"]
884 )
885 except self.model.DoesNotExist: # pylint: disable=no-member
886 messages.error(self.request, _("ID does not exist"))
887 return HttpResponseRedirect(reverse_lazy("profile"))
889 # TODO: Check for a reasonable expiration time
890 confirmed = self.model_confirmed()
891 confirmed.user = unconfirmed.user
892 confirmed.ip_address = get_client_ip(self.request)[0]
893 confirmed.openid = unconfirmed.openid
894 confirmed.save()
896 unconfirmed.delete()
898 # If there is a single image in this user's profile
899 # assign it to the new id
900 if self.request.user.photo_set.count() == 1:
901 confirmed.set_photo(self.request.user.photo_set.first())
903 # Also allow user to login using this OpenID (if not already taken)
904 if not UserOpenID.objects.filter( # pylint: disable=no-member
905 claimed_id=confirmed.openid
906 ).exists():
907 user_openid = UserOpenID()
908 user_openid.user = self.request.user
909 user_openid.claimed_id = confirmed.openid
910 user_openid.display_id = confirmed.openid
911 user_openid.save()
912 return HttpResponseRedirect(reverse_lazy("profile"))
914 def get(self, request, *args, **kwargs):
915 """
916 Handle get - confirm openid
917 """
918 return self.do_request(request.GET, *args, **kwargs)
920 def post(self, request, *args, **kwargs):
921 """
922 Handle post - confirm openid
923 """
924 return self.do_request(request.POST, *args, **kwargs)
927@method_decorator(login_required, name="dispatch")
928class CropPhotoView(TemplateView):
929 """
930 View class for cropping photos
931 """
933 template_name = "crop_photo.html"
934 success_url = reverse_lazy("profile")
935 model = Photo
937 def get(self, request, *args, **kwargs):
938 photo = self.model.objects.get(
939 pk=kwargs["pk"], user=request.user
940 ) # pylint: disable=no-member
941 email = request.GET.get("email")
942 openid = request.GET.get("openid")
943 return render(
944 self.request,
945 self.template_name,
946 {
947 "photo": photo,
948 "email": email,
949 "openid": openid,
950 },
951 )
953 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
954 """
955 Handle post - crop photo
956 """
957 photo = self.model.objects.get(
958 pk=kwargs["pk"], user=request.user
959 ) # pylint: disable=no-member
960 dimensions = {
961 "x": int(float(request.POST["x"])),
962 "y": int(float(request.POST["y"])),
963 "w": int(float(request.POST["w"])),
964 "h": int(float(request.POST["h"])),
965 }
966 email = openid = None
967 if "email" in request.POST:
968 with contextlib.suppress(ConfirmedEmail.DoesNotExist):
969 email = ConfirmedEmail.objects.get(email=request.POST["email"])
970 if "openid" in request.POST:
971 with contextlib.suppress(ConfirmedOpenId.DoesNotExist):
972 openid = ConfirmedOpenId.objects.get( # pylint: disable=no-member
973 openid=request.POST["openid"]
974 )
975 return photo.perform_crop(request, dimensions, email, openid)
978@method_decorator(login_required, name="dispatch") # pylint: disable=too-many-ancestors
979class UserPreferenceView(FormView, UpdateView):
980 """
981 View class for user preferences view/update
982 """
984 template_name = "preferences.html"
985 model = UserPreference
986 form_class = UpdatePreferenceForm
987 success_url = reverse_lazy("user_preference")
989 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
990 """
991 Process POST-ed data from this form
992 """
993 userpref = None
994 try:
995 userpref = self.request.user.userpreference
996 except ObjectDoesNotExist:
997 userpref = UserPreference(user=self.request.user)
998 userpref.theme = request.POST["theme"]
999 userpref.save()
1000 try:
1001 if request.POST["email"] != self.request.user.email:
1002 addresses = list(
1003 self.request.user.confirmedemail_set.all().values_list(
1004 "email", flat=True
1005 )
1006 )
1007 if request.POST["email"] not in addresses:
1008 messages.error(
1009 self.request,
1010 _(f'Mail address not allowed: {request.POST["email"]}'),
1011 )
1012 else:
1013 self.request.user.email = request.POST["email"]
1014 self.request.user.save()
1015 messages.info(self.request, _("Mail address changed."))
1016 except Exception as e: # pylint: disable=broad-except
1017 messages.error(self.request, _(f"Error setting new mail address: {e}"))
1019 try:
1020 if request.POST["first_name"] or request.POST["last_name"]:
1021 if request.POST["first_name"] != self.request.user.first_name:
1022 self.request.user.first_name = request.POST["first_name"]
1023 messages.info(self.request, _("First name changed."))
1024 if request.POST["last_name"] != self.request.user.last_name:
1025 self.request.user.last_name = request.POST["last_name"]
1026 messages.info(self.request, _("Last name changed."))
1027 self.request.user.save()
1028 except Exception as e: # pylint: disable=broad-except
1029 messages.error(self.request, _(f"Error setting names: {e}"))
1031 return HttpResponseRedirect(reverse_lazy("user_preference"))
1033 def get(self, request, *args, **kwargs):
1034 return render(
1035 self.request,
1036 self.template_name,
1037 {
1038 "THEMES": UserPreference.THEMES,
1039 },
1040 )
1042 def get_object(self, queryset=None):
1043 (obj, created) = UserPreference.objects.get_or_create(
1044 user=self.request.user
1045 ) # pylint: disable=no-member,unused-variable
1046 return obj
1049@method_decorator(login_required, name="dispatch")
1050class UploadLibravatarExportView(SuccessMessageMixin, FormView):
1051 """
1052 View class responsible for libravatar user data export upload
1053 """
1055 template_name = "upload_libravatar_export.html"
1056 form_class = UploadLibravatarExportForm
1057 success_message = _("Successfully uploaded")
1058 success_url = reverse_lazy("profile")
1059 model = User
1061 def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
1062 """
1063 Handle post request - choose items to import
1064 """
1065 if "save" in kwargs: # pylint: disable=too-many-nested-blocks
1066 if kwargs["save"] == "save":
1067 for arg in request.POST:
1068 if arg.startswith("email_"):
1069 email = request.POST[arg]
1070 if not ConfirmedEmail.objects.filter(
1071 email=email
1072 ) and not UnconfirmedEmail.objects.filter(
1073 email=email
1074 ): # pylint: disable=no-member
1075 try:
1076 unconfirmed = UnconfirmedEmail.objects.create( # pylint: disable=no-member
1077 user=request.user, email=email
1078 )
1079 unconfirmed.save()
1080 unconfirmed.send_confirmation_mail(
1081 url=request.build_absolute_uri("/")[:-1]
1082 )
1083 messages.info(
1084 request,
1085 "%s: %s"
1086 % (
1087 email,
1088 _(
1089 "address added successfully,\
1090 confirmation mail sent"
1091 ),
1092 ),
1093 )
1094 except Exception as exc: # pylint: disable=broad-except
1095 # DEBUG
1096 print(
1097 f"Exception during adding mail address ({email}): {exc}"
1098 )
1100 if arg.startswith("photo"):
1101 try:
1102 data = base64.decodebytes(bytes(request.POST[arg], "utf-8"))
1103 except binascii.Error as exc:
1104 logger.warning(f"Cannot decode photo: {exc}")
1105 continue
1106 try:
1107 pilobj = Image.open(BytesIO(data))
1108 out = BytesIO()
1109 pilobj.save(out, pilobj.format, quality=JPEG_QUALITY)
1110 out.seek(0)
1111 photo = Photo()
1112 photo.user = request.user
1113 photo.ip_address = get_client_ip(request)[0]
1114 photo.format = file_format(pilobj.format)
1115 photo.data = out.read()
1116 photo.save()
1117 except Exception as exc: # pylint: disable=broad-except
1118 logger.error(f"Exception during save: {exc}")
1119 continue
1121 return HttpResponseRedirect(reverse_lazy("profile"))
1122 return super().post(request, args, kwargs)
1124 def form_valid(self, form):
1125 data = self.request.FILES["export_file"]
1126 try:
1127 items = libravatar_read_gzdata(data.read())
1128 # DEBUG print(items)
1129 return render(
1130 self.request,
1131 "choose_libravatar_export.html",
1132 {
1133 "emails": items["emails"],
1134 "photos": items["photos"],
1135 },
1136 )
1137 except Exception as e:
1138 messages.error(self.request, _(f"Unable to parse file: {e}"))
1139 return HttpResponseRedirect(reverse_lazy("upload_export"))
1142@method_decorator(login_required, name="dispatch")
1143class ResendConfirmationMailView(View):
1144 """
1145 View class for resending confirmation mail
1146 """
1148 model = UnconfirmedEmail
1150 def get(self, request, *args, **kwargs): # pylint: disable=unused-argument
1151 """
1152 Handle post - resend confirmation mail for unconfirmed e-mail address
1153 """
1154 try:
1155 email = self.model.objects.get( # pylint: disable=no-member
1156 user=request.user, id=kwargs["email_id"]
1157 )
1158 except self.model.DoesNotExist: # pragma: no cover pylint: disable=no-member
1159 messages.error(request, _("ID does not exist"))
1160 else:
1161 try:
1162 email.send_confirmation_mail(url=request.build_absolute_uri("/")[:-1])
1163 messages.success(
1164 request, f'{_("Confirmation mail sent to")}: {email.email}'
1165 )
1166 except Exception as exc: # pylint: disable=broad-except
1167 messages.error(
1168 request,
1169 f'{_("Unable to send confirmation email for")} {email.email}: {exc}',
1170 )
1171 return HttpResponseRedirect(reverse_lazy("profile"))
1174class IvatarLoginView(LoginView):
1175 """
1176 View class for login
1177 """
1179 template_name = "login.html"
1181 @trace_authentication("login_attempt")
1182 def get(self, request, *args, **kwargs):
1183 """
1184 Handle get for login view
1185 """
1186 if request.user:
1187 if request.user.is_authenticated:
1188 # Respect the 'next' parameter if present
1189 next_url = request.GET.get("next")
1190 if next_url:
1191 return HttpResponseRedirect(next_url)
1192 return HttpResponseRedirect(reverse_lazy("profile"))
1193 return super().get(self, request, args, kwargs)
1195 @trace_authentication("login_post")
1196 def post(self, request, *args, **kwargs):
1197 """
1198 Handle login form submission
1199 """
1200 return super().post(request, *args, **kwargs)
1202 def get_context_data(self, **kwargs):
1203 context = super().get_context_data(**kwargs)
1204 context["with_fedora"] = SOCIAL_AUTH_FEDORA_KEY is not None
1205 return context
1208@method_decorator(login_required, name="dispatch")
1209class ProfileView(TemplateView):
1210 """
1211 View class for profile
1212 """
1214 template_name = "profile.html"
1216 def get(self, request, *args, **kwargs):
1217 if "profile_username" in kwargs:
1218 if not request.user.is_staff:
1219 return HttpResponseRedirect(reverse_lazy("profile"))
1220 with contextlib.suppress(Exception):
1221 u = User.objects.get(username=kwargs["profile_username"])
1222 request.user = u
1223 self._confirm_claimed_openid()
1224 return super().get(self, request, args, kwargs)
1226 def get_context_data(self, **kwargs):
1227 """
1228 Provide additional context data, like if max_photos is reached
1229 already or not.
1230 """
1231 context = super().get_context_data(**kwargs)
1232 context["max_photos"] = False
1233 if self.request.user:
1234 if self.request.user.photo_set.all().count() >= MAX_NUM_PHOTOS:
1235 context["max_photos"] = True
1236 return context
1238 def _confirm_claimed_openid(self):
1239 openids = self.request.user.useropenid_set.all()
1240 # If there is only one OpenID, we eventually need to add it to
1241 # the user account
1242 if openids.count() == 1:
1243 # Already confirmed, skip
1244 if ConfirmedOpenId.objects.filter( # pylint: disable=no-member
1245 openid=openids.first().claimed_id
1246 ).exists():
1247 return
1248 # For whatever reason, this is in unconfirmed state, skip
1249 if UnconfirmedOpenId.objects.filter( # pylint: disable=no-member
1250 openid=openids.first().claimed_id
1251 ).exists():
1252 return
1253 logger.debug(f"need to confirm: {openids.first()}")
1254 confirmed = ConfirmedOpenId()
1255 confirmed.user = self.request.user
1256 confirmed.ip_address = get_client_ip(self.request)[0]
1257 confirmed.openid = openids.first().claimed_id
1258 confirmed.save()
1261class PasswordResetView(PasswordResetViewOriginal):
1262 """
1263 View class for password reset
1264 """
1266 def post(self, request, *args, **kwargs):
1267 """
1268 Since we have the mail addresses in ConfirmedEmail model,
1269 we need to set the email on the user object in order for the
1270 PasswordResetView class to pick up the correct user.
1271 In case we have the mail address in the User object, we still
1272 need to assign a random password in order for PasswordResetView
1273 class to pick up the user - else it will silently do nothing.
1274 """
1275 if "email" in request.POST:
1276 user = None
1278 # Try to find the user via the normal user class
1279 # TODO: How to handle the case that multiple user accounts
1280 # could have the same password set?
1281 user = User.objects.filter(email=request.POST["email"]).first()
1283 # If we didn't find the user in the previous step,
1284 # try the ConfirmedEmail class instead.
1285 # If we find the user there, we need to set the mail
1286 # attribute on the user object accordingly
1287 if not user:
1288 with contextlib.suppress(ObjectDoesNotExist):
1289 confirmed_email = ConfirmedEmail.objects.get(
1290 email=request.POST["email"]
1291 )
1292 user = confirmed_email.user
1293 user.email = confirmed_email.email
1294 user.save()
1295 # If we found the user, set a random password. Else, the
1296 # ResetPasswordView class will silently ignore the password
1297 # reset request
1298 if user:
1299 if not user.password or user.password.startswith("!"):
1300 random_pass = get_random_string(12)
1301 user.set_password(random_pass)
1302 user.save()
1304 # Whatever happens above, let the original function handle the rest
1305 return super().post(self, request, args, kwargs)
1308@method_decorator(login_required, name="dispatch")
1309class DeleteAccountView(SuccessMessageMixin, FormView):
1310 """
1311 View class for account deletion
1312 """
1314 template_name = "delete.html"
1315 form_class = DeleteAccountForm
1316 success_url = reverse_lazy("home")
1318 def get(self, request, *args, **kwargs):
1319 return super().get(self, request, args, kwargs)
1321 def post(self, request, *args, **kwargs):
1322 """
1323 Handle account deletion
1324 """
1325 if request.user.password:
1326 if "password" in request.POST:
1327 if not request.user.check_password(request.POST["password"]):
1328 messages.error(request, _("Incorrect password"))
1329 return HttpResponseRedirect(reverse_lazy("delete"))
1330 else:
1331 messages.error(request, _("No password given"))
1332 return HttpResponseRedirect(reverse_lazy("delete"))
1334 # should delete all confirmed/unconfirmed/photo objects
1335 request.user.delete()
1336 return super().post(self, request, args, kwargs)
1339@method_decorator(login_required, name="dispatch")
1340class ExportView(SuccessMessageMixin, TemplateView):
1341 """
1342 View class responsible for libravatar user data export
1343 """
1345 template_name = "export.html"
1346 model = User
1348 def get(self, request, *args, **kwargs):
1349 return super().get(self, request, args, kwargs)
1351 def post(self, request, *args, **kwargs):
1352 """
1353 Handle real export
1354 """
1355 SCHEMA_ROOT = "https://www.libravatar.org/schemas/export/0.2"
1356 SCHEMA_XSD = f"{SCHEMA_ROOT}/export.xsd"
1358 def xml_header():
1359 return (
1360 """<?xml version="1.0" encoding="UTF-8"?>"""
1361 '''<user xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"'''
1362 ''' xsi:schemaLocation="%s %s"'''
1363 """ xmlns="%s">\n""" % (SCHEMA_ROOT, SCHEMA_XSD, SCHEMA_ROOT)
1364 )
1366 def xml_footer():
1367 return "</user>\n"
1369 def xml_account(user):
1370 escaped_username = saxutils.quoteattr(user.username)
1371 escaped_password = saxutils.quoteattr(user.password)
1372 return " <account username={} password={}/>\n".format(
1373 escaped_username,
1374 escaped_password,
1375 )
1377 def xml_email(user):
1378 returnstring = " <emails>\n"
1379 for email in user.confirmedemail_set.all():
1380 returnstring += (
1381 ' <email photo_id="'
1382 + str(email.photo_id)
1383 + '">'
1384 + str(email.email)
1385 + "</email>"
1386 + "\n"
1387 )
1388 returnstring += " </emails>\n"
1389 return returnstring
1391 def xml_openid(user):
1392 returnstring = " <openids>\n"
1393 for openid in user.confirmedopenid_set.all():
1394 returnstring += (
1395 ' <openid photo_id="'
1396 + str(openid.photo_id)
1397 + '">'
1398 + str(openid.openid)
1399 + "</openid>"
1400 + "\n"
1401 )
1402 returnstring += " </openids>\n"
1403 return returnstring
1405 def xml_photos(user):
1406 s = " <photos>\n"
1407 for photo in user.photo_set.all():
1408 encoded_photo = base64.b64encode(photo.data)
1409 if encoded_photo:
1410 s += (
1411 """ <photo id="%s" encoding="base64" format=%s>"""
1412 """%s"""
1413 """</photo>\n"""
1414 % (photo.id, saxutils.quoteattr(photo.format), encoded_photo)
1415 )
1416 s += " </photos>\n"
1417 return s
1419 user = request.user
1421 photos = []
1422 for photo in user.photo_set.all():
1423 photo_details = {"data": photo.data, "format": photo.format}
1424 photos.append(photo_details)
1426 bytesobj = BytesIO()
1427 data = gzip.GzipFile(fileobj=bytesobj, mode="w")
1428 data.write(bytes(xml_header(), "utf-8"))
1429 data.write(bytes(xml_account(user), "utf-8"))
1430 data.write(bytes(xml_email(user), "utf-8"))
1431 data.write(bytes(xml_openid(user), "utf-8"))
1432 data.write(bytes(xml_photos(user), "utf-8"))
1433 data.write(bytes(xml_footer(), "utf-8"))
1434 data.close()
1435 bytesobj.seek(0)
1437 response = HttpResponse(content_type="application/gzip")
1438 response["Content-Disposition"] = (
1439 f'attachment; filename="libravatar-export_{user.username}.xml.gz"'
1440 )
1441 response.write(bytesobj.read())
1442 return response