Coverage for app/venv/lib/python3.14/site-packages/weblate/accounts/views.py: 23%
986 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5from __future__ import annotations
7import os
8import re
9from base64 import b32encode
10from binascii import unhexlify
11from collections import defaultdict
12from datetime import timedelta
13from typing import TYPE_CHECKING, Any, ClassVar
14from urllib.parse import quote
16import qrcode
17import qrcode.image.svg
18import social_django.utils
19from django.conf import settings
20from django.contrib.auth import REDIRECT_FIELD_NAME, get_backends
21from django.contrib.auth import login as auth_login
22from django.contrib.auth import logout as auth_logout
23from django.contrib.auth.decorators import login_required
24from django.contrib.auth.views import LoginView, RedirectURLMixin
25from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError
26from django.core.mail.message import EmailMessage
27from django.core.signing import BadSignature, SignatureExpired, TimestampSigner
28from django.db import transaction
29from django.db.models import Count, Q
30from django.http import Http404, HttpResponse, HttpResponseRedirect, JsonResponse
31from django.http.response import HttpResponseServerError
32from django.middleware.csrf import rotate_token
33from django.shortcuts import get_object_or_404, redirect, render
34from django.template.loader import render_to_string
35from django.urls import reverse
36from django.utils import timezone
37from django.utils.cache import patch_response_headers
38from django.utils.decorators import method_decorator
39from django.utils.functional import cached_property
40from django.utils.http import urlencode
41from django.utils.safestring import mark_safe
42from django.utils.translation import gettext, gettext_lazy
43from django.views.decorators.cache import never_cache
44from django.views.decorators.csrf import csrf_exempt
45from django.views.decorators.http import require_POST
46from django.views.generic import (
47 DetailView,
48 FormView,
49 ListView,
50 TemplateView,
51 UpdateView,
52 View,
53)
54from django_otp import DEVICE_ID_SESSION_KEY
55from django_otp import login as otp_login
56from django_otp.models import Device
57from django_otp.plugins.otp_static.models import StaticDevice, StaticToken
58from django_otp.plugins.otp_totp.models import TOTPDevice
59from django_otp.util import random_hex
60from django_otp_webauthn.exceptions import OTPWebAuthnApiError
61from django_otp_webauthn.models import WebAuthnCredential
62from django_otp_webauthn.views import (
63 BeginCredentialAuthenticationView,
64 CompleteCredentialAuthenticationView,
65)
66from rest_framework.authtoken.models import Token
67from social_core.actions import do_auth
68from social_core.backends.base import BaseAuth
69from social_core.exceptions import (
70 AuthAlreadyAssociated,
71 AuthCanceled,
72 AuthException,
73 AuthFailed,
74 AuthForbidden,
75 AuthMissingParameter,
76 AuthStateForbidden,
77 AuthStateMissing,
78 AuthTokenError,
79 AuthUnreachableProvider,
80 InvalidEmail,
81 MissingBackend,
82)
83from social_django.utils import load_backend, load_strategy
84from social_django.views import complete, disconnect
86from weblate.accounts.auth import WeblateUserBackend
87from weblate.accounts.avatar import get_avatar_image, get_fallback_avatar_url
88from weblate.accounts.forms import (
89 CommitForm,
90 ContactForm,
91 DashboardSettingsForm,
92 EmailForm,
93 EmptyConfirmForm,
94 GroupAddForm,
95 GroupRemoveForm,
96 LanguagesForm,
97 LoginForm,
98 NotificationForm,
99 OTPTokenForm,
100 PasswordConfirmForm,
101 ProfileForm,
102 RegistrationForm,
103 ResetForm,
104 SetPasswordForm,
105 SubscriptionForm,
106 TOTPDeviceForm,
107 TOTPTokenForm,
108 UserForm,
109 UserSearchForm,
110 UserSettingsForm,
111 WebAuthnTokenForm,
112)
113from weblate.accounts.models import AuditLog, Subscription, VerifiedEmail
114from weblate.accounts.notifications import (
115 NOTIFICATIONS,
116 NotificationFrequency,
117 NotificationScope,
118 get_email_headers,
119 send_notification_email,
120)
121from weblate.accounts.pipeline import EmailAlreadyAssociated, UsernameAlreadyAssociated
122from weblate.accounts.utils import (
123 SESSION_SECOND_FACTOR_SOCIAL,
124 SESSION_SECOND_FACTOR_TOTP,
125 SESSION_SECOND_FACTOR_USER,
126 SESSION_WEBAUTHN_AUDIT,
127 get_key_name,
128 lock_user,
129 remove_user,
130)
131from weblate.auth.forms import UserEditForm
132from weblate.auth.models import (
133 AuthenticatedHttpRequest,
134 Invitation,
135 User,
136 get_anonymous,
137)
138from weblate.auth.utils import format_address, get_auth_keys
139from weblate.logger import LOGGER
140from weblate.trans.models import Change, Component, Project, Suggestion, Translation
141from weblate.trans.models.component import translation_prefetch_tasks
142from weblate.trans.models.project import prefetch_project_flags
143from weblate.trans.util import redirect_next
144from weblate.utils import messages
145from weblate.utils.errors import add_breadcrumb, report_error
146from weblate.utils.ratelimit import check_rate_limit, session_ratelimit_post
147from weblate.utils.request import get_ip_address, get_user_agent
148from weblate.utils.stats import prefetch_stats
149from weblate.utils.token import get_token
150from weblate.utils.version import USER_AGENT
151from weblate.utils.views import get_paginator, parse_path
152from weblate.utils.zammad import ZammadError, submit_zammad_ticket
154if TYPE_CHECKING: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 from django.forms import Form
157 from weblate.accounts.forms import ProfileBaseForm
158 from weblate.accounts.types import DeviceType
159 from weblate.auth.models import AuthenticatedHttpRequest
161CONTACT_TEMPLATE = """
162Message from %(name)s <%(email)s>:
164%(message)s
165"""
168MESSAGE_TEMPLATE = """
169{message}
171--
172User: {username}
173IP address: {address}
174User agent: {agent}
175"""
177CONTACT_SUBJECTS = {
178 "lang": "New language request",
179 "reg": "Registration problems",
180 "hosting": "Commercial hosting",
181 "account": "Suspicious account activity",
182 "trial": "Trial extension request",
183}
185ANCHOR_RE = re.compile(r"^#[a-z]+$")
187NOTIFICATION_PREFIX_TEMPLATE = "notifications__{}"
189# Override python-social-auth User-Agent header
190settings.SOCIAL_AUTH_USER_AGENT = USER_AGENT
193class EmailSentView(TemplateView):
194 r"""Class for rendering "E-mail sent" page."""
196 template_name = "accounts/email-sent.html"
198 do_password_reset: bool
199 do_account_remove: bool
201 def get_context_data(self, **kwargs):
202 """Create context for rendering page."""
203 context = super().get_context_data(**kwargs)
204 context["validity"] = settings.AUTH_TOKEN_VALID // 3600
205 context["is_reset"] = False
206 context["is_remove"] = False
207 if self.do_password_reset:
208 context["title"] = gettext("Password reset")
209 context["is_reset"] = True
210 elif self.do_account_remove:
211 context["title"] = gettext("Remove account")
212 context["is_remove"] = True
213 else:
214 context["title"] = gettext("User registration")
216 return context
218 def get(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
219 if not request.session.get("registration-email-sent"):
220 return redirect("home")
222 self.do_password_reset = request.session["password_reset"]
223 self.do_account_remove = request.session["account_remove"]
225 # Remove session for not authenticated user here.
226 # It is no longer needed and will just cause problems
227 # with multiple registrations from single browser.
228 if not request.user.is_authenticated:
229 request.session.flush()
230 else:
231 request.session.pop("registration-email-sent")
233 return super().get(request, *args, **kwargs)
236def mail_admins_contact(
237 request: AuthenticatedHttpRequest,
238 *,
239 subject: str,
240 message: str,
241 context: dict[str, Any],
242 name: str,
243 email: str,
244 to: list[str],
245) -> None:
246 """Send a message to the admins, as defined by the ADMINS setting."""
247 LOGGER.info("contact form from %s", email)
248 subject = f"{settings.EMAIL_SUBJECT_PREFIX}{subject}"
249 body = MESSAGE_TEMPLATE.format(
250 message=message % context,
251 address=get_ip_address(request),
252 agent=get_user_agent(request),
253 username=request.user.username,
254 )
256 if not to and settings.ADMINS:
257 to = [a[1] for a in settings.ADMINS]
258 elif not settings.ADMINS:
259 messages.error(request, gettext("Could not send message to administrator."))
260 LOGGER.error("ADMINS not configured, cannot send message")
261 return
263 if settings.ZAMMAD_URL and len(to) == 1 and to[0].endswith("@weblate.org"):
264 try:
265 submit_zammad_ticket(
266 title=subject,
267 body=body,
268 name=name,
269 email=email,
270 )
271 except ZammadError as error:
272 messages.error(request, str(error))
273 except OSError:
274 report_error("Could not create ticket")
275 messages.error(
276 request, gettext("Could not open a ticket, please try again later.")
277 )
278 else:
279 messages.success(
280 request,
281 gettext("Your request has been sent, you will shortly hear from us."),
282 )
284 else:
285 headers = get_email_headers("contact")
286 sender = format_address(name, email)
288 if settings.CONTACT_FORM == "reply-to":
289 headers["Reply-To"] = sender
290 from_email = settings.DEFAULT_FROM_EMAIL
291 else:
292 from_email = sender
294 mail = EmailMessage(
295 subject=subject, body=body, to=to, from_email=from_email, headers=headers
296 )
298 mail.send(fail_silently=False)
300 messages.success(
301 request,
302 gettext("Your request has been sent, you will shortly hear from us."),
303 )
306def redirect_profile(page: str | None = None):
307 url = reverse("profile")
308 if page and ANCHOR_RE.match(page):
309 url = f"{url}{page}"
310 return HttpResponseRedirect(url)
313def get_notification_forms(request: AuthenticatedHttpRequest):
314 user = request.user
315 subscriptions: dict[tuple[NotificationScope, int, int], dict[str, int]] = (
316 defaultdict(dict)
317 )
318 initials: dict[tuple[NotificationScope, int, int], dict[str, Any]] = {}
319 key: tuple[NotificationScope, int, int]
321 # Ensure watched, admin and all scopes are visible
322 for needed in (
323 NotificationScope.SCOPE_WATCHED,
324 NotificationScope.SCOPE_ADMIN,
325 NotificationScope.SCOPE_ALL,
326 ):
327 key = (needed, -1, -1)
328 subscriptions[key] = {}
329 initials[key] = {"scope": needed, "project": None, "component": None}
330 active = (NotificationScope.SCOPE_WATCHED, -1, -1)
332 # Include additional scopes from request
333 if "notify_project" in request.GET:
334 try:
335 project = user.allowed_projects.get(pk=request.GET["notify_project"])
336 active = key = (NotificationScope.SCOPE_PROJECT, project.pk, -1)
337 subscriptions[key] = {}
338 initials[key] = {
339 "scope": NotificationScope.SCOPE_PROJECT,
340 "project": project,
341 "component": None,
342 }
343 except (ObjectDoesNotExist, ValueError):
344 pass
345 if "notify_component" in request.GET:
346 try:
347 component = Component.objects.filter_access(user).get(
348 pk=request.GET["notify_component"],
349 )
350 active = key = (NotificationScope.SCOPE_COMPONENT, -1, component.pk)
351 subscriptions[key] = {}
352 initials[key] = {
353 "scope": NotificationScope.SCOPE_COMPONENT,
354 "component": component,
355 }
356 except (ObjectDoesNotExist, ValueError):
357 pass
359 # Populate scopes from the database
360 for subscription in user.subscription_set.select_related("project", "component"):
361 key = (
362 subscription.scope,
363 subscription.project_id or -1,
364 subscription.component_id or -1,
365 )
366 subscriptions[key][subscription.notification] = subscription.frequency
367 initials[key] = {
368 "scope": subscription.scope,
369 "project": subscription.project,
370 "component": subscription.component,
371 }
373 # Generate forms
374 for i, details in enumerate(sorted(subscriptions.items())):
375 yield NotificationForm(
376 user=user,
377 show_default=i > 1,
378 removable=i > 2,
379 subscriptions=details[1],
380 is_active=details[0] == active,
381 initial=initials[details[0]],
382 prefix=NOTIFICATION_PREFIX_TEMPLATE.format(i),
383 data=request.POST if request.method == "POST" else None,
384 )
385 for i in range(len(subscriptions), 200):
386 prefix = NOTIFICATION_PREFIX_TEMPLATE.format(i)
387 if prefix + "-scope" in request.POST or i < len(subscriptions):
388 yield NotificationForm(
389 user=user,
390 show_default=i > 1,
391 removable=i > 2,
392 subscriptions={},
393 is_active=i == 0,
394 prefix=prefix,
395 data=request.POST,
396 initial=initials[details[0]],
397 )
400@never_cache
401@login_required
402def user_profile(request: AuthenticatedHttpRequest):
403 user = request.user
404 profile = user.profile
405 profile.fixup_profile(request)
407 form_classes: list[type[ProfileBaseForm | UserForm]] = [
408 LanguagesForm,
409 SubscriptionForm,
410 UserSettingsForm,
411 DashboardSettingsForm,
412 ProfileForm,
413 CommitForm,
414 UserForm,
415 ]
416 forms = [form.from_request(request) for form in form_classes]
417 forms.extend(get_notification_forms(request))
418 all_backends = get_auth_keys()
420 if request.method == "POST":
421 if all(form.is_valid() for form in forms):
422 # Save changes
423 for form in forms:
424 if hasattr(form, "audit"):
425 form.audit(request)
426 form.save()
428 messages.success(request, gettext("Your profile has been updated."))
430 # Redirect after saving (and possibly changing language)
431 return redirect_profile(request.POST.get("activetab"))
432 elif not user.has_usable_password() and "email" in all_backends:
433 messages.warning(request, render_to_string("accounts/password-warning.html"))
435 social = user.social_auth.all()
436 social_names = [assoc.provider for assoc in social]
437 new_backends = [
438 x for x in sorted(all_backends) if x == "email" or x not in social_names
439 ]
440 user_translation_ids = set(
441 Change.objects.filter(
442 user=user, timestamp__gte=timezone.now() - timedelta(days=90)
443 ).values_list("translation", flat=True)
444 )
445 license_components = (
446 Component.objects.filter_access(user)
447 .filter(translation__id__in=user_translation_ids)
448 .exclude(license="")
449 .prefetch(alerts=False)
450 .distinct()
451 .order_by("license")
452 )
454 return render(
455 request,
456 "accounts/profile.html",
457 {
458 "languagesform": forms[0],
459 "subscriptionform": forms[1],
460 "usersettingsform": forms[2],
461 "dashboardsettingsform": forms[3],
462 "profileform": forms[4],
463 "commitform": forms[5],
464 "userform": forms[6],
465 "notification_forms": forms[7:],
466 "all_forms": forms,
467 "user_groups": user.cached_groups,
468 "profile": profile,
469 "title": gettext("User profile"),
470 "licenses": license_components,
471 "associated": social,
472 "new_backends": new_backends,
473 "has_email_auth": "email" in all_backends,
474 "auditlog": user.auditlog_set.order()[:20],
475 "totp_keys": user.totpdevice_set.all(),
476 "webauthn_keys": user.webauthncredential_set.all(),
477 "recovery_keys_count": StaticToken.objects.filter(
478 device__user=user
479 ).count(),
480 "bootstrap_5": True,
481 },
482 )
485@login_required
486@session_ratelimit_post("remove")
487@never_cache
488def user_remove(request: AuthenticatedHttpRequest):
489 is_confirmation = "remove_confirm" in request.session
490 if is_confirmation:
491 if request.method == "POST":
492 remove_user(request.user, request)
493 rotate_token(request)
494 auth_logout(request)
495 messages.success(request, gettext("Your account has been removed."))
496 return redirect("home")
497 confirm_form = EmptyConfirmForm(request)
499 elif request.method == "POST":
500 confirm_form = PasswordConfirmForm(request, request.POST)
501 if confirm_form.is_valid():
502 store_userid(request, remove=True)
503 request.GET = {"email": request.user.email} # type: ignore[assignment]
504 AuditLog.objects.create(
505 request.user, request, "removal-request", **request.GET
506 )
507 return social_complete(request, "email")
508 else:
509 confirm_form = PasswordConfirmForm(request)
511 return render(
512 request,
513 "accounts/removal.html",
514 {"confirm_form": confirm_form, "is_confirmation": is_confirmation},
515 )
518@session_ratelimit_post("confirm")
519@never_cache
520def confirm(request: AuthenticatedHttpRequest):
521 details = request.session.get("reauthenticate")
522 if not details:
523 return redirect("home")
525 if request.method == "POST":
526 confirm_form = PasswordConfirmForm(
527 request, request.POST, user=User.objects.get(pk=details["user_pk"])
528 )
529 if confirm_form.is_valid():
530 request.session.pop("reauthenticate")
531 request.session["reauthenticate_done"] = True
532 return redirect("social:complete", backend=details["backend"])
533 else:
534 confirm_form = PasswordConfirmForm(request)
536 context = {"confirm_form": confirm_form}
537 context.update(details)
539 return render(request, "accounts/confirm.html", context)
542def get_initial_contact(request: AuthenticatedHttpRequest):
543 """Fill in initial contact form fields from request."""
544 initial = {}
545 if request.user.is_authenticated:
546 initial["name"] = request.user.full_name
547 initial["email"] = request.user.email
548 return initial
551@never_cache
552def contact(request: AuthenticatedHttpRequest):
553 if request.method == "POST":
554 form = ContactForm(
555 request=request,
556 hide_captcha=request.user.is_authenticated,
557 data=request.POST,
558 )
559 if not check_rate_limit("message", request):
560 messages.error(
561 request, gettext("Too many messages sent, please try again later.")
562 )
563 elif form.is_valid():
564 mail_admins_contact(
565 request,
566 subject=form.cleaned_data["subject"],
567 message=CONTACT_TEMPLATE,
568 context=form.cleaned_data,
569 name=form.cleaned_data["name"],
570 email=form.cleaned_data["email"],
571 to=settings.ADMINS_CONTACT,
572 )
573 return redirect("home")
574 else:
575 initial = get_initial_contact(request)
576 if request.GET.get("t") in CONTACT_SUBJECTS:
577 initial["subject"] = CONTACT_SUBJECTS[request.GET["t"]]
578 form = ContactForm(
579 request=request, hide_captcha=request.user.is_authenticated, initial=initial
580 )
582 return render(
583 request,
584 "accounts/contact.html",
585 {"form": form, "title": gettext("Contact")},
586 )
589@login_required
590@session_ratelimit_post("hosting")
591@never_cache
592def hosting(request: AuthenticatedHttpRequest):
593 """Form for hosting request."""
594 if not settings.OFFER_HOSTING:
595 return redirect("home")
597 from weblate.billing.models import Billing
599 billings = (
600 Billing.objects.for_user(request.user)
601 .filter(state=Billing.STATE_TRIAL)
602 .order_by("-payment", "expiry")
603 )
605 return render(
606 request,
607 "accounts/hosting.html",
608 {
609 "title": gettext("Hosting"),
610 "billings": billings,
611 },
612 )
615@login_required
616@session_ratelimit_post("trial")
617@never_cache
618def trial(request: AuthenticatedHttpRequest):
619 """Form for hosting request."""
620 if not settings.OFFER_HOSTING:
621 return redirect("home")
623 plan = request.POST.get("plan", "640k")
625 # Avoid frequent requests for a trial for same user
626 if plan != "libre" and request.user.auditlog_set.filter(activity="trial").exists():
627 messages.error(
628 request,
629 gettext(
630 "Seems you've already requested a trial period recently. "
631 "Please contact us with your inquiry so we can find the "
632 "best solution for you."
633 ),
634 )
635 return redirect(reverse("contact") + "?t=trial")
637 if request.method == "POST":
638 from weblate.billing.models import Billing, BillingEvent, Plan
640 AuditLog.objects.create(request.user, request, "trial")
641 billing = Billing.objects.create(
642 plan=Plan.objects.get(slug=plan),
643 state=Billing.STATE_TRIAL,
644 expiry=timezone.now() + timedelta(days=14),
645 )
646 billing.billinglog_set.create(event=BillingEvent.CREATED, user=request.user)
647 billing.owners.add(request.user)
648 messages.info(
649 request,
650 gettext(
651 "Your trial period is now up and running; "
652 "create your translation project and start Weblating!"
653 ),
654 )
655 return redirect(reverse("create-project") + f"?billing={billing.pk}")
657 return render(request, "accounts/trial.html", {"title": gettext("Gratis trial")})
660class UserPage(UpdateView):
661 model = User
662 template_name = "accounts/user.html"
663 slug_field = "username"
664 slug_url_kwarg = "user"
665 context_object_name = "page_user"
666 form_class = UserEditForm
668 group_form = None
669 request: AuthenticatedHttpRequest
671 def post(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
672 if not request.user.has_perm("user.edit"):
673 raise PermissionDenied
674 user = self.object = self.get_object()
675 if "add_group" in request.POST:
676 self.group_form = GroupAddForm(request.POST)
677 if self.group_form.is_valid():
678 user.add_team(request, self.group_form.cleaned_data["add_group"])
679 return HttpResponseRedirect(self.get_success_url() + "#groups")
680 if "remove_group" in request.POST:
681 form = GroupRemoveForm(request.POST)
682 if form.is_valid():
683 user.remove_team(request, form.cleaned_data["remove_group"])
684 return HttpResponseRedirect(self.get_success_url() + "#groups")
685 if "remove_user" in request.POST:
686 remove_user(user, request, skip_notify=True)
687 return HttpResponseRedirect(self.get_success_url() + "#groups")
689 if "remove_2fa" in request.POST:
690 for device in user.profile.second_factors:
691 key_name = get_key_name(device)
692 device.delete()
693 AuditLog.objects.create(user, None, "twofactor-remove", device=key_name)
694 return HttpResponseRedirect(self.get_success_url() + "#edit")
696 if "disable_password" in request.POST:
697 lock_user(user, "admin-locked")
698 return HttpResponseRedirect(self.get_success_url() + "#edit")
700 return super().post(request, *args, **kwargs)
702 def get_queryset(self):
703 return super().get_queryset().select_related("profile")
705 def get_context_data(self, **kwargs):
706 """Create context for rendering page."""
707 context = super().get_context_data(**kwargs)
708 user = self.object
709 request = self.request
711 allowed_projects = request.user.allowed_projects
713 # Filter all user activity
714 all_changes = Change.objects.last_changes(request.user).filter(user=user)
716 # Filter where project is active
717 user_translation_ids = set(
718 all_changes.content()
719 .filter(timestamp__gte=timezone.now() - timedelta(days=90))
720 .values_list("translation", flat=True)
721 )
722 user_translations = (
723 Translation.objects.prefetch()
724 .filter(
725 id__in=list(user_translation_ids)[:10],
726 component__project__in=allowed_projects,
727 )
728 .order()
729 )
731 context["page_profile"] = user.profile
732 # Last user activity
733 context["last_changes"] = all_changes.recent()
734 context["last_changes_url"] = urlencode({"user": user.username})
735 context["page_user_translations"] = translation_prefetch_tasks(
736 prefetch_stats(user_translations)
737 )
738 owned = (user.owned_projects & allowed_projects.distinct()).order()[:11]
739 context["page_owned_projects_more"] = len(owned) == 11
740 context["page_owned_projects"] = prefetch_project_flags(
741 prefetch_stats(owned[:10])
742 )
743 watched = (user.watched_projects & allowed_projects).order()[:11]
744 context["page_watched_projects_more"] = len(watched) == 11
745 context["page_watched_projects"] = prefetch_project_flags(
746 prefetch_stats(watched[:10])
747 )
748 context["user_languages"] = user.profile.all_languages[:7]
749 context["group_form"] = self.group_form or GroupAddForm()
750 context["page_user_groups"] = (
751 user.groups.annotate(Count("user"))
752 .prefetch_related("defining_project")
753 .order()
754 )
755 context["bootstrap_5"] = True
756 return context
759def user_contributions(request: AuthenticatedHttpRequest, user: str):
760 page_user = get_object_or_404(User, username=user)
761 user_translation_ids = set(
762 Change.objects.content()
763 .filter(user=page_user)
764 .values_list("translation", flat=True)
765 )
766 user_translations = (
767 Translation.objects.filter_access(request.user)
768 .prefetch()
769 .filter(
770 id__in=user_translation_ids,
771 )
772 .order()
773 )
774 return render(
775 request,
776 "accounts/user_contributions.html",
777 {
778 "page_user": page_user,
779 "page_profile": page_user.profile,
780 "page_user_translations": translation_prefetch_tasks(
781 prefetch_stats(get_paginator(request, user_translations))
782 ),
783 },
784 )
787def user_avatar(request: AuthenticatedHttpRequest, user: str, size: int):
788 """User avatar view."""
789 allowed_sizes = (
790 # Used in top navigation
791 24,
792 # In text avatars
793 32,
794 # 80 pixels used when linked with weblate.org
795 80,
796 # Public profile
797 128,
798 )
799 if size not in allowed_sizes:
800 msg = f"Not supported size: {size}"
801 raise Http404(msg)
803 avatar_user = get_object_or_404(User, username=user)
804 email = avatar_user.email
806 # Deleted users
807 if email == f"noreply+{avatar_user.pk}@weblate.org":
808 return redirect(
809 os.path.join(settings.STATIC_URL, "state/ghost.svg"), permanent=True
810 )
811 # Bot and anonymous accounts
812 if not email or (email.startswith("noreply") and email.endswith("@weblate.org")):
813 return redirect(get_fallback_avatar_url(int(size)), permanent=True)
814 # Project API tokens
815 if email.endswith("@bots.noreply.weblate.org"):
816 return redirect(get_fallback_avatar_url(int(size), "api"), permanent=True)
818 response = HttpResponse(
819 content_type="image/png", content=get_avatar_image(avatar_user, size)
820 )
822 patch_response_headers(response, 3600 * 24 * 7)
824 return response
827def redirect_single(request: AuthenticatedHttpRequest, backend: str):
828 """Redirect user to single authentication backend."""
829 return render(
830 request,
831 "accounts/redirect.html",
832 {"backend": backend, "next": request.GET.get("next")},
833 )
836class BaseLoginView(LoginView):
837 def form_invalid(self, form):
838 rotate_token(self.request)
839 return super().form_invalid(form)
841 def form_valid(self, form):
842 """Security check complete. Log the user in."""
843 user = form.get_user()
844 if user.profile.has_2fa:
845 # Store session indication for second factor
846 self.request.session[SESSION_SECOND_FACTOR_USER] = (user.id, user.backend)
847 # Redirect to second factor login
848 redirect_to = self.request.POST.get(
849 self.redirect_field_name, self.request.GET.get(self.redirect_field_name)
850 )
851 login_params: dict[str, str] = {}
852 if redirect_to:
853 login_params[self.redirect_field_name] = redirect_to
854 login_url = reverse(
855 "2fa-login", kwargs={"backend": user.profile.get_second_factor_type()}
856 )
857 return HttpResponseRedirect(f"{login_url}?{urlencode(login_params)}")
858 auth_login(self.request, user)
859 return HttpResponseRedirect(self.get_success_url())
862class WeblateLoginView(BaseLoginView):
863 """Login handler, just a wrapper around standard Django login."""
865 form_class = LoginForm # type: ignore[assignment]
866 template_name = "accounts/login.html"
867 redirect_authenticated_user = True
869 @cached_property
870 def has_email_auth(self) -> bool:
871 return "email" in get_auth_keys()
873 @cached_property
874 def show_login_form(self) -> bool:
875 return self.has_email_auth or any(
876 not isinstance(backend, (BaseAuth, WeblateUserBackend))
877 for backend in get_backends()
878 )
880 def get_context_data(self, **kwargs):
881 context = super().get_context_data(**kwargs)
882 auth_backends = get_auth_keys()
883 context["login_backends"] = [x for x in sorted(auth_backends) if x != "email"]
884 context["can_reset"] = self.has_email_auth
885 # Show login form for e-mail login or any third-party Django auth backend such as LDAP
886 context["show_login_form"] = self.show_login_form
887 context["title"] = gettext("Sign in")
888 return context
890 @method_decorator(never_cache)
891 def dispatch(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
892 # Redirect signed in users to profile
893 if request.user.is_authenticated:
894 return redirect_profile()
896 # Redirect if there is only one backend
897 auth_backends = get_auth_keys()
898 if len(auth_backends) == 1 and "email" not in auth_backends:
899 return redirect_single(request, auth_backends.pop())
901 if request.method == "POST" and not self.show_login_form:
902 return redirect("login")
904 return super().dispatch(request, *args, **kwargs)
907class WeblateLogoutView(TemplateView):
908 """
909 Logout handler, just a reimplementation of standard Django logout.
911 - no redirect support
912 - login_required decorator
913 """
915 http_method_names = ["post", "options"] # noqa: RUF012
916 template_name = "registration/logged_out.html"
917 request: AuthenticatedHttpRequest
919 @method_decorator(require_POST)
920 @method_decorator(login_required)
921 @method_decorator(never_cache)
922 def post(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
923 """Logout may be done via POST."""
924 auth_logout(request)
925 request.user = get_anonymous()
926 return super().get(request, *args, **kwargs)
928 def get_context_data(self, **kwargs):
929 context = super().get_context_data(**kwargs)
930 context["title"] = gettext("Signed out")
931 return context
934def fake_email_sent(request: AuthenticatedHttpRequest, reset: bool = False):
935 """Fake redirect to e-mail sent page."""
936 request.session["registration-email-sent"] = True
937 request.session["password_reset"] = reset
938 request.session["account_remove"] = False
939 return redirect("email-sent")
942@never_cache
943def register(request: AuthenticatedHttpRequest):
944 """Registration form."""
945 # Fetch invitation
946 invitation: Invitation | None = None
947 initial = {}
948 if invitation_pk := request.session.get("invitation_link"):
949 try:
950 invitation = Invitation.objects.get(pk=invitation_pk)
951 except Invitation.DoesNotExist:
952 del request.session["invitation_link"]
953 else:
954 initial["email"] = invitation.email
955 initial["username"] = invitation.username
956 initial["fullname"] = invitation.full_name
958 # Allow registration at all?
959 registration_open = settings.REGISTRATION_OPEN or bool(invitation)
961 # Hide captcha for invitations
962 hide_captcha = invitation is not None
964 # Get list of allowed backends
965 backends = get_auth_keys()
966 if settings.REGISTRATION_ALLOW_BACKENDS and not invitation:
967 backends &= set(settings.REGISTRATION_ALLOW_BACKENDS)
968 elif not registration_open:
969 backends = set()
971 if request.method == "POST" and "email" in backends:
972 form = RegistrationForm(
973 request=request, data=request.POST, hide_captcha=hide_captcha
974 )
975 if form.is_valid():
976 if form.cleaned_data["email_user"]:
977 AuditLog.objects.create(
978 form.cleaned_data["email_user"], request, "connect"
979 )
980 return fake_email_sent(request)
981 store_userid(request)
982 return social_complete(request, "email")
983 else:
984 form = RegistrationForm(
985 request=request, initial=initial, hide_captcha=hide_captcha
986 )
988 # Redirect if there is only one backend
989 if len(backends) == 1 and "email" not in backends and not invitation:
990 return redirect_single(request, backends.pop())
992 return render(
993 request,
994 "accounts/register.html",
995 {
996 "registration_email": "email" in backends,
997 "registration_backends": backends - {"email"},
998 "title": gettext("User registration"),
999 "form": form,
1000 "invitation": invitation,
1001 },
1002 )
1005@login_required
1006@never_cache
1007def email_login(request: AuthenticatedHttpRequest):
1008 """Connect e-mail."""
1009 if request.method == "POST":
1010 form = EmailForm(request=request, data=request.POST)
1011 if form.is_valid():
1012 email_user = form.cleaned_data["email_user"]
1013 if email_user and email_user != request.user:
1014 AuditLog.objects.create(
1015 form.cleaned_data["email_user"], request, "connect"
1016 )
1017 return fake_email_sent(request)
1018 store_userid(request)
1019 return social_complete(request, "email")
1020 else:
1021 form = EmailForm(request=request)
1023 return render(
1024 request,
1025 "accounts/email.html",
1026 {"title": gettext("Register e-mail"), "form": form},
1027 )
1030@login_required
1031@session_ratelimit_post("password")
1032@never_cache
1033def password(request: AuthenticatedHttpRequest):
1034 """Password change / set form."""
1035 do_change = True
1036 change_form = None
1037 usable = request.user.has_usable_password()
1039 if "email" not in get_auth_keys() and not usable:
1040 messages.error(
1041 request,
1042 gettext("Cannot reset password, e-mail authentication is turned off."),
1043 )
1044 return redirect("profile")
1046 if usable:
1047 if request.method == "POST":
1048 change_form = PasswordConfirmForm(request, request.POST)
1049 do_change = change_form.is_valid()
1050 else:
1051 change_form = PasswordConfirmForm(request)
1052 do_change = False
1054 if request.method == "POST":
1055 form = SetPasswordForm(request.user, request.POST)
1056 if form.is_valid() and do_change:
1057 # Clear flag forcing user to set password
1058 redirect_page = "#account"
1059 if "show_set_password" in request.session:
1060 del request.session["show_set_password"]
1061 redirect_page = ""
1063 # Change the password
1064 form.save(request)
1066 return redirect_profile(redirect_page)
1067 else:
1068 form = SetPasswordForm(request.user)
1070 return render(
1071 request,
1072 "accounts/password.html",
1073 {"title": gettext("Change password"), "change_form": change_form, "form": form},
1074 )
1077def reset_password_set(request: AuthenticatedHttpRequest):
1078 """Perform actual password reset."""
1079 user = User.objects.get(pk=request.session["perform_reset"])
1080 if user.has_usable_password():
1081 request.session.flush()
1082 request.session.set_expiry(None)
1083 messages.error(request, gettext("Password reset has been already completed."))
1084 return redirect("login")
1085 if request.method == "POST":
1086 form = SetPasswordForm(user, request.POST)
1087 if form.is_valid():
1088 request.session.set_expiry(None)
1089 form.save(request, delete_session=True)
1090 return redirect("login")
1091 else:
1092 form = SetPasswordForm(user)
1093 return render(
1094 request,
1095 "accounts/reset.html",
1096 {
1097 "title": gettext("Password reset"),
1098 "form": form,
1099 "second_stage": True,
1100 },
1101 )
1104def get_registration_hint(email: str) -> str | None:
1105 domain = email.rsplit("@", 1)[-1]
1106 return settings.REGISTRATION_HINTS.get(domain)
1109@never_cache
1110def reset_password(request: AuthenticatedHttpRequest):
1111 """Password reset handling."""
1112 if request.user.is_authenticated:
1113 return redirect_profile()
1114 if "email" not in get_auth_keys():
1115 messages.error(
1116 request,
1117 gettext("Cannot reset password, e-mail authentication is turned off."),
1118 )
1119 return redirect("login")
1121 # We're already in the reset phase
1122 if "perform_reset" in request.session:
1123 return reset_password_set(request)
1124 if request.method == "POST":
1125 form = ResetForm(request=request, data=request.POST)
1126 if form.is_valid():
1127 if form.cleaned_data["email_user"]:
1128 audit = AuditLog.objects.create(
1129 form.cleaned_data["email_user"], request, "reset-request"
1130 )
1131 if not audit.check_rate_limit(request):
1132 store_userid(request, reset=True)
1133 return social_complete(request, "email")
1134 else:
1135 email = form.cleaned_data["email"]
1136 send_notification_email(
1137 None,
1138 [email],
1139 "reset-nonexisting",
1140 context={
1141 "address": get_ip_address(request),
1142 "user_agent": get_user_agent(request),
1143 "registration_hint": get_registration_hint(email),
1144 },
1145 )
1146 return fake_email_sent(request, True)
1147 else:
1148 form = ResetForm(request=request)
1150 return render(
1151 request,
1152 "accounts/reset.html",
1153 {
1154 "title": gettext("Password reset"),
1155 "form": form,
1156 "second_stage": False,
1157 },
1158 )
1161@require_POST
1162@login_required
1163@session_ratelimit_post("reset_api")
1164def reset_api_key(request: AuthenticatedHttpRequest):
1165 """Reset user API key."""
1166 # Need to delete old token as key is primary key
1167 with transaction.atomic():
1168 Token.objects.filter(user=request.user).delete()
1169 Token.objects.create(user=request.user, key=get_token("wlu"))
1171 return redirect_profile("#api")
1174@require_POST
1175@login_required
1176@session_ratelimit_post("userdata")
1177def userdata(request: AuthenticatedHttpRequest):
1178 response = JsonResponse(request.user.profile.dump_data())
1179 response["Content-Disposition"] = 'attachment; filename="weblate.json"'
1180 return response
1183@require_POST
1184@login_required
1185def watch(request: AuthenticatedHttpRequest, path):
1186 user = request.user
1187 redirect_obj = obj = parse_path(request, path, (Component, Project))
1188 if isinstance(obj, Component):
1189 project = obj.project
1191 # Mute project level subscriptions
1192 mute_real(
1193 user, scope=NotificationScope.SCOPE_PROJECT, component=None, project=project
1194 )
1195 # Manually enable component level subscriptions
1196 for default_subscription in user.subscription_set.filter(
1197 scope=NotificationScope.SCOPE_WATCHED
1198 ):
1199 subscription, created = user.subscription_set.get_or_create(
1200 notification=default_subscription.notification,
1201 scope=NotificationScope.SCOPE_COMPONENT,
1202 component=obj,
1203 project=None,
1204 defaults={"frequency": default_subscription.frequency},
1205 )
1206 if not created and subscription.frequency != default_subscription.frequency:
1207 subscription.frequency = default_subscription.frequency
1208 subscription.save(update_fields=["frequency"])
1210 # Watch project
1211 obj = project
1212 user.profile.watched.add(obj)
1213 return redirect_next(request.GET.get("next"), redirect_obj)
1216@require_POST
1217@login_required
1218def unwatch(request: AuthenticatedHttpRequest, path):
1219 obj = parse_path(request, path, (Project,))
1220 request.user.profile.watched.remove(obj)
1221 request.user.subscription_set.filter(
1222 Q(project=obj) | Q(component__project=obj)
1223 ).delete()
1224 return redirect_next(request.GET.get("next"), obj)
1227def mute_real(user: User, **kwargs) -> None:
1228 for notification_cls in NOTIFICATIONS:
1229 if notification_cls.ignore_watched:
1230 continue
1231 try:
1232 subscription = user.subscription_set.get_or_create(
1233 notification=notification_cls.get_name(),
1234 defaults={"frequency": NotificationFrequency.FREQ_NONE},
1235 **kwargs,
1236 )[0]
1237 except Subscription.MultipleObjectsReturned:
1238 subscriptions = user.subscription_set.filter(
1239 notification=notification_cls.get_name(), **kwargs
1240 )
1241 # Remove extra subscriptions
1242 for subscription in subscriptions[1:]:
1243 subscription.delete()
1244 subscription = subscriptions[0]
1245 if subscription.frequency != NotificationFrequency.FREQ_NONE:
1246 subscription.frequency = NotificationFrequency.FREQ_NONE
1247 subscription.save(update_fields=["frequency"])
1250@require_POST
1251@login_required
1252def mute(request: AuthenticatedHttpRequest, path):
1253 obj = parse_path(request, path, (Component, Project))
1254 if isinstance(obj, Component):
1255 mute_real(
1256 request.user,
1257 scope=NotificationScope.SCOPE_COMPONENT,
1258 component=obj,
1259 project=None,
1260 )
1261 return redirect(
1262 "{}?notify_component={}#notifications".format(reverse("profile"), obj.pk)
1263 )
1264 mute_real(
1265 request.user, scope=NotificationScope.SCOPE_PROJECT, component=None, project=obj
1266 )
1267 return redirect(
1268 "{}?notify_project={}#notifications".format(reverse("profile"), obj.pk)
1269 )
1272class SuggestionView(ListView):
1273 paginate_by = 25
1274 model = Suggestion
1276 def get_queryset(self):
1277 if self.kwargs["user"] == "-":
1278 user = None
1279 else:
1280 user = get_object_or_404(User, username=self.kwargs["user"])
1281 return (
1282 Suggestion.objects.filter_access(self.request.user)
1283 .filter(user=user)
1284 .order()
1285 )
1287 def get_context_data(self, *, object_list=None, **kwargs):
1288 result = super().get_context_data(object_list=object_list, **kwargs)
1289 if self.kwargs["user"] == "-":
1290 user = User.objects.get(username=settings.ANONYMOUS_USER_NAME)
1291 else:
1292 user = get_object_or_404(User, username=self.kwargs["user"])
1293 result["page_user"] = user
1294 result["page_profile"] = user.profile
1295 return result
1298def store_userid(
1299 request: AuthenticatedHttpRequest, *, reset: bool = False, remove: bool = False
1300) -> None:
1301 """Store user ID in the session."""
1302 request.session["social_auth_user"] = request.user.pk
1303 request.session["password_reset"] = reset
1304 request.session["account_remove"] = remove
1307@require_POST
1308@login_required
1309def social_disconnect(
1310 request: AuthenticatedHttpRequest, backend: str, association_id: str | None = None
1311):
1312 """
1313 Disconnect social authentication.
1315 Wrapper around social_django.views.disconnect:
1317 - Requires POST (to avoid CSRF on auth)
1318 - Blocks disconnecting last entry
1319 """
1320 # Block removal of last social auth
1321 if request.user.social_auth.count() <= 1:
1322 messages.error(request, gettext("Could not remove user identity"))
1323 return redirect_profile("#account")
1325 # Block removal of last verified email
1326 verified = VerifiedEmail.objects.filter(social__user=request.user).exclude(
1327 social__provider=backend, social_id=association_id
1328 )
1329 if not verified.exists():
1330 messages.error(
1331 request,
1332 gettext("Add another identity by confirming your e-mail address first."),
1333 )
1334 return redirect_profile("#account")
1336 return disconnect(request, backend, association_id)
1339@never_cache
1340@require_POST
1341def social_auth(request: AuthenticatedHttpRequest, backend: str):
1342 """
1343 Social authentication endpoint.
1345 Wrapper around social_django.views.auth:
1347 - Incorporates modified social_djang.utils.psa
1348 - Requires POST (to avoid CSRF on auth)
1349 - Stores current user in session (to avoid CSRF upon completion)
1350 - Stores session ID in the request URL if needed
1351 """
1352 # Fill in idp in case it is not provided
1353 if backend == "saml" and "idp" not in request.GET:
1354 request.GET = request.GET.copy() # type: ignore[assignment]
1355 request.GET["idp"] = "weblate" # type: ignore[misc]
1356 store_userid(request)
1357 uri = reverse("social:complete", args=(backend,))
1358 request.social_strategy = load_strategy(request)
1359 try:
1360 request.backend = load_backend(request.social_strategy, backend, uri)
1361 except MissingBackend:
1362 msg = "Backend not found"
1363 raise Http404(msg) from None
1365 try:
1366 return do_auth(request.backend, redirect_name=REDIRECT_FIELD_NAME)
1367 except AuthException as error:
1368 report_error("Could not authenticate")
1369 messages.error(request, gettext("Could not authenticate: %s") % error)
1370 return redirect("login")
1373def auth_fail(request: AuthenticatedHttpRequest, message: str):
1374 messages.error(request, message)
1375 return redirect(reverse("login"))
1378def registration_fail(request: AuthenticatedHttpRequest, message: str):
1379 messages.error(request, gettext("Could not complete registration.") + " " + message)
1380 messages.info(
1381 request,
1382 gettext("Please check if you have already registered an account.")
1383 + " "
1384 + gettext(
1385 "You can also request a new password, if you have lost your credentials."
1386 ),
1387 )
1389 return redirect(reverse("login"))
1392def auth_redirect_token(request: AuthenticatedHttpRequest):
1393 return auth_fail(
1394 request,
1395 gettext(
1396 "Try registering again to verify your identity, "
1397 "the confirmation link probably expired."
1398 ),
1399 )
1402def auth_redirect_state(request: AuthenticatedHttpRequest):
1403 return auth_fail(
1404 request, gettext("Could not authenticate due to invalid session state.")
1405 )
1408def handle_missing_parameter(
1409 request: AuthenticatedHttpRequest, backend: str, error: AuthMissingParameter
1410):
1411 if backend != "email" and error.parameter == "email":
1412 messages = [
1413 gettext("Got no e-mail address from third party authentication service.")
1414 ]
1415 if "email" in get_auth_keys():
1416 # Show only if e-mail authentication is turned on
1417 messages.append(gettext("Please register using e-mail instead."))
1418 return auth_fail(request, " ".join(messages))
1419 if error.parameter in {"email", "user", "expires"}:
1420 return auth_redirect_token(request)
1421 if error.parameter in {"state", "code", "RelayState"}:
1422 return auth_redirect_state(request)
1423 if error.parameter == "disabled":
1424 return auth_fail(request, gettext("New registrations are turned off."))
1425 return None
1428@csrf_exempt
1429@never_cache
1430def social_complete(request: AuthenticatedHttpRequest, backend: str):
1431 """
1432 Social authentication completion endpoint.
1434 Wrapper around social_django.views.complete:
1436 - Handles backend errors gracefully
1437 - Intermediate page (autosubmitted by JavaScript) to avoid
1438 confirmations by bots
1439 """
1440 if (
1441 "partial_token" in request.GET
1442 and "verification_code" in request.GET
1443 and "confirm" not in request.GET
1444 ):
1445 return render(
1446 request,
1447 "accounts/token.html",
1448 {
1449 "partial_token": request.GET["partial_token"],
1450 "verification_code": request.GET["verification_code"],
1451 "backend": backend,
1452 },
1453 )
1454 try:
1455 response = complete(request, backend)
1456 except InvalidEmail:
1457 report_error("Could not register")
1458 return auth_redirect_token(request)
1459 except AuthMissingParameter as error:
1460 report_error("Could not register")
1461 result = handle_missing_parameter(request, backend, error)
1462 if result:
1463 return result
1464 raise
1465 except (AuthStateMissing, AuthStateForbidden):
1466 report_error("Could not register")
1467 return auth_redirect_state(request)
1468 except AuthFailed:
1469 report_error("Could not register")
1470 return auth_fail(
1471 request,
1472 gettext(
1473 "Could not authenticate, probably due to an expired token "
1474 "or connection error."
1475 ),
1476 )
1477 except AuthCanceled:
1478 report_error("Could not register")
1479 return auth_fail(request, gettext("Authentication cancelled."))
1480 except AuthForbidden:
1481 report_error("Could not register")
1482 return auth_fail(request, gettext("The server does not allow authentication."))
1483 except EmailAlreadyAssociated:
1484 return registration_fail(
1485 request,
1486 gettext(
1487 "The supplied e-mail address is already in use for another account."
1488 ),
1489 )
1490 except UsernameAlreadyAssociated:
1491 return registration_fail(
1492 request,
1493 gettext("The supplied username is already in use for another account."),
1494 )
1495 except AuthTokenError as error:
1496 report_error("Could not authenticate")
1497 return registration_fail(
1498 request,
1499 gettext("Authentication failed: %s") % error,
1500 )
1501 except AuthAlreadyAssociated:
1502 return registration_fail(
1503 request,
1504 gettext(
1505 "The supplied user identity is already in use for another account."
1506 ),
1507 )
1508 except AuthUnreachableProvider:
1509 return registration_fail(
1510 request,
1511 gettext("The authentication provider could not be reached."),
1512 )
1513 except ValidationError as error:
1514 report_error("Could not register")
1515 return registration_fail(request, str(error))
1517 # Finish second factor authentication
1518 if persistent_id := request.session.pop(DEVICE_ID_SESSION_KEY, None):
1519 device = Device.from_persistent_id(persistent_id)
1520 otp_login(request, device)
1522 return response
1525@login_required
1526@require_POST
1527def subscribe(request: AuthenticatedHttpRequest):
1528 if "onetime" in request.POST:
1529 component = Component.objects.get(pk=request.POST["component"])
1530 request.user.check_access_component(component)
1531 subscription = Subscription(
1532 user=request.user,
1533 notification=request.POST["onetime"],
1534 scope=NotificationScope.SCOPE_COMPONENT,
1535 frequency=NotificationFrequency.FREQ_INSTANT,
1536 project=component.project,
1537 component=component,
1538 onetime=True,
1539 )
1540 try:
1541 subscription.full_clean()
1542 subscription.save()
1543 except ValidationError:
1544 pass
1545 messages.success(request, gettext("Notification settings adjusted."))
1546 return redirect_profile("#notifications")
1549def unsubscribe(request: AuthenticatedHttpRequest):
1550 if "i" in request.GET:
1551 signer = TimestampSigner()
1552 try:
1553 subscription = Subscription.objects.get(
1554 pk=int(signer.unsign(request.GET["i"], max_age=24 * 3600))
1555 )
1556 subscription.frequency = NotificationFrequency.FREQ_NONE
1557 subscription.save(update_fields=["frequency"])
1558 messages.success(request, gettext("Notification settings adjusted."))
1559 except (BadSignature, SignatureExpired, Subscription.DoesNotExist):
1560 messages.error(
1561 request,
1562 gettext(
1563 "The notification change link is no longer valid, "
1564 "please sign in to configure notifications."
1565 ),
1566 )
1568 return redirect_profile("#notifications")
1571@csrf_exempt
1572@never_cache
1573def saml_metadata(request: AuthenticatedHttpRequest):
1574 if "social_core.backends.saml.SAMLAuth" not in settings.AUTHENTICATION_BACKENDS:
1575 raise Http404
1577 # Generate metadata
1578 complete_url = reverse("social:complete", args=("saml",))
1579 saml_backend = social_django.utils.load_backend(
1580 load_strategy(request), "saml", complete_url
1581 )
1582 metadata, errors = saml_backend.generate_metadata_xml()
1584 # Handle errors
1585 if errors:
1586 add_breadcrumb(category="auth", message="SAML errors", errors=errors)
1587 report_error("SAML metadata", level="error")
1588 return HttpResponseServerError(content=", ".join(errors))
1590 return HttpResponse(content=metadata, content_type="text/xml")
1593@method_decorator(login_required, name="dispatch")
1594class UserList(ListView):
1595 paginate_by = 50
1596 model = User
1597 form_class = UserSearchForm
1598 initial_query = ""
1600 def get_base_queryset(self):
1601 return User.objects.filter(is_active=True, is_bot=False)
1603 def get_queryset(self):
1604 users = self.get_base_queryset()
1605 form = self.form
1606 if form.is_valid():
1607 search = form.cleaned_data.get("q", "")
1608 if search:
1609 users = users.search(search, parser=form.fields["q"].parser)
1610 else:
1611 users = users.order()
1613 return users.order_by(self.sort_query)
1615 def setup(self, request: AuthenticatedHttpRequest, *args, **kwargs) -> None: # type: ignore[override]
1616 super().setup(request, *args, **kwargs)
1617 if "q" in request.GET:
1618 form = self.form_class(request.GET)
1619 else:
1620 form = self.form_class({"q": self.initial_query})
1621 self.form = form
1622 self.sort_query = ""
1623 if form.is_valid():
1624 self.sort_query = form.cleaned_data.get("sort_by", "")
1625 if not self.sort_query:
1626 self.sort_query = "-date_joined"
1628 def get_context_data(self, **kwargs):
1629 """Create context for rendering page."""
1630 context = super().get_context_data(**kwargs)
1631 context["form"] = self.form
1632 context["sort_query"] = self.sort_query
1633 context["sort_name"] = self.form.sort_choices[self.sort_query.strip("-")]
1634 context["sort_choices"] = self.form.sort_choices
1635 context["search_items"] = (
1636 ("q", self.form.cleaned_data.get("q", "").strip()),
1637 ("sort_by", self.sort_query),
1638 )
1639 context["query_string"] = urlencode(context["search_items"])
1640 return context
1643@method_decorator(login_required, name="dispatch")
1644class RecoveryCodesView(TemplateView):
1645 template_name = "accounts/recovery-codes.html"
1647 def post(self, request, *args, **kwargs):
1648 user = request.user
1649 # Delete all existing tokens
1650 StaticToken.objects.filter(device__user=user).delete()
1652 # Get device
1653 device = StaticDevice.objects.filter(user=user).first()
1654 if device is None:
1655 device = StaticDevice.objects.create(user=user, name="Backup Code")
1657 # Generate tokens
1658 for _i in range(16):
1659 token = StaticToken.random_token()
1660 device.token_set.create(token=token)
1662 AuditLog.objects.create(user, request, "recovery-generate")
1664 return redirect("recovery-codes")
1666 def get(self, request, *args, **kwargs):
1667 user = request.user
1668 AuditLog.objects.create(user, request, "recovery-show")
1669 return super().get(request, *args, **kwargs)
1671 def get_context_data(self, **kwargs):
1672 result = super().get_context_data(**kwargs)
1673 user = self.request.user
1674 recovery_codes = StaticToken.objects.filter(device__user=user).values_list(
1675 "token", flat=True
1676 )
1677 result["recovery_codes"] = recovery_codes
1678 result["recovery_codes_str"] = "\n".join(recovery_codes)
1679 return result
1682@method_decorator(login_required, name="dispatch")
1683class WebAuthnCredentialView(DetailView):
1684 model = WebAuthnCredential
1686 message_remove = gettext_lazy("The security key was removed.")
1687 message_add = gettext_lazy("The security key %s was registered.")
1689 def get_queryset(self):
1690 return super().get_queryset().filter(user=self.request.user)
1692 def get(self, request, *args, **kwargs):
1693 return redirect_profile("#account")
1695 def post(self, request, *args, **kwargs):
1696 obj = self.get_object()
1697 if "delete" in request.POST:
1698 key_name = get_key_name(obj)
1699 obj.delete()
1700 AuditLog.objects.create(
1701 request.user, request, "twofactor-remove", device=key_name
1702 )
1703 messages.success(request, self.message_remove)
1704 elif "name" in request.POST:
1705 obj.name = request.POST["name"]
1706 obj.save(update_fields=["name"])
1707 key_name = get_key_name(obj)
1708 audit_pk = request.session.pop(SESSION_WEBAUTHN_AUDIT, None)
1709 if isinstance(obj, WebAuthnCredential) and audit_pk is not None:
1710 audit = AuditLog.objects.get(pk=audit_pk)
1711 audit.params = {"device": key_name}
1712 audit.save(update_fields=["params"])
1713 messages.success(
1714 request,
1715 self.message_add % key_name,
1716 )
1717 return redirect_profile("#account")
1720@method_decorator(login_required, name="dispatch")
1721class TOTPDetailView(WebAuthnCredentialView):
1722 model = TOTPDevice
1724 message_remove = gettext_lazy("The authentication app was removed.")
1725 message_add = gettext_lazy("The authentication app %s was registered.")
1728@method_decorator(login_required, name="dispatch")
1729class TOTPView(FormView):
1730 template_name = "accounts/totp.html"
1731 form_class = TOTPDeviceForm
1732 session_key = SESSION_SECOND_FACTOR_TOTP
1734 @cached_property
1735 def totp_key(self) -> str:
1736 key = self.request.session.get(self.session_key, None)
1737 if key is None:
1738 key = random_hex(20)
1739 self.request.session[self.session_key] = key
1740 return key
1742 @cached_property
1743 def totp_key_b32(self) -> str:
1744 key = self.totp_key
1745 rawkey = unhexlify(key.encode("ascii"))
1746 return b32encode(rawkey).decode("utf-8")
1748 @cached_property
1749 def totp_url(self) -> str:
1750 # For a complete run-through of all the parameters, have a look at the
1751 # specs at:
1752 # https://github.com/google/google-authenticator/wiki/Key-Uri-Format
1754 accountname = self.request.user.username
1755 issuer = settings.SITE_TITLE
1756 label = f"{issuer}: {accountname}"
1758 # Ensure that the secret parameter is the FIRST parameter of the URI, this
1759 # allows Microsoft Authenticator to work.
1760 query = (
1761 ("secret", self.totp_key_b32),
1762 ("digits", "6"),
1763 ("issuer", issuer),
1764 )
1766 return f"otpauth://totp/{quote(label)}?{urlencode(query)}"
1768 @property
1769 def totp_svg(self):
1770 image = qrcode.make(self.totp_url, image_factory=qrcode.image.svg.SvgPathImage)
1771 return mark_safe(image.to_string(encoding="unicode")) # noqa: S308
1773 def get_context_data(self, **kwargs):
1774 """Create context for rendering page."""
1775 context = super().get_context_data(**kwargs)
1776 context["totp_key_b32"] = self.totp_key_b32
1777 context["totp_url"] = self.totp_url
1778 context["totp_svg"] = self.totp_svg
1779 return context
1781 def get_form_kwargs(self):
1782 kwargs = super().get_form_kwargs()
1783 kwargs["key"] = self.totp_key
1784 kwargs["user"] = self.request.user
1785 return kwargs
1787 def form_valid(self, form: TOTPDeviceForm):
1788 user = self.request.user
1789 device = form.save()
1790 AuditLog.objects.create(
1791 user,
1792 self.request,
1793 "twofactor-add",
1794 device=get_key_name(device),
1795 )
1796 if form.cleaned_data["remove_previous"]:
1797 for old in user.totpdevice_set.exclude(pk=device.pk):
1798 key_name = get_key_name(old)
1799 old.delete()
1800 AuditLog.objects.create(
1801 user, self.request, "twofactor-remove", device=key_name
1802 )
1804 return redirect_profile("#account")
1807class SecondFactorMixin(View):
1808 def second_factor_completed(self, device: Device) -> None:
1809 # Store audit log entry about used device and update last used device type
1810 user = self.get_user()
1811 user.profile.log_2fa(self.request, device)
1812 del self.request.session[SESSION_SECOND_FACTOR_USER]
1814 if not self.request.session.get(SESSION_SECOND_FACTOR_SOCIAL):
1815 # Keep login on social pipeline if we got here from it
1816 auth_login(self.request, user)
1817 # Perform OTP login
1818 otp_login(self.request, device)
1819 else:
1820 self.request.session[DEVICE_ID_SESSION_KEY] = device.persistent_id
1821 # This is completed in social_complete after completing social login
1823 def second_factor_failed(self) -> None:
1824 user = self.get_user()
1825 user.profile.log_2fa_failed(self.request, self.get_backend())
1827 def get_user(self) -> User:
1828 try:
1829 user_id, backend = self.request.session[SESSION_SECOND_FACTOR_USER]
1830 except KeyError as error:
1831 raise Http404 from error
1832 try:
1833 user = User.objects.get(pk=user_id)
1834 except User.DoesNotExist as error:
1835 raise Http404 from error
1836 user.backend = backend
1837 return user
1840class SecondFactorLoginView(SecondFactorMixin, RedirectURLMixin, FormView):
1841 template_name = "accounts/2fa.html"
1842 next_page = settings.LOGIN_REDIRECT_URL
1843 forms: ClassVar[dict[str, Form]] = {
1844 "totp": TOTPTokenForm,
1845 "webauthn": WebAuthnTokenForm,
1846 "recovery": OTPTokenForm,
1847 }
1849 def get_backend(self) -> DeviceType:
1850 backend = self.kwargs["backend"]
1851 if backend not in self.forms:
1852 raise Http404
1853 return backend
1855 def get_form_class(self):
1856 return self.forms[self.get_backend()]
1858 def get_form_kwargs(self):
1859 kwargs = super().get_form_kwargs()
1860 kwargs["user"] = self.user
1861 kwargs["request"] = self.request
1862 return kwargs
1864 def get_context_data(self, **kwargs):
1865 """Create context for rendering page."""
1866 context = super().get_context_data(**kwargs)
1867 context["second_factor_types"] = self.user.profile.second_factor_types - {
1868 self.get_backend()
1869 }
1870 context["next"] = self.request.GET.get("next", "")
1871 context["title"] = gettext("Second factor sign in")
1872 return context
1874 @method_decorator(session_ratelimit_post("second_factor"))
1875 def dispatch(self, request: AuthenticatedHttpRequest, *args, **kwargs): # type: ignore[override]
1876 self.user = self.get_user()
1877 return super().dispatch(request, *args, **kwargs)
1879 def form_valid(self, form):
1880 self.second_factor_completed(form.user.otp_device)
1882 return HttpResponseRedirect(self.get_success_url())
1884 def form_invalid(self, form):
1885 self.second_factor_failed()
1886 return super().form_invalid(form)
1889class WeblateBeginCredentialAuthenticationView(
1890 SecondFactorMixin, BeginCredentialAuthenticationView
1891):
1892 pass
1895@method_decorator(session_ratelimit_post("second_factor"), name="dispatch")
1896class WeblateCompleteCredentialAuthenticationView(
1897 SecondFactorMixin, CompleteCredentialAuthenticationView
1898):
1899 def complete_auth(self, device: WebAuthnCredential) -> User:
1900 return self.second_factor_completed(device)
1902 def post(self, *args, **kwargs):
1903 try:
1904 return super().post(*args, **kwargs)
1905 except OTPWebAuthnApiError:
1906 self.second_factor_failed()
1907 raise