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