Coverage for ivatar/views.py: 78%
746 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-20 05:10 +0000
1"""
2views under /
3"""
5import contextlib
6from io import BytesIO
7from os import path
8import hashlib
9import logging
10import threading
11from typing import Dict, Tuple, Optional
12from ivatar.utils import urlopen, Bluesky
13from ivatar.access_stats import stats_manager
14import urllib.parse
15from urllib.error import HTTPError, URLError
16from ssl import SSLError
17from django.conf import settings
18from django.views.generic.base import TemplateView, View
19from django.http import HttpResponse, HttpResponseRedirect
20from django.http import HttpResponseNotFound, JsonResponse
21from django.core.exceptions import ObjectDoesNotExist
22from django.core.cache import cache, caches
23from django.utils.cache import get_conditional_response
24from django.utils.http import http_date
25from django.utils.translation import gettext_lazy as _
26from django.urls import reverse_lazy
27from django.db.models import Q
28from django.contrib.auth.models import User
30from PIL import Image
32from monsterid.id import build_monster as BuildMonster
33import Identicon
34from pydenticon5 import Pydenticon5
35from .robohash import create_robohash
36from .pagan_optimized import create_optimized_pagan
38# Access settings through django.conf.settings
39AVATAR_MAX_SIZE = getattr(settings, "AVATAR_MAX_SIZE", 512)
40JPEG_QUALITY = getattr(settings, "JPEG_QUALITY", 85)
41DEFAULT_AVATAR_SIZE = getattr(settings, "DEFAULT_AVATAR_SIZE", 80)
43from .ivataraccount.models import ConfirmedEmail, ConfirmedOpenId
44from .ivataraccount.models import UnconfirmedEmail, UnconfirmedOpenId
45from .ivataraccount.models import Photo
46from .ivataraccount.models import pil_format, file_format
47from .utils import is_trusted_url, mm_ng, resize_animated_gif, get_resample_filter
48from .photo_cache import (
49 get_cached_photo,
50 set_cached_photo,
51 render_photo_thumbnail,
52)
54# Import OpenTelemetry with graceful degradation
55from .telemetry_utils import trace_avatar_operation, get_telemetry_metrics
57avatar_metrics = get_telemetry_metrics()
59# Initialize loggers
60logger = logging.getLogger("ivatar")
61security_logger = logging.getLogger("ivatar.security")
64def get_size(request, size=DEFAULT_AVATAR_SIZE):
65 """
66 Get size from the URL arguments
67 """
68 sizetemp = None
69 if "s" in request.GET:
70 sizetemp = request.GET["s"]
71 if "size" in request.GET:
72 sizetemp = request.GET["size"]
73 if sizetemp:
74 if sizetemp not in ["", "0"]:
75 with contextlib.suppress(ValueError):
76 if int(sizetemp) > 0:
77 size = int(sizetemp)
78 size = min(size, int(AVATAR_MAX_SIZE))
79 return size
82# Thread-safe in-memory cache for default avatar PNG bytes
83_default_avatars_png_cache: Dict[Tuple[str, str, int], bytes] = {}
84_default_avatars_png_cache_lock = threading.Lock()
85_MAX_DEFAULT_PNG_CACHE_SIZE = 500 # Bounded in-process cache size
88def negotiate_image_format(request, original_format: str) -> Tuple[str, bool]:
89 """
90 Negotiate optimal image format based on client Accept headers.
91 Returns (negotiated_format, is_negotiated).
92 If client accepts image/webp and ENABLE_WEBP_CONTENT_NEGOTIATION is enabled,
93 returns ('webp', True).
94 Otherwise returns (normalized_original_format, False).
95 """
96 normalized = "jpeg" if original_format in ("jpg", "jpeg") else original_format
97 if not getattr(settings, "ENABLE_WEBP_CONTENT_NEGOTIATION", True):
98 return normalized, False
100 accept = request.headers.get("Accept", request.META.get("HTTP_ACCEPT", ""))
101 if "image/webp" in accept:
102 return "webp", True
104 return normalized, False
107def _get_cached_default_avatar(
108 style: str, digest: str, size: int
109) -> Optional[BytesIO]:
110 cache_key = (style, digest, size)
111 # 1. First check fast in-memory process cache
112 with _default_avatars_png_cache_lock:
113 if cache_key in _default_avatars_png_cache:
114 return BytesIO(_default_avatars_png_cache[cache_key])
116 # 2. Fall back to shared Redis / Django cache
117 redis_key = f"default_avatar:{style}:{digest}:{size}"
118 with contextlib.suppress(Exception):
119 data_bytes = cache.get(redis_key)
120 if data_bytes is not None:
121 # Populate local in-process cache
122 with _default_avatars_png_cache_lock:
123 if len(_default_avatars_png_cache) >= _MAX_DEFAULT_PNG_CACHE_SIZE:
124 remove_keys = list(_default_avatars_png_cache.keys())[:50]
125 for rk in remove_keys:
126 del _default_avatars_png_cache[rk]
127 _default_avatars_png_cache[cache_key] = data_bytes
128 return BytesIO(data_bytes)
130 return None
133def _set_cached_default_avatar(
134 style: str, digest: str, size: int, png_bytes: bytes
135) -> None:
136 cache_key = (style, digest, size)
137 # 1. Save in in-process memory cache
138 with _default_avatars_png_cache_lock:
139 if len(_default_avatars_png_cache) >= _MAX_DEFAULT_PNG_CACHE_SIZE:
140 # Evict oldest 50 entries
141 remove_keys = list(_default_avatars_png_cache.keys())[:50]
142 for rk in remove_keys:
143 del _default_avatars_png_cache[rk]
144 _default_avatars_png_cache[cache_key] = png_bytes
146 # 2. Save in shared Redis / Django cache
147 redis_key = f"default_avatar:{style}:{digest}:{size}"
148 with contextlib.suppress(Exception):
149 timeout = getattr(settings, "DEFAULT_AVATAR_CACHE_TIMEOUT", 604800) # 7 days default
150 cache.set(redis_key, png_bytes, timeout=timeout)
153class CachingHttpResponse(HttpResponse):
154 """
155 Handle caching of response
156 """
158 def __init__(
159 self,
160 uri,
161 content=b"",
162 content_type=None,
163 status=200, # pylint: disable=too-many-arguments
164 reason=None,
165 charset=None,
166 ):
167 if settings.CACHE_RESPONSE:
168 cache_alias = getattr(settings, "AVATAR_CACHE_ALIAS", "filesystem")
169 caches[cache_alias].set(
170 uri,
171 {
172 "content": content,
173 "content_type": content_type,
174 "status": status,
175 "reason": reason,
176 "charset": charset,
177 },
178 )
179 super().__init__(content, content_type, status, reason, charset)
182class AvatarImageView(TemplateView):
183 """
184 View to return (binary) image, based on OpenID/Email (both by digest)
185 """
187 # User photos are cached in Redis / memory via photo_cache module
189 def options(self, request, *args, **kwargs):
190 response = HttpResponse("", content_type="text/plain")
191 response["Allow"] = "404 mm mp retro pagan wavatar monsterid robohash identicon"
192 return response
194 @trace_avatar_operation("avatar_request")
195 def get(
196 self, request, *args, **kwargs
197 ): # pylint: disable=too-many-branches,too-many-statements,too-many-locals,too-many-return-statements
198 """
199 Override get from parent class
200 """
201 model = ConfirmedEmail
202 size = get_size(request)
203 imgformat = "png"
204 obj = None
205 default = None
206 forcedefault = settings.FORCEDEFAULT
207 gravatarredirect = settings.DEFAULT_GRAVATARREDIRECT
208 gravatarproxy = settings.DEFAULT_GRAVATARPROXY
209 uri = request.build_absolute_uri()
210 client_accepts_webp = (
211 "image/webp" in request.headers.get("Accept", request.META.get("HTTP_ACCEPT", ""))
212 and getattr(settings, "ENABLE_WEBP_CONTENT_NEGOTIATION", True)
213 )
214 uri_cache_key = f"{uri}:webp" if client_accepts_webp else f"{uri}:standard"
216 # Check the cache first
217 if settings.CACHE_RESPONSE:
218 cache_alias = getattr(settings, "AVATAR_CACHE_ALIAS", "filesystem")
219 if centry := caches[cache_alias].get(uri_cache_key):
220 format_type = "webp" if client_accepts_webp else imgformat
221 # Record cache hit
222 avatar_metrics.record_cache_hit(size=str(size), format_type=format_type)
223 # For DEBUG purpose only
224 # print('Cached entry for %s' % uri)
225 resp = HttpResponse(
226 centry["content"],
227 content_type=centry["content_type"],
228 status=centry["status"],
229 reason=centry["reason"],
230 charset=centry["charset"],
231 )
232 if getattr(settings, "ENABLE_WEBP_CONTENT_NEGOTIATION", True):
233 resp["Vary"] = "Accept"
234 if "digest" in kwargs:
235 etag = f'"{kwargs["digest"]}"'
236 resp["Etag"] = etag
237 conditional_response = get_conditional_response(
238 request, etag=etag, response=resp
239 )
240 if conditional_response is not resp:
241 conditional_response["Etag"] = etag
242 if "Vary" in resp:
243 conditional_response["Vary"] = resp["Vary"]
244 return conditional_response
245 return resp
246 else:
247 # Record cache miss
248 avatar_metrics.record_cache_miss(size=str(size), format_type=imgformat)
250 # In case no digest at all is provided, return to home page
251 if "digest" not in kwargs:
252 return HttpResponseRedirect(reverse_lazy("home"))
254 if "d" in request.GET:
255 default = request.GET["d"]
256 if "default" in request.GET:
257 default = request.GET["default"]
259 if default is not None:
260 if settings.TRUSTED_DEFAULT_URLS is None:
261 logger.warning("Query parameter `default` is disabled.")
262 default = None
263 elif default.find("://") > 0:
264 # Check if it's trusted, if not, reset to None
265 trusted_url = is_trusted_url(default, settings.TRUSTED_DEFAULT_URLS)
267 if not trusted_url:
268 security_logger.warning(
269 f"Default URL is not in trusted URLs: '{default}'; Kicking it!"
270 )
271 default = None
273 if "f" in request.GET:
274 if request.GET["f"] == "y":
275 forcedefault = True
276 if "forcedefault" in request.GET:
277 if request.GET["forcedefault"] == "y":
278 forcedefault = True
280 if "gravatarredirect" in request.GET:
281 if request.GET["gravatarredirect"] == "y":
282 gravatarredirect = True
284 if "gravatarproxy" in request.GET:
285 if request.GET["gravatarproxy"] == "n":
286 gravatarproxy = False
288 def redir_default(default=None):
289 url = (
290 reverse_lazy("avatar_view", args=[kwargs["digest"]])
291 + "?s=%i" % size
292 + "&forcedefault=y"
293 )
294 if default is not None:
295 encoded_default = urllib.parse.quote(
296 str(default), safe=":/?#[]@!$&'()*+,;=-_.~%"
297 )
298 url += f"&default={encoded_default}"
299 return HttpResponseRedirect(url)
301 digest = kwargs.get("digest", "")
302 digest_len = len(digest)
304 # Optimization: Only look up in relevant tables/fields based on digest length
305 # MD5 is 32 chars, SHA256 is 64 chars
306 if digest_len == 32:
307 with contextlib.suppress(ObjectDoesNotExist):
308 obj = (
309 ConfirmedEmail.objects.select_related("photo")
310 .defer("photo__data")
311 .get(digest=digest)
312 )
313 elif digest_len == 64:
314 with contextlib.suppress(ObjectDoesNotExist):
315 obj = (
316 ConfirmedEmail.objects.select_related("photo")
317 .defer("photo__data")
318 .get(digest_sha256=digest)
319 )
321 if not obj:
322 # OpenID digests are always SHA256 (64 chars)
323 with contextlib.suppress(Exception):
324 obj = (
325 ConfirmedOpenId.objects.select_related("photo")
326 .defer("photo__data")
327 .filter(
328 Q(digest=digest)
329 | Q(alt_digest1=digest)
330 | Q(alt_digest2=digest)
331 | Q(alt_digest3=digest)
332 )
333 .first()
334 )
336 # Handle the special case of Bluesky
337 if obj:
338 if obj.bluesky_handle:
339 return HttpResponseRedirect(
340 reverse_lazy("blueskyproxy", args=[digest])
341 )
342 # If that mail/openid doesn't exist, or has no photo linked to it
343 if not obj or not obj.photo or forcedefault:
344 gravatar_url = (
345 "https://secure.gravatar.com/avatar/"
346 + kwargs["digest"]
347 + "?s=%i" % size
348 )
350 # If we have redirection to Gravatar enabled, this overrides all
351 # default= settings, except forcedefault!
352 if gravatarredirect and not forcedefault:
353 return HttpResponseRedirect(gravatar_url)
355 # Request to proxy Gravatar image - only if not forcedefault
356 if gravatarproxy and not forcedefault:
357 url = (
358 reverse_lazy("gravatarproxy", args=[kwargs["digest"]])
359 + "?s=%i" % size
360 )
361 # Ensure we do not convert None to string 'None'
362 if default:
363 encoded_default = urllib.parse.quote(
364 str(default), safe=":/?#[]@!$&'()*+,;=-_.~%"
365 )
366 url += f"&default={encoded_default}"
367 return HttpResponseRedirect(url)
369 # Return the default URL, as specified, or 404 Not Found, if default=404
370 if default:
371 # Proxy to gravatar to generate wavatar - lazy me
372 if str(default) == "wavatar":
373 url = (
374 reverse_lazy("gravatarproxy", args=[kwargs["digest"]])
375 + "?s=%i" % size
376 + f"&default={default}&f=y"
377 )
378 return HttpResponseRedirect(url)
380 if str(default) == str(404):
381 return HttpResponseNotFound(_("<h1>Image not found</h1>"))
383 try:
384 if str(default) in {"monsterid", "robohash", "retro", "pagan", "identicon", "mmng"}:
385 etag = f'"{kwargs["digest"]}"'
386 conditional_response = get_conditional_response(request, etag=etag)
387 if conditional_response:
388 conditional_response["Etag"] = etag
389 conditional_response["Vary"] = ""
390 return conditional_response
392 if str(default) == "monsterid":
393 data = _get_cached_default_avatar(
394 "monsterid", kwargs["digest"], size
395 )
396 if not data:
397 monsterdata = BuildMonster(
398 seed=kwargs["digest"], size=(size, size)
399 )
400 data = BytesIO()
401 monsterdata.save(data, "PNG")
402 _set_cached_default_avatar(
403 "monsterid", kwargs["digest"], size, data.getvalue()
404 )
405 avatar_metrics.record_avatar_generated(
406 size=str(size), format_type="png", source="monsterid"
407 )
408 return self._return_cached_response(data, uri_cache_key)
409 if str(default) == "robohash":
410 roboset = request.GET.get("robohash") or "any"
411 data = create_robohash(kwargs["digest"], size, roboset)
412 avatar_metrics.record_avatar_generated(
413 size=str(size), format_type="png", source="robohash"
414 )
415 return self._return_cached_response(data, uri_cache_key)
416 if str(default) == "retro":
417 data = _get_cached_default_avatar(
418 "retro", kwargs["digest"], size
419 )
420 if not data:
421 identicon = Identicon.render(kwargs["digest"])
422 img = Image.open(BytesIO(identicon))
423 img = img.resize((size, size), get_resample_filter(size, img.size))
424 data = BytesIO()
425 img.save(data, "PNG")
426 _set_cached_default_avatar(
427 "retro", kwargs["digest"], size, data.getvalue()
428 )
429 avatar_metrics.record_avatar_generated(
430 size=str(size), format_type="png", source="retro"
431 )
432 return self._return_cached_response(data, uri_cache_key)
433 if str(default) == "pagan":
434 data = create_optimized_pagan(kwargs["digest"], size)
435 avatar_metrics.record_avatar_generated(
436 size=str(size), format_type="png", source="pagan"
437 )
438 return self._return_cached_response(data, uri_cache_key)
439 if str(default) == "identicon":
440 data = _get_cached_default_avatar(
441 "identicon", kwargs["digest"], size
442 )
443 if not data:
444 p = Pydenticon5()
445 # In order to make use of the whole 32 bytes digest, we need to redigest them.
446 newdigest = hashlib.md5(
447 bytes(kwargs["digest"], "utf-8")
448 ).hexdigest()
449 img = p.draw(newdigest, size, 0)
450 data = BytesIO()
451 img.save(data, "PNG")
452 _set_cached_default_avatar(
453 "identicon", kwargs["digest"], size, data.getvalue()
454 )
455 avatar_metrics.record_avatar_generated(
456 size=str(size), format_type="png", source="identicon"
457 )
458 return self._return_cached_response(data, uri_cache_key)
459 if str(default) == "mmng":
460 data = _get_cached_default_avatar(
461 "mmng", kwargs["digest"], size
462 )
463 if not data:
464 mmngimg = mm_ng(idhash=kwargs["digest"], size=size)
465 data = BytesIO()
466 mmngimg.save(data, "PNG")
467 _set_cached_default_avatar(
468 "mmng", kwargs["digest"], size, data.getvalue()
469 )
470 avatar_metrics.record_avatar_generated(
471 size=str(size), format_type="png", source="mmng"
472 )
473 return self._return_cached_response(data, uri_cache_key)
474 if str(default) == "wavatar":
475 url = (
476 reverse_lazy("gravatarproxy", args=[kwargs["digest"]])
477 + f"?s={size}&default=wavatar"
478 )
479 return HttpResponseRedirect(url)
480 if str(default) in {"mm", "mp"}:
481 return self._redirect_static_w_size("mm", size)
482 except Exception as exc:
483 logger.error(f"Error generating default avatar {default}: {exc}")
484 return self._redirect_static_w_size("nobody", size)
486 return HttpResponseRedirect(default)
488 return self._redirect_static_w_size("nobody", size)
489 try:
490 imgformat = obj.photo.format
491 target_format, is_negotiated = negotiate_image_format(request, imgformat)
492 etag = f'"{kwargs["digest"]}"'
493 last_modified = (
494 int(obj.photo.add_date.timestamp())
495 if hasattr(obj.photo, "add_date") and obj.photo.add_date
496 else None
497 )
499 # Fast-path: Check conditional request (If-None-Match / If-Modified-Since)
500 conditional_response = get_conditional_response(
501 request,
502 etag=etag,
503 last_modified=last_modified,
504 )
505 if conditional_response:
506 stats_manager.update_access_count(obj.photo)
507 stats_manager.update_access_count(obj)
508 conditional_response["Etag"] = etag
509 if last_modified:
510 conditional_response["Last-Modified"] = http_date(last_modified)
511 conditional_response["Vary"] = (
512 "Accept"
513 if getattr(settings, "ENABLE_WEBP_CONTENT_NEGOTIATION", True)
514 else ""
515 )
516 return conditional_response
518 # Check cached resized photo in Redis / L1 memory cache
519 data = get_cached_photo(obj.photo.id, size, target_format)
520 if data is not None:
521 avatar_metrics.record_cache_hit(
522 size=str(size),
523 format_type=target_format,
524 )
525 else:
526 avatar_metrics.record_cache_miss(
527 size=str(size),
528 format_type=target_format,
529 )
530 thumb_bytes = render_photo_thumbnail(
531 obj.photo.data, imgformat, size, target_format=target_format
532 )
533 data = BytesIO(thumb_bytes)
535 set_cached_photo(
536 obj.photo.id,
537 size,
538 target_format,
539 thumb_bytes,
540 )
542 # Record avatar generation metrics
543 avatar_metrics.record_avatar_generated(
544 size=str(size),
545 format_type=target_format,
546 source="uploaded" if obj else "generated",
547 )
549 data.seek(0)
550 stats_manager.update_access_count(obj.photo)
551 stats_manager.update_access_count(obj)
553 response = CachingHttpResponse(
554 uri_cache_key, data, content_type=f"image/{target_format}"
555 )
556 response["Etag"] = etag
557 if last_modified:
558 response["Last-Modified"] = http_date(last_modified)
559 # Set Vary: Accept so edge proxies/CDNs cache WebP and JPEG/PNG separately
560 response["Vary"] = (
561 "Accept"
562 if getattr(settings, "ENABLE_WEBP_CONTENT_NEGOTIATION", True)
563 else ""
564 )
565 return response
566 except Exception as exc:
567 logger.error(f"Error processing photo for {kwargs['digest']}: {exc}")
568 return redir_default(default)
570 def _redirect_static_w_size(self, arg0, size):
571 """
572 Helper method to redirect to static image with size i/a
573 """
574 # If mm is explicitly given, we need to catch that
575 static_img = path.join("static", "img", arg0, f"{str(size)}.png")
576 if not path.isfile(static_img):
577 # We trust this exists!!!
578 static_img = path.join("static", "img", arg0, "512.png")
579 # We trust static/ is mapped to /static/
580 return HttpResponseRedirect(f"/{static_img}")
582 def _return_cached_response(self, data, uri):
583 data.seek(0)
584 response = CachingHttpResponse(uri, data, content_type="image/png")
585 # Remove Vary header for images since language doesn't matter
586 response["Vary"] = ""
587 return response
589 @trace_avatar_operation("generate_png")
590 def _return_cached_png(self, arg0, data, uri):
591 arg0.save(data, "PNG")
592 return self._return_cached_response(data, uri)
595class GravatarProxyView(View):
596 """
597 Proxy request to Gravatar and return the image from there
598 """
600 # TODO: Do cache images!! Memcached?
602 @trace_avatar_operation("gravatar_proxy")
603 def get(
604 self, request, *args, **kwargs
605 ): # pylint: disable=too-many-branches,too-many-statements,too-many-locals,no-self-use,unused-argument,too-many-return-statements
606 """
607 Override get from parent class
608 """
610 def redir_default(default=None):
611 url = (
612 reverse_lazy("avatar_view", args=[kwargs["digest"]])
613 + "?s=%i" % size
614 + "&forcedefault=y"
615 )
616 if default is not None:
617 encoded_default = urllib.parse.quote(
618 str(default), safe=":/?#[]@!$&'()*+,;=-_.~%"
619 )
620 url += f"&default={encoded_default}"
621 return HttpResponseRedirect(url)
623 size = get_size(request)
624 gravatarimagedata = None
625 default = None
627 with contextlib.suppress(Exception):
628 if str(request.GET["default"]) != "None":
629 default = request.GET["default"]
630 if str(default) != "wavatar":
631 # This part is special/hackish
632 # Check if the image returned by Gravatar is their default image, if so,
633 # redirect to our default instead.
634 gravatar_test_url = (
635 "https://secure.gravatar.com/avatar/"
636 + kwargs["digest"]
637 + "?s=%i&d=%i" % (50, 404)
638 )
639 gravatar_test_key = (
640 f"gravatar_test:{hashlib.sha256(gravatar_test_url.encode('utf-8')).hexdigest()}"
641 )
642 if cache.get(gravatar_test_key) == "default":
643 # DEBUG only
644 # print("Cached Gravatar response: Default.")
645 return redir_default(default)
646 try:
647 urlopen(gravatar_test_url)
648 # Cache negative check / presence check for 10 minutes (600s)
649 cache.set(gravatar_test_key, "found", 600)
650 except HTTPError as exc:
651 if exc.code == 404:
652 cache.set(gravatar_test_key, "default", 600)
653 else:
654 logger.warning(f"Gravatar test url fetch failed: {exc}")
655 return redir_default(default)
657 gravatar_url = (
658 "https://secure.gravatar.com/avatar/" + kwargs["digest"] + "?s=%i" % size
659 )
660 if default:
661 encoded_default = urllib.parse.quote(
662 str(default), safe=":/?#[]@!$&'()*+,;=-_.~%"
663 )
664 gravatar_url += f"&d={encoded_default}"
666 # Cache key for proxied image payload (hashed to ensure safe characters and bounded length)
667 gravatar_url_hash = hashlib.sha256(gravatar_url.encode("utf-8")).hexdigest()
668 proxy_cache_key = f"gravatar_proxy_img:{gravatar_url_hash}"
669 gravatar_err_key = f"gravatar_err:{gravatar_url_hash}"
670 cache_alias = getattr(settings, "AVATAR_CACHE_ALIAS", "default")
672 if cached_payload := caches[cache_alias].get(proxy_cache_key):
673 response = HttpResponse(
674 cached_payload["content"], content_type=cached_payload["content_type"]
675 )
676 response["Vary"] = ""
677 return response
679 try:
680 if cache.get(gravatar_err_key) == "err":
681 logger.warning(
682 f"Cached Gravatar fetch failed with URL error: {gravatar_url}"
683 )
684 avatar_metrics.record_external_request("gravatar", 0) # Cached error
685 return redir_default(default)
687 gravatarimagedata = urlopen(gravatar_url)
688 avatar_metrics.record_external_request("gravatar", 200)
689 except HTTPError as exc:
690 if exc.code not in [404, 503]:
691 logger.warning(
692 f"Gravatar fetch failed with an unexpected {exc.code} HTTP error: {gravatar_url}"
693 )
694 avatar_metrics.record_external_request("gravatar", exc.code)
695 cache.set(gravatar_err_key, "err", 30)
696 return redir_default(default)
697 except URLError as exc:
698 logger.warning(f"Gravatar fetch failed with URL error: {exc.reason}")
699 avatar_metrics.record_external_request("gravatar", 0) # Network error
700 cache.set(gravatar_err_key, "err", 30)
701 return redir_default(default)
702 except SSLError as exc:
703 logger.warning(f"Gravatar fetch failed with SSL error: {exc.reason}")
704 avatar_metrics.record_external_request("gravatar", 0) # SSL error
705 cache.set(gravatar_err_key, "err", 30)
706 return redir_default(default)
707 try:
708 raw_bytes = gravatarimagedata.read()
709 data = BytesIO(raw_bytes)
710 img = Image.open(data)
711 content_type = f"image/{file_format(img.format)}"
713 # Cache successful response payload for 1 hour (3600 seconds)
714 caches[cache_alias].set(
715 proxy_cache_key,
716 {"content": raw_bytes, "content_type": content_type},
717 3600,
718 )
720 response = HttpResponse(raw_bytes, content_type=content_type)
721 # Remove Vary header for images since language doesn't matter
722 response["Vary"] = ""
723 return response
725 except ValueError as exc:
726 logger.error(f"Value error: {exc}")
727 return redir_default(default)
729 # We shouldn't reach this point... But make sure we do something
730 return redir_default(default)
733class BlueskyProxyView(View):
734 """
735 Proxy request to Bluesky and return the image from there
736 """
738 @trace_avatar_operation("bluesky_proxy")
739 def get(
740 self, request, *args, **kwargs
741 ): # pylint: disable=too-many-branches,too-many-statements,too-many-locals,no-self-use,unused-argument,too-many-return-statements
742 """
743 Override get from parent class
744 """
746 def redir_default(default=None):
747 url = (
748 reverse_lazy("avatar_view", args=[kwargs["digest"]])
749 + "?s=%i" % size
750 + "&forcedefault=y"
751 )
752 if default is not None:
753 url += f"&default={default}"
754 return HttpResponseRedirect(url)
756 size = get_size(request)
757 logger.debug(f"Bluesky avatar size requested: {size}")
758 blueskyimagedata = None
759 default = None
761 with contextlib.suppress(Exception):
762 if str(request.GET["default"]) != "None":
763 default = request.GET["default"]
764 digest = kwargs.get("digest", "")
765 digest_len = len(digest)
766 identity = None
768 # Optimization: Only look up in relevant tables/fields based on digest length
769 if digest_len == 32:
770 try:
771 identity = ConfirmedEmail.objects.filter(digest=digest).first()
772 except Exception as exc:
773 logger.warning(f"Exception: {exc}")
774 elif digest_len == 64:
775 try:
776 identity = ConfirmedEmail.objects.filter(digest_sha256=digest).first()
777 except Exception as exc:
778 logger.warning(f"Exception: {exc}")
780 if not identity:
781 try:
782 identity = ConfirmedOpenId.objects.filter(
783 Q(digest=digest)
784 | Q(alt_digest1=digest)
785 | Q(alt_digest2=digest)
786 | Q(alt_digest3=digest)
787 ).first()
788 except Exception as exc:
789 logger.warning(f"Exception: {exc}")
791 # If still no identity is found, redirect to the default
792 if not identity:
793 return redir_default(default)
795 bs = Bluesky()
796 bluesky_url = None
797 # Try with the cache first
798 with contextlib.suppress(Exception):
799 if cache.get(identity.bluesky_handle):
800 bluesky_url = cache.get(identity.bluesky_handle)
801 if not bluesky_url:
802 try:
803 bluesky_url = bs.get_avatar(identity.bluesky_handle)
804 cache.set(identity.bluesky_handle, bluesky_url)
805 except Exception: # pylint: disable=bare-except
806 return redir_default(default)
808 # Stable ETag identifies the payload by URL + size (Bluesky avatar
809 # URLs embed the image revision). Evaluate conditional requests first
810 # so revalidating clients skip both the payload cache and download.
811 bs_cache_timeout = getattr(settings, "BLUESKY_CACHE_TIMEOUT", 3600)
812 bs_image_timeout = getattr(settings, "BLUESKY_IMAGE_TIMEOUT", 5)
813 etag = f'"{hashlib.md5(f"{bluesky_url}:{size}".encode("utf-8")).hexdigest()}"'
814 conditional_response = get_conditional_response(request, etag=etag)
815 if conditional_response:
816 conditional_response["Etag"] = etag
817 conditional_response["Vary"] = ""
818 return conditional_response
820 try:
821 if cache.get(bluesky_url) == "err":
822 logger.warning(
823 f"Cached Bluesky fetch failed with URL error: {bluesky_url}"
824 )
825 avatar_metrics.record_external_request("bluesky", 0) # Cached error
826 return redir_default(default)
828 # EXPERIMENTAL: cache resized payload like gravatarproxy
829 _bs_alias = getattr(settings, "AVATAR_CACHE_ALIAS", "default")
830 _bs_img_key = (
831 f"bluesky_proxy_img:{hashlib.sha256(bluesky_url.encode('utf-8')).hexdigest()}:{size}"
832 )
833 if _bs_cached := caches[_bs_alias].get(_bs_img_key):
834 response = HttpResponse(
835 _bs_cached["content"], content_type=_bs_cached["content_type"]
836 )
837 response["Etag"] = etag
838 response["Cache-Control"] = f"public, max-age={bs_cache_timeout}"
839 response["Vary"] = ""
840 return response
842 blueskyimagedata = urlopen(bluesky_url, timeout=bs_image_timeout)
843 avatar_metrics.record_external_request("bluesky", 200)
844 except HTTPError as exc:
845 if exc.code not in [404, 503]:
846 print(
847 f"Bluesky fetch failed with an unexpected {exc.code} HTTP error: {bluesky_url}"
848 )
849 avatar_metrics.record_external_request("bluesky", exc.code)
850 cache.set(bluesky_url, "err", 30)
851 return redir_default(default)
852 except URLError as exc:
853 logger.warning(f"Bluesky fetch failed with URL error: {exc.reason}")
854 avatar_metrics.record_external_request("bluesky", 0) # Network error
855 cache.set(bluesky_url, "err", 30)
856 return redir_default(default)
857 except SSLError as exc:
858 logger.warning(f"Bluesky fetch failed with SSL error: {exc.reason}")
859 avatar_metrics.record_external_request("bluesky", 0) # SSL error
860 cache.set(bluesky_url, "err", 30)
861 return redir_default(default)
862 try:
863 data = BytesIO(blueskyimagedata.read())
864 img = Image.open(data)
865 img_format = img.format
866 if max(img.size) > size:
867 aspect = img.size[0] / float(img.size[1])
868 if aspect > 1:
869 new_size = (size, int(size / aspect))
870 else:
871 new_size = (int(size * aspect), size)
872 img = img.resize(new_size, get_resample_filter(size, img.size))
873 data = BytesIO()
874 img.save(data, format=img_format)
876 data.seek(0)
877 _bs_payload = data.read()
878 _bs_content_type = f"image/{file_format(img_format)}"
879 caches[_bs_alias].set(
880 _bs_img_key,
881 {"content": _bs_payload, "content_type": _bs_content_type},
882 bs_cache_timeout,
883 )
884 response = HttpResponse(_bs_payload, content_type=_bs_content_type)
885 response["Etag"] = etag
886 response["Cache-Control"] = f"public, max-age={bs_cache_timeout}"
887 # Remove Vary header for images since language doesn't matter
888 response["Vary"] = ""
889 return response
890 except ValueError as exc:
891 logger.error(f"Value error: {exc}")
892 return redir_default(default)
894 # We shouldn't reach this point... But make sure we do something
895 return redir_default(default)
898STATS_CACHE_KEY = "stats:view:payload"
901class StatsView(TemplateView, JsonResponse):
902 """
903 Return stats
904 """
906 def get(
907 self, request, *args, **kwargs
908 ): # pylint: disable=too-many-branches,too-many-statements,too-many-locals,no-self-use,unused-argument,too-many-return-statements
909 # The payload is identical for all callers and expensive to build
910 # (full scans over the photo blob table), so serve it from cache.
911 stats_cache_timeout = getattr(settings, "STATS_CACHE_TIMEOUT", 3600)
912 with contextlib.suppress(Exception):
913 cached_stats = cache.get(STATS_CACHE_KEY)
914 if cached_stats is not None:
915 return JsonResponse(cached_stats)
917 retval = {
918 "users": User.objects.count(),
919 "mails": ConfirmedEmail.objects.count(),
920 "openids": ConfirmedOpenId.objects.count(), # pylint: disable=no-member
921 "unconfirmed_mails": UnconfirmedEmail.objects.count(), # pylint: disable=no-member
922 "unconfirmed_openids": UnconfirmedOpenId.objects.count(), # pylint: disable=no-member
923 "avatars": Photo.objects.count(), # pylint: disable=no-member
924 }
926 # Top 10 viewed avatars
927 top_photos = Photo.objects.select_related("stat").filter(
928 stat__isnull=False
929 ).order_by("-stat__access_count")[:10]
930 top_photos_data = []
931 for photo in top_photos:
932 # Find the associated email or openid with highest access count
933 associated_emails = photo.emails.select_related("stat").filter(
934 stat__isnull=False
935 ).order_by("-stat__access_count")
936 associated_openids = photo.openids.select_related("stat").filter(
937 stat__isnull=False
938 ).order_by("-stat__access_count")
940 # Get the one with highest access count
941 top_associated = None
942 if associated_emails and associated_openids:
943 if (
944 associated_emails[0].stat.access_count
945 >= associated_openids[0].stat.access_count
946 ):
947 top_associated = associated_emails[0]
948 else:
949 top_associated = associated_openids[0]
950 elif associated_emails:
951 top_associated = associated_emails[0]
952 elif associated_openids:
953 top_associated = associated_openids[0]
955 if top_associated:
956 if hasattr(top_associated, "email"):
957 # It's a ConfirmedEmail
958 top_photos_data.append(
959 {
960 "access_count": top_associated.stat.access_count,
961 "avatar_url": f"https://libravatar.org/avatar/{top_associated.digest_sha256}",
962 }
963 )
964 else:
965 # It's a ConfirmedOpenId
966 top_photos_data.append(
967 {
968 "access_count": top_associated.stat.access_count,
969 "avatar_url": f"https://libravatar.org/avatar/{top_associated.digest}",
970 }
971 )
973 retval["top_viewed_avatars"] = top_photos_data
975 # Top 10 queried email addresses
976 top_emails = ConfirmedEmail.objects.select_related("stat").filter(
977 stat__isnull=False
978 ).order_by("-stat__access_count")[:10]
979 top_emails_data = []
980 for email in top_emails:
981 top_emails_data.append(
982 {
983 "access_count": email.stat.access_count,
984 "avatar_url": f"https://libravatar.org/avatar/{email.digest_sha256}",
985 }
986 )
988 retval["top_queried_emails"] = top_emails_data
990 # Top 10 queried OpenIDs
991 top_openids = ConfirmedOpenId.objects.select_related("stat").filter(
992 stat__isnull=False
993 ).order_by("-stat__access_count")[:10]
994 top_openids_data = []
995 for openid in top_openids:
996 top_openids_data.append(
997 {
998 "access_count": openid.stat.access_count,
999 "avatar_url": f"https://libravatar.org/avatar/{openid.digest}",
1000 }
1001 )
1003 retval["top_queried_openids"] = top_openids_data
1005 # Photo format distribution
1006 from django.db.models import Count
1008 format_distribution = (
1009 Photo.objects.values("format")
1010 .annotate(count=Count("format"))
1011 .order_by("-count")
1012 )
1013 retval["photo_format_distribution"] = list(format_distribution)
1015 # User activity statistics
1016 users_with_multiple_photos = (
1017 User.objects.annotate(photo_count=Count("photo"))
1018 .filter(photo_count__gt=1)
1019 .count()
1020 )
1021 users_with_both_email_and_openid = (
1022 User.objects.filter(
1023 confirmedemail__isnull=False, confirmedopenid__isnull=False
1024 )
1025 .distinct()
1026 .count()
1027 )
1029 # Calculate average photos per user
1030 total_photos = Photo.objects.count()
1031 total_users = User.objects.count()
1032 avg_photos_per_user = total_photos / total_users if total_users > 0 else 0
1034 retval["user_activity"] = {
1035 "users_with_multiple_photos": users_with_multiple_photos,
1036 "users_with_both_email_and_openid": users_with_both_email_and_openid,
1037 "average_photos_per_user": round(avg_photos_per_user, 2),
1038 }
1040 # Bluesky handles statistics
1041 bluesky_emails = ConfirmedEmail.objects.filter(
1042 bluesky_handle__isnull=False
1043 ).count()
1044 bluesky_openids = ConfirmedOpenId.objects.filter(
1045 bluesky_handle__isnull=False
1046 ).count()
1047 total_bluesky_handles = bluesky_emails + bluesky_openids
1049 # Top Bluesky handles by access count
1050 retval["bluesky_handles"] = {
1051 "total_bluesky_handles": total_bluesky_handles,
1052 "bluesky_emails": bluesky_emails,
1053 "bluesky_openids": bluesky_openids,
1054 }
1056 # Average photo size statistics using raw SQL
1057 from django.db import connection
1059 with connection.cursor() as cursor:
1060 # SQL to calculate average photo size
1061 cursor.execute("""
1062 SELECT
1063 COUNT(*) as photo_count,
1064 AVG(LENGTH(data)) as avg_size_bytes
1065 FROM ivataraccount_photo
1066 WHERE data IS NOT NULL
1067 """)
1068 result = cursor.fetchone()
1070 if result and result[0] > 0:
1071 photo_count, avg_size_bytes = result
1072 # Convert to float in case database returns string
1073 avg_size_bytes = float(avg_size_bytes) if avg_size_bytes else 0
1074 avg_size_kb = round(avg_size_bytes / 1024, 2) if avg_size_bytes else 0
1075 avg_size_mb = (
1076 round(avg_size_bytes / (1024 * 1024), 2) if avg_size_bytes else 0
1077 )
1079 retval["photo_size_stats"] = {
1080 "average_size_bytes": (
1081 round(avg_size_bytes, 2) if avg_size_bytes else 0
1082 ),
1083 "average_size_kb": avg_size_kb,
1084 "average_size_mb": avg_size_mb,
1085 "total_photos_analyzed": photo_count,
1086 }
1087 else:
1088 retval["photo_size_stats"] = {
1089 "average_size_bytes": 0,
1090 "average_size_kb": 0,
1091 "average_size_mb": 0,
1092 "total_photos_analyzed": 0,
1093 }
1095 # For potential duplicate photos, we'll check for photos with the same format and size
1096 # Note: This is not definitive - different images can have the same format and size
1097 # but it's a good indicator of potential duplicates that might warrant investigation
1098 with connection.cursor() as cursor:
1099 cursor.execute("""
1100 SELECT
1101 format,
1102 LENGTH(data) as file_size,
1103 COUNT(*) as count
1104 FROM ivataraccount_photo
1105 WHERE data IS NOT NULL
1106 GROUP BY format, LENGTH(data)
1107 HAVING COUNT(*) > 1
1108 ORDER BY count DESC
1109 LIMIT 10
1110 """)
1111 duplicate_groups = cursor.fetchall()
1113 total_potential_duplicate_photos = sum(
1114 group[2] for group in duplicate_groups
1115 )
1117 # Convert to list of dictionaries for JSON serialization
1118 duplicate_groups_detail = [
1119 {"format": group[0], "file_size": group[1], "count": group[2]}
1120 for group in duplicate_groups
1121 ]
1123 retval["potential_duplicate_photos"] = {
1124 "potential_duplicate_groups": len(duplicate_groups),
1125 "total_potential_duplicate_photos": total_potential_duplicate_photos,
1126 "potential_duplicate_groups_detail": duplicate_groups_detail,
1127 "note": "Potential duplicates are identified by matching file format and size - not definitive duplicates",
1128 }
1130 # Cache statistics for avatar generators
1131 from .robohash import OptimizedRobohash
1132 import ivatar.robohash as robohash_mod
1133 from .pagan_optimized import OptimizedPagan
1134 import ivatar.pagan_optimized as pagan_mod
1136 robohash_stats = OptimizedRobohash.get_cache_stats()
1137 robohash_png_cache = getattr(robohash_mod, "_robohash_png_cache", {})
1138 robohash_png_lock = getattr(robohash_mod, "_robohash_png_cache_lock", None)
1139 if robohash_png_lock:
1140 with robohash_png_lock:
1141 robohash_png_count = len(robohash_png_cache)
1142 else:
1143 robohash_png_count = len(robohash_png_cache)
1145 pagan_stats = OptimizedPagan.get_cache_stats()
1146 pagan_png_cache = getattr(pagan_mod, "_pagan_png_cache", {})
1147 pagan_png_lock = getattr(pagan_mod, "_pagan_png_cache_lock", None)
1148 if pagan_png_lock:
1149 with pagan_png_lock:
1150 pagan_png_count = len(pagan_png_cache)
1151 else:
1152 pagan_png_count = len(pagan_png_cache)
1154 with _default_avatars_png_cache_lock:
1155 default_avatars_cache_stats = {}
1156 for (style, _digest, _size) in _default_avatars_png_cache.keys():
1157 default_avatars_cache_stats[style] = default_avatars_cache_stats.get(style, 0) + 1
1159 retval["cache_stats"] = {
1160 "robohash": {
1161 "assembly_cache_items": robohash_stats["cache_size"],
1162 "assembly_cache_limit": robohash_stats["max_cache_size"],
1163 "png_cache_items": robohash_png_count,
1164 "png_cache_limit": getattr(robohash_mod, "_MAX_PNG_CACHE_SIZE", getattr(settings, "ROBOHASH_CACHE_SIZE", 150)),
1165 },
1166 "pagan": {
1167 "avatar_cache_items": pagan_stats["size"],
1168 "avatar_cache_limit": pagan_stats["max_size"],
1169 "png_cache_items": pagan_png_count,
1170 "png_cache_limit": getattr(pagan_mod, "_MAX_PAGAN_PNG_CACHE_SIZE", getattr(settings, "PAGAN_CACHE_SIZE", 100)),
1171 },
1172 "default_avatars": {
1173 "cache_items_total": sum(default_avatars_cache_stats.values()),
1174 "cache_limit_total": _MAX_DEFAULT_PNG_CACHE_SIZE,
1175 "cache_items_by_style": {
1176 "monsterid": default_avatars_cache_stats.get("monsterid", 0),
1177 "retro": default_avatars_cache_stats.get("retro", 0),
1178 "identicon": default_avatars_cache_stats.get("identicon", 0),
1179 "mmng": default_avatars_cache_stats.get("mmng", 0),
1180 "wavatar": default_avatars_cache_stats.get("wavatar", 0),
1181 },
1182 },
1183 }
1185 with contextlib.suppress(Exception):
1186 cache.set(STATS_CACHE_KEY, retval, stats_cache_timeout)
1188 return JsonResponse(retval)
1191# Thread-safe version cache - cached indefinitely since container restarts on changes
1192_version_cache = None
1193_version_cache_lock = threading.Lock()
1196def _get_git_info_from_files():
1197 """
1198 Safely extract git information from .git files without subprocess calls
1199 """
1200 try:
1201 # Get the project root directory
1202 project_root = path.dirname(path.dirname(path.abspath(__file__)))
1203 git_dir = path.join(project_root, ".git")
1205 if not path.exists(git_dir):
1206 return None
1208 # Read HEAD to get current branch/commit
1209 head_file = path.join(git_dir, "HEAD")
1210 if not path.exists(head_file):
1211 return None
1213 with open(head_file) as f:
1214 head_content = f.read().strip()
1216 # Parse HEAD content
1217 if head_content.startswith("ref: "):
1218 # We're on a branch
1219 branch_ref = head_content[5:] # Remove 'ref: '
1220 branch_name = path.basename(branch_ref)
1222 # Read the commit hash from the ref
1223 ref_file = path.join(git_dir, branch_ref)
1224 if path.exists(ref_file):
1225 with open(ref_file) as f:
1226 commit_hash = f.read().strip()
1227 else:
1228 return None
1229 else:
1230 # Detached HEAD state
1231 commit_hash = head_content
1232 branch_name = "detached"
1234 # Try to get commit date from git log file (if available)
1235 # Optimize: read only the last line instead of entire file
1236 commit_date = None
1237 log_file = path.join(git_dir, "logs", "HEAD")
1238 if path.exists(log_file):
1239 try:
1240 with open(log_file, "rb") as f:
1241 # Seek to end and read backwards to find last line
1242 f.seek(0, 2) # Seek to end
1243 file_size = f.tell()
1245 # Read backwards in chunks to find the last line
1246 chunk_size = min(1024, file_size)
1247 f.seek(max(0, file_size - chunk_size))
1248 chunk = f.read().decode("utf-8", errors="ignore")
1250 # Find the last non-empty line
1251 lines = chunk.split("\n")
1252 last_line = None
1253 for line in reversed(lines):
1254 if line.strip():
1255 last_line = line.strip()
1256 break
1258 if last_line:
1259 # Git log format: <old_hash> <new_hash> <author> <timestamp> <timezone> <message>
1260 # The format uses spaces, not tabs
1261 parts = last_line.split()
1262 if len(parts) >= 6:
1263 # Extract timestamp and convert to readable date
1264 # Format: <old_hash> <new_hash> <author_name> <author_email> <timestamp> <timezone> <message>
1265 # We need to find the timestamp which is after the author email
1266 for i, part in enumerate(parts):
1267 if part.isdigit() and len(part) == 10: # Unix timestamp
1268 import datetime
1270 timestamp = int(part)
1271 commit_date = datetime.datetime.fromtimestamp(
1272 timestamp
1273 ).strftime("%Y-%m-%d %H:%M:%S %z")
1274 break
1275 except (ValueError, IndexError, UnicodeDecodeError):
1276 pass
1278 # Fallback: try to get date from commit object if available
1279 if not commit_date and len(commit_hash) == 40:
1280 try:
1281 commit_dir = path.join(git_dir, "objects", commit_hash[:2])
1282 commit_file = path.join(commit_dir, commit_hash[2:])
1283 if path.exists(commit_file):
1284 # This would require decompressing the git object, which is complex
1285 # For now, we'll use a placeholder
1286 commit_date = "unknown"
1287 except Exception:
1288 commit_date = "unknown"
1290 # Get deployment date from directory modification time
1291 # Use .git directory as it's updated during deployment (checkout/pull)
1292 deployment_date = None
1293 if path.exists(git_dir):
1294 try:
1295 import datetime
1297 mtime = path.getmtime(git_dir)
1298 deployment_date = datetime.datetime.fromtimestamp(mtime).strftime(
1299 "%Y-%m-%d %H:%M:%S %z"
1300 )
1301 except Exception:
1302 deployment_date = "unknown"
1304 return {
1305 "commit_hash": commit_hash,
1306 "short_hash": commit_hash[:7] if len(commit_hash) >= 7 else commit_hash,
1307 "branch": branch_name,
1308 "commit_date": commit_date or "unknown",
1309 "deployment_date": deployment_date or "unknown",
1310 "deployment_status": "active",
1311 "version": f"{branch_name}-{commit_hash[:7] if len(commit_hash) >= 7 else commit_hash}",
1312 }
1314 except Exception as exc:
1315 logger.warning(f"Failed to read git info from files: {exc}")
1316 return None
1319def _get_cached_version_info():
1320 """
1321 Get cached version information, loading it if not available
1322 Since containers restart on content changes, cache indefinitely
1323 """
1324 global _version_cache
1326 with _version_cache_lock:
1327 # If cache is None or contains an error, try to reload it
1328 if _version_cache is None or "error" in _version_cache:
1329 # First try to read from a baked-in version.json file
1330 # This is preferred for containerized deployments
1331 try:
1332 project_root = path.dirname(path.dirname(path.abspath(__file__)))
1333 version_file = path.join(project_root, "version.json")
1334 if path.exists(version_file):
1335 with open(version_file) as f:
1336 import json
1338 _version_cache = json.load(f)
1339 logger.info("Loaded version info from version.json")
1340 except Exception as exc:
1341 logger.warning(f"Failed to read version.json: {exc}")
1343 # If that fails, try to get version info from git files
1344 if _version_cache is None or "error" in _version_cache:
1345 _version_cache = _get_git_info_from_files()
1347 # If that also fails, return error (but don't cache it indefinitely if we can help it)
1348 if _version_cache is None:
1349 _version_cache = {
1350 "error": "Unable to determine version - .git directory or version.json not found",
1351 "deployment_status": "unknown",
1352 }
1354 return _version_cache
1357class DeploymentVersionView(View):
1358 """
1359 View to return deployment version information for CI/CD verification
1360 Uses cached version info to prevent DDoS attacks and improve performance
1361 """
1363 def get(self, request, *args, **kwargs):
1364 """
1365 Return cached deployment version information including application version
1366 """
1367 from django.conf import settings
1369 version_info = _get_cached_version_info()
1371 if "error" in version_info:
1372 # Even on error, include the application version if available
1373 try:
1374 version_info["application_version"] = getattr(
1375 settings, "IVATAR_VERSION", "unknown"
1376 )
1377 except Exception:
1378 pass
1379 return JsonResponse(version_info, status=500)
1381 # Add application version to the response
1382 try:
1383 version_info["application_version"] = getattr(
1384 settings, "IVATAR_VERSION", "unknown"
1385 )
1386 except Exception:
1387 version_info["application_version"] = "unknown"
1389 return JsonResponse(version_info)
1390import time
1391import hashlib
1392from django.contrib.auth.mixins import LoginRequiredMixin
1393from django.views.generic import TemplateView
1394from django.test import RequestFactory
1396class PerfTestView(LoginRequiredMixin, TemplateView):
1397 template_name = "perf_test.html"
1399 def get_context_data(self, **kwargs):
1400 context = super().get_context_data(**kwargs)
1402 email = "test@example.com"
1403 md5_digest = hashlib.md5(email.encode()).hexdigest()
1404 sha256_digest = hashlib.sha256(email.encode()).hexdigest()
1406 factory = RequestFactory()
1407 view = AvatarImageView.as_view()
1409 results = []
1410 total_duration_ms = 0.0
1412 scenarios = [
1413 {"name": "Dummy MD5, Size 80", "digest": md5_digest, "params": {"s": 80}},
1414 {"name": "Dummy MD5, Size 512", "digest": md5_digest, "params": {"s": 512}},
1415 {"name": "Dummy SHA256, Size 80", "digest": sha256_digest, "params": {"s": 80}},
1416 {"name": "Dummy SHA256, Size 512", "digest": sha256_digest, "params": {"s": 512}},
1417 ]
1419 default_types = ['robohash', 'wavatar', 'monsterid', 'identicon', 'mm', 'mp', 'retro', 'pagan']
1420 for d in default_types:
1421 scenarios.append({
1422 "name": f"Dummy MD5, Size 80, Default: {d}",
1423 "digest": md5_digest,
1424 "params": {"s": 80, "d": d, "f": "y"}
1425 })
1427 from ivatar.ivataraccount.models import ConfirmedEmail, ConfirmedOpenId
1428 for email in ConfirmedEmail.objects.filter(user=self.request.user):
1429 scenarios.append({
1430 "name": f"User Email MD5 ({email.email}), Size 80",
1431 "digest": email.digest,
1432 "params": {"s": 80}
1433 })
1434 scenarios.append({
1435 "name": f"User Email SHA256 ({email.email}), Size 512",
1436 "digest": email.digest_sha256,
1437 "params": {"s": 512}
1438 })
1439 for d in default_types:
1440 scenarios.append({
1441 "name": f"User Email SHA256 ({email.email}), Size 80, Default: {d}",
1442 "digest": email.digest_sha256,
1443 "params": {"s": 80, "d": d, "f": "y"}
1444 })
1446 for openid in ConfirmedOpenId.objects.filter(user=self.request.user):
1447 scenarios.append({
1448 "name": f"User OpenID SHA256, Size 80",
1449 "digest": openid.digest,
1450 "params": {"s": 80}
1451 })
1452 scenarios.append({
1453 "name": f"User OpenID SHA256, Size 512",
1454 "digest": openid.digest,
1455 "params": {"s": 512}
1456 })
1457 for d in default_types:
1458 scenarios.append({
1459 "name": f"User OpenID SHA256, Size 80, Default: {d}",
1460 "digest": openid.digest,
1461 "params": {"s": 80, "d": d, "f": "y"}
1462 })
1464 for sc in scenarios:
1465 query_string = "&".join([f"{k}={v}" for k, v in sc["params"].items()])
1466 url = f"/avatar/{sc['digest']}?{query_string}"
1467 request = factory.get(url)
1469 start_time = time.time()
1470 try:
1471 response = view(request, digest=sc['digest'])
1472 status = response.status_code
1473 except Exception as e:
1474 status = str(e)
1475 end_time = time.time()
1477 duration_ms = (end_time - start_time) * 1000
1478 total_duration_ms += duration_ms
1480 results.append({
1481 "name": sc["name"],
1482 "digest": sc["digest"],
1483 "params": sc["params"],
1484 "duration_ms": round(duration_ms, 2),
1485 "duration_str": f"{duration_ms:.2f}",
1486 "status": status,
1487 })
1489 context['perf_results'] = results
1490 context['total_duration_str'] = f"{total_duration_ms:.2f}"
1491 return context