Coverage for app/venv/lib/python3.14/site-packages/weblate/auth/models.py: 64%
611 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
4from __future__ import annotations
6import re
7import uuid
8from collections import defaultdict
9from collections.abc import (
10 Iterable,
11)
12from contextvars import ContextVar
13from functools import cache as functools_cache
14from itertools import chain
15from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict, cast
17import sentry_sdk
18from appconf import AppConf
19from django.conf import settings
20from django.contrib.auth.base_user import AbstractBaseUser, BaseUserManager
21from django.contrib.auth.hashers import make_password
22from django.contrib.auth.models import Group as DjangoGroup
23from django.core.validators import MinValueValidator
24from django.db import models
25from django.db.models import Prefetch, Q, UniqueConstraint
26from django.db.models.functions import Upper
27from django.db.models.signals import m2m_changed, post_save
28from django.dispatch import receiver
29from django.http import Http404, HttpRequest
30from django.urls import reverse
31from django.utils import timezone
32from django.utils.functional import cached_property
33from django.utils.translation import gettext, gettext_lazy, pgettext
35from weblate.auth.data import (
36 ACL_GROUPS,
37 GLOBAL_PERM_NAMES,
38 PERMISSION_NAMES,
39 SELECTION_ALL,
40 SELECTION_ALL_PROTECTED,
41 SELECTION_ALL_PUBLIC,
42 SELECTION_COMPONENT_LIST,
43 SELECTION_MANUAL,
44)
45from weblate.auth.permissions import SPECIALS, check_global_permission, check_permission
46from weblate.auth.utils import (
47 create_anonymous,
48 format_address,
49 is_django_permission,
50 migrate_groups,
51 migrate_permissions,
52 migrate_roles,
53)
54from weblate.lang.models import Language
55from weblate.trans.defines import FULLNAME_LENGTH, USERNAME_LENGTH
56from weblate.trans.fields import RegexField
57from weblate.trans.models import Component, ComponentList, Project
58from weblate.utils.decorators import disable_for_loaddata
59from weblate.utils.fields import EmailField, UsernameField
60from weblate.utils.search import parse_query
61from weblate.utils.validators import CRUD_RE, validate_fullname, validate_username
63if TYPE_CHECKING: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 from collections.abc import Iterable, Mapping
66 from django_otp.models import Device
67 from social_core.backends.base import BaseAuth
68 from social_django.models import DjangoStorage
70 from weblate.accounts.models import Subscription
71 from weblate.accounts.strategy import WeblateStrategy
72 from weblate.auth.results import PermissionResult
73 from weblate.wladmin.models import SupportStatusDict
75 SimplePermissionList = list[tuple[set[str], set[int] | None]]
77 # This is SimplePermissionList with additional None instead of permissions
78 # to indicate user block
79 PermissionList = list[tuple[set[str] | None, set[int] | None]]
81 PermissionCacheType = dict[int, PermissionList]
82 SimplePermissionCacheType = dict[int, SimplePermissionList]
84 class PermissionsDictType(TypedDict, total=False):
85 projects: PermissionCacheType
86 components: SimplePermissionCacheType
89class Permission(models.Model):
90 codename = models.CharField(max_length=100, unique=True)
91 name = models.CharField(max_length=200)
93 class Meta:
94 verbose_name = "Permission"
95 verbose_name_plural = "Permissions"
97 def __str__(self) -> str:
98 name = gettext(self.name)
99 if self.codename in GLOBAL_PERM_NAMES:
100 return gettext("%s (site-wide permission)") % name
101 return name
104class Role(models.Model):
105 name = models.CharField(
106 verbose_name=gettext_lazy("Name"), max_length=200, unique=True
107 )
108 permissions = models.ManyToManyField(
109 Permission,
110 verbose_name=gettext_lazy("Permissions"),
111 blank=True,
112 help_text=gettext_lazy("Choose permissions granted to this role."),
113 )
115 class Meta:
116 verbose_name = "Role"
117 verbose_name_plural = "Roles"
119 def __str__(self) -> str:
120 return pgettext("Access-control role", self.name)
123class GroupQuerySet(models.QuerySet["Group"]):
124 def order(self):
125 """Ordering in project scope by priority."""
126 return self.order_by("defining_project__name", "name")
129class Group(models.Model):
130 name = models.CharField(gettext_lazy("Name"), max_length=150)
131 roles = models.ManyToManyField(
132 Role,
133 verbose_name=gettext_lazy("Roles"),
134 blank=True,
135 help_text=gettext_lazy("Choose roles granted to this team."),
136 )
138 defining_project = models.ForeignKey(
139 "trans.Project",
140 related_name="defined_groups",
141 on_delete=models.deletion.CASCADE,
142 null=True,
143 blank=True,
144 )
146 project_selection = models.IntegerField(
147 verbose_name=gettext_lazy("Project selection"),
148 choices=(
149 (SELECTION_MANUAL, gettext_lazy("As defined")),
150 (SELECTION_ALL, gettext_lazy("All projects")),
151 (SELECTION_ALL_PUBLIC, gettext_lazy("All public projects")),
152 (SELECTION_ALL_PROTECTED, gettext_lazy("All protected projects")),
153 (SELECTION_COMPONENT_LIST, gettext_lazy("From component list")),
154 ),
155 default=SELECTION_MANUAL,
156 )
157 projects = models.ManyToManyField(
158 "trans.Project", verbose_name=gettext_lazy("Projects"), blank=True
159 )
160 components = models.ManyToManyField(
161 "trans.Component",
162 verbose_name=gettext_lazy("Components"),
163 blank=True,
164 help_text=gettext_lazy(
165 "Empty selection grants access to all components in project scope."
166 ),
167 )
168 componentlists = models.ManyToManyField(
169 "trans.ComponentList",
170 verbose_name=gettext_lazy("Component lists"),
171 blank=True,
172 )
174 language_selection = models.IntegerField(
175 verbose_name=gettext_lazy("Language selection"),
176 choices=(
177 (SELECTION_MANUAL, gettext_lazy("As defined")),
178 (SELECTION_ALL, gettext_lazy("All languages")),
179 ),
180 default=SELECTION_MANUAL,
181 )
182 languages = models.ManyToManyField(
183 "lang.Language", verbose_name=gettext_lazy("Languages"), blank=True
184 )
186 internal = models.BooleanField(
187 verbose_name=gettext_lazy("Internal Weblate team"), default=False
188 )
190 admins = models.ManyToManyField(
191 "weblate_auth.User",
192 verbose_name=gettext_lazy("Team administrators"),
193 blank=True,
194 help_text=gettext_lazy(
195 "The administrator can add or remove users from a team."
196 ),
197 related_name="administered_group_set",
198 )
199 enforced_2fa = models.BooleanField(
200 verbose_name=gettext_lazy("Enforced two-factor authentication"),
201 default=False,
202 help_text=gettext_lazy(
203 "Requires users to have two-factor authentication configured."
204 ),
205 )
207 objects = GroupQuerySet.as_manager()
209 class Meta:
210 verbose_name = "Group"
211 verbose_name_plural = "Groups"
213 def __str__(self) -> str:
214 if self.defining_project:
215 return pgettext("Per-project access-control team name", self.name)
216 return pgettext("Access-control team name", self.name)
218 def save(self, *args, **kwargs) -> None:
219 super().save(*args, **kwargs)
220 if self.language_selection == SELECTION_ALL:
221 self.languages.clear()
222 if self.project_selection in {
223 SELECTION_ALL,
224 SELECTION_ALL_PUBLIC,
225 SELECTION_ALL_PROTECTED,
226 }:
227 self.projects.clear()
228 elif self.project_selection == SELECTION_COMPONENT_LIST:
229 self.projects.set(
230 Project.objects.filter(
231 component__componentlist__in=self.componentlists.all()
232 ),
233 clear=True,
234 )
236 def get_absolute_url(self) -> str:
237 return reverse("team", kwargs={"pk": self.pk})
239 def long_name(self):
240 if self.defining_project:
241 return f"{self.defining_project} / {self}"
242 return str(self)
245bot_cache = ContextVar("bot_cache", default=dict)
248class UserManager(BaseUserManager["User"]):
249 def _create_user(self, username, email, password, **extra_fields):
250 """Create and save a User with the given fields."""
251 if not username: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 msg = "The given username must be set"
253 raise ValueError(msg)
254 email = self.normalize_email(email)
255 username = self.model.normalize_username(username)
256 user = self.model(username=username, email=email, **extra_fields)
257 user.set_password(password)
258 user.save(using=self._db)
259 return user
261 def create_user(self, username, email=None, password=None, **extra_fields):
262 extra_fields.setdefault("is_superuser", False)
263 return self._create_user(username, email, password, **extra_fields)
265 def create_superuser(self, username, email, password, **extra_fields):
266 extra_fields.setdefault("is_superuser", True)
268 if extra_fields.get("is_superuser") is not True:
269 msg = "Superuser must have is_superuser=True."
270 raise ValueError(msg)
272 return self._create_user(username, email, password, **extra_fields)
274 def get_or_create_bot(self, *, scope: str, name: str, verbose: str) -> User:
275 cached = bot_cache.get({})
276 username = f"{scope}:{name}"
277 try:
278 return cached[username]
279 except KeyError:
280 user = self.get_or_create(
281 username=username,
282 defaults={
283 "is_bot": True,
284 "full_name": verbose,
285 "email": f"noreply-{scope}-{name}@weblate.org",
286 "is_active": False,
287 "password": make_password(None),
288 },
289 )[0]
290 cached[username] = user
291 return user
294class UserQuerySet(models.QuerySet["User"]):
295 def having_perm(self, perm, project):
296 """
297 All users having explicit permission on a project.
299 Note: This intentionally does not list superusers or site-wide permissions
300 given using project_selection.
301 """
302 return self.filter(
303 groups__roles__permissions__codename=perm, groups__projects=project
304 ).distinct()
306 def all_admins(self, project):
307 """All admins in a project."""
308 return self.having_perm("project.edit", project)
310 def all_reviewers(self, project: Project) -> UserQuerySet:
311 """All reviewers in a project."""
312 return self.having_perm("unit.review", project)
314 def order(self):
315 return self.order_by("username")
317 def search(
318 self,
319 query: str,
320 parser: Literal["plain", "user", "superuser"] = "user",
321 **context,
322 ):
323 """High level wrapper for searching."""
324 if parser == "plain":
325 result = self.filter(
326 Q(username__icontains=query) | Q(full_name__icontains=query)
327 )
328 else:
329 filters, annotations = parse_query(query, parser=parser, **context)
330 result = self.annotate(**annotations).filter(filters)
331 return result.distinct()
333 def get_author_by_email(
334 self,
335 author_name: str | None,
336 author_email: str | None,
337 fallback: User | None,
338 request: AuthenticatedHttpRequest,
339 ) -> User | None:
340 from weblate.accounts.models import AuditLog
342 if author_email and (fallback is None or not fallback.has_email(author_email)):
343 author, created = User.objects.get_or_create(
344 email=author_email,
345 defaults={
346 "username": author_email,
347 "full_name": author_name or author_email,
348 },
349 )
350 if created:
351 AuditLog.objects.create(author, request, "autocreated")
352 if fallback is None and author.is_anonymous:
353 return author
354 if author.is_active and not author.is_bot and not author.is_anonymous:
355 return author
356 return fallback
358 def get_or_create(
359 self,
360 defaults: Mapping[str, Any] | None = None,
361 **kwargs: Any, # noqa: ANN401
362 ) -> tuple[User, bool]:
363 filtered: dict[str, Any] | None
364 extra: dict[str, Any]
365 if defaults is None: 365 ↛ 366line 365 didn't jump to line 366 because the condition on line 365 was never true
366 filtered = None
367 extra = {}
368 else:
369 filtered = {
370 name: value
371 for name, value in defaults.items()
372 if name not in User.DUMMY_FIELDS
373 }
374 extra = {
375 name: value
376 for name, value in defaults.items()
377 if name in User.DUMMY_FIELDS
378 }
380 user, created = super().get_or_create(defaults=filtered, **kwargs)
381 if created:
382 user.extra_data = extra
383 user.save()
384 return user, created
387@functools_cache
388def get_anonymous() -> User:
389 """Return an anonymous user."""
390 return User.objects.select_related("profile").get(
391 username=settings.ANONYMOUS_USER_NAME
392 )
395def convert_groups(objs):
396 """Convert Django Group objects to Weblate ones."""
397 objs = list(objs)
398 for idx, obj in enumerate(objs):
399 if isinstance(obj, DjangoGroup): 399 ↛ 400line 399 didn't jump to line 400 because the condition on line 399 was never true
400 objs[idx] = Group.objects.get_or_create(name=obj.name)[0]
401 return objs
404def wrap_group(func):
405 """Replace Django Group instances by Weblate Group instances."""
407 def group_wrapper(self, *objs, **kwargs):
408 objs = convert_groups(objs)
409 return func(self, *objs, **kwargs)
411 return group_wrapper
414def wrap_group_list(func):
415 """Replace Django Group instances by Weblate Group instances."""
417 def group_list_wrapper(self, objs, **kwargs):
418 objs = convert_groups(objs)
419 return func(self, objs, **kwargs)
421 return group_list_wrapper
424class GroupManyToManyField(models.ManyToManyField):
425 """Customized field to accept Django Groups objects as well."""
427 def contribute_to_class(
428 self, cls: type[models.Model], name: str, private_only: bool = False, **kwargs
429 ) -> None:
430 super().contribute_to_class(cls, name, private_only=private_only, **kwargs)
432 # Get related descriptor
433 descriptor = getattr(cls, self.name)
435 # We care only on forward relation
436 if not descriptor.reverse: 436 ↛ exitline 436 didn't return from function 'contribute_to_class' because the condition on line 436 was always true
437 # We are running in a migration
438 if isinstance(descriptor.rel.model, str):
439 return
441 # Get related manager class
442 related_manager_cls = descriptor.related_manager_cls
444 # Monkey patch it to accept Django Group instances as well
445 related_manager_cls.add = wrap_group(related_manager_cls.add)
446 related_manager_cls.remove = wrap_group(related_manager_cls.remove)
447 related_manager_cls.set = wrap_group_list(related_manager_cls.set)
450class User(AbstractBaseUser):
451 username = UsernameField(
452 gettext_lazy("Username"),
453 max_length=USERNAME_LENGTH,
454 unique=True,
455 help_text=gettext_lazy(
456 "Username may only contain letters, "
457 "numbers or the following characters: @ . + - _"
458 ),
459 validators=[validate_username],
460 error_messages={
461 "unique": gettext_lazy("A user with that username already exists.")
462 },
463 )
464 full_name = models.CharField(
465 gettext_lazy("Full name"),
466 max_length=FULLNAME_LENGTH,
467 blank=False,
468 validators=[validate_fullname],
469 )
470 email = EmailField(
471 gettext_lazy("E-mail"),
472 blank=False,
473 null=True,
474 unique=True,
475 )
476 is_superuser = models.BooleanField(
477 gettext_lazy("Superuser status"),
478 default=False,
479 help_text=gettext_lazy("User has all possible permissions."),
480 )
481 is_active = models.BooleanField(
482 gettext_lazy("Active"),
483 default=True,
484 help_text=gettext_lazy("Mark user as inactive instead of removing."),
485 )
486 is_bot = models.BooleanField(
487 "Robot user",
488 default=False,
489 db_index=True,
490 )
491 date_expires = models.DateTimeField(
492 gettext_lazy("Expires"),
493 null=True,
494 blank=True,
495 default=None,
496 validators=[MinValueValidator(timezone.now)],
497 help_text=gettext_lazy("The account will be disabled after the expiry."),
498 )
499 date_joined = models.DateTimeField(
500 gettext_lazy("Date joined"), default=timezone.now
501 )
502 groups = GroupManyToManyField(
503 Group,
504 verbose_name=gettext_lazy("Teams"),
505 blank=True,
506 help_text=gettext_lazy(
507 "The user is granted all permissions included in membership of these teams."
508 ),
509 )
511 objects = UserManager.from_queryset(UserQuerySet)()
513 # social_auth integration
514 social_auth: DjangoStorage
516 # django_otp integration (via OTPMiddleware)
517 otp_device: Device
519 EMAIL_FIELD = "email"
520 USERNAME_FIELD = "username"
521 REQUIRED_FIELDS = ["email", "full_name"] # noqa: RUF012
522 DUMMY_FIELDS = ("first_name", "last_name", "is_staff")
524 class Meta:
525 verbose_name = "User"
526 verbose_name_plural = "Users"
527 constraints = [ # noqa: RUF012
528 UniqueConstraint(Upper("username"), name="weblate_auth_user_username_ci"),
529 UniqueConstraint(Upper("email"), name="weblate_auth_user_email_ci"),
530 ]
532 def __str__(self) -> str:
533 return self.full_name
535 def save(self, *args, **kwargs) -> None:
536 from weblate.accounts.models import AuditLog
538 original = None
539 if self.pk:
540 original = User.objects.get(pk=self.pk)
541 if self.is_anonymous:
542 self.is_active = False
543 # Generate full name from parts
544 # This is needed with LDAP authentication when the
545 # server does not contain full name
546 if "first_name" in self.extra_data and "last_name" in self.extra_data: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 self.full_name = "{first_name} {last_name}".format(**self.extra_data)
548 elif "first_name" in self.extra_data: 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 self.full_name = self.extra_data["first_name"]
550 elif "last_name" in self.extra_data: 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 self.full_name = self.extra_data["last_name"]
552 if not self.email:
553 self.email = None
554 if not self.is_active:
555 self.date_expires = None
556 super().save(*args, **kwargs)
557 self.clear_cache()
558 if ( 558 ↛ 565line 558 didn't jump to line 565 because the condition on line 558 was never true
559 original
560 and original.is_active != self.is_active
561 and self.full_name != "Deleted User"
562 and not self.is_anonymous
563 ):
564 activity: str
565 if original.date_expires and not self.is_active:
566 activity = "disabled-expiry"
567 elif self.is_active:
568 activity = "enabled"
569 else:
570 activity = "disabled"
571 AuditLog.objects.create(user=self, request=None, activity=activity)
573 def get_absolute_url(self) -> str:
574 return reverse("user_page", kwargs={"user": self.username})
576 def __init__(self, *args, **kwargs) -> None:
577 self.extra_data: dict[str, str] = {}
578 self.cla_cache: dict[tuple[int, int], bool] = {}
579 self._permissions: PermissionsDictType = {}
580 self.current_subscription: Subscription | None = None
581 for name in self.DUMMY_FIELDS:
582 if name in kwargs: 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true
583 self.extra_data[name] = kwargs.pop(name)
584 super().__init__(*args, **kwargs)
586 def clear_cache(self) -> None:
587 self.cla_cache = {}
588 self._permissions = {}
589 perm_caches = (
590 "project_permissions",
591 "component_permissions",
592 "allowed_projects",
593 "needs_component_restrictions_filter",
594 "needs_project_filter",
595 "watched_projects",
596 "owned_projects",
597 "managed_projects",
598 "cached_groups",
599 )
600 for name in perm_caches:
601 if name in self.__dict__:
602 del self.__dict__[name]
604 def has_usable_password(self):
605 # For some reason Django says that empty string is a valid password
606 return self.password and super().has_usable_password()
608 @cached_property
609 def is_anonymous(self):
610 return self.username == settings.ANONYMOUS_USER_NAME
612 def is_verified(self) -> bool:
613 # django_otp overrides this method in OTPMiddleware
614 return False
616 @cached_property
617 def is_authenticated(self) -> bool: # type: ignore[override]
618 return not self.is_anonymous
620 def get_full_name(self):
621 return self.full_name
623 def get_short_name(self):
624 return self.full_name
626 def __setattr__(self, name, value) -> None:
627 """Mimic first/last name for third-party auth and ignore is_staff flag."""
628 if name in self.DUMMY_FIELDS: 628 ↛ 629line 628 didn't jump to line 629 because the condition on line 628 was never true
629 self.extra_data[name] = value
630 else:
631 super().__setattr__(name, value)
633 def has_module_perms(self, module):
634 """Compatibility API for admin interface."""
635 return self.is_superuser
637 @property
638 def is_staff(self):
639 """Compatibility API for admin interface."""
640 return self.is_superuser
642 @property
643 def first_name(self) -> str:
644 """Compatibility API for third-party modules."""
645 return ""
647 @property
648 def last_name(self):
649 """Compatibility API for third-party modules."""
650 return self.full_name
652 def has_perms(self, perm_list, obj=None) -> bool:
653 return all(self.has_perm(perm, obj) for perm in perm_list)
655 def has_perm(self, perm: str, obj=None) -> PermissionResult | bool:
656 """Permission check."""
657 # Weblate global scope permissions
658 if perm in GLOBAL_PERM_NAMES:
659 return check_global_permission(self, perm)
661 # Compatibility API for admin interface
662 if is_django_permission(perm): 662 ↛ 663line 662 didn't jump to line 663 because the condition on line 662 was never true
663 if not self.is_superuser:
664 return False
666 # Check permissions restrictions
667 allowed = settings.AUTH_RESTRICT_ADMINS.get(self.username)
668 return allowed is None or perm in allowed
670 # Validate perms
671 if perm not in SPECIALS and perm not in PERMISSION_NAMES: 671 ↛ 672line 671 didn't jump to line 672 because the condition on line 671 was never true
672 msg = f"Invalid permission: {perm}"
673 raise ValueError(msg)
675 # Special permission functions
676 if perm in SPECIALS:
677 return SPECIALS[perm](self, perm, obj)
679 # Generic permission
680 return check_permission(self, perm, obj)
682 def can_access_project(self, project):
683 """Check access to given project."""
684 if self.is_superuser:
685 return True
686 return self.get_project_permissions(project) != []
688 def get_project_permissions(self, project: Project) -> SimplePermissionList:
689 # Build a fresh list as we need to merge them
690 result: SimplePermissionList = []
691 # This relies on project_permission being defaultdict(list)
692 result.extend(self.project_permissions[project.pk]) # type: ignore[arg-type]
693 # Apply blocking
694 if result == [(None, None)]:
695 return []
696 if project.access_control == Project.ACCESS_PUBLIC:
697 result.extend(
698 self.project_permissions[-SELECTION_ALL_PUBLIC] # type: ignore[arg-type]
699 )
700 elif project.access_control == Project.ACCESS_PROTECTED:
701 result.extend(
702 self.project_permissions[-SELECTION_ALL_PROTECTED] # type: ignore[arg-type]
703 )
704 result.extend(
705 self.project_permissions[-SELECTION_ALL] # type: ignore[arg-type]
706 )
707 return result
709 def check_access(self, project) -> None:
710 """Raise an error if user is not allowed to access this project."""
711 if not self.can_access_project(project):
712 msg = "Access denied"
713 raise Http404(msg)
715 def can_access_component(self, component):
716 """Check access to given component."""
717 if self.is_superuser: 717 ↛ 719line 717 didn't jump to line 719 because the condition on line 717 was always true
718 return True
719 if not self.can_access_project(component.project):
720 return False
721 return not component.restricted or component.pk in self.component_permissions
723 def check_access_component(self, component) -> None:
724 """Raise an error if user is not allowed to access this component."""
725 if not self.can_access_component(component): 725 ↛ 726line 725 didn't jump to line 726 because the condition on line 725 was never true
726 msg = "Access denied"
727 raise Http404(msg)
729 @cached_property
730 def allowed_projects(self):
731 """List of allowed projects."""
732 if self.is_superuser: 732 ↛ 735line 732 didn't jump to line 735 because the condition on line 732 was always true
733 return Project.objects.order()
734 # All public and protected projects are accessible
735 acls = {Project.ACCESS_PUBLIC, Project.ACCESS_PROTECTED}
736 if -SELECTION_ALL in self.project_permissions:
737 acls.add(Project.ACCESS_PRIVATE)
738 acls.add(Project.ACCESS_CUSTOM)
739 condition = Q(access_control__in=acls)
741 # Add project-specific allowance
742 restricted = {-SELECTION_ALL_PUBLIC, -SELECTION_ALL_PROTECTED, -SELECTION_ALL}
743 project_ids = {key for key in self.project_permissions if key not in restricted}
744 if project_ids:
745 condition |= Q(pk__in=project_ids)
747 return Project.objects.filter(condition).order()
749 @cached_property
750 def needs_component_restrictions_filter(self):
751 if self.is_superuser: 751 ↛ 753line 751 didn't jump to line 753 because the condition on line 751 was always true
752 return False
753 return self.allowed_projects.filter(component__restricted=True).exists()
755 @cached_property
756 def needs_project_filter(self):
757 if self.is_superuser: 757 ↛ 759line 757 didn't jump to line 759 because the condition on line 757 was always true
758 return False
759 return self.allowed_projects.count() != Project.objects.all().count()
761 @cached_property
762 def watched_projects(self):
763 """
764 List of watched projects.
766 Ensure ACL filtering applies (the user could have been removed
767 from the project meanwhile)
768 """
769 return (self.profile.watched.all() & self.allowed_projects).order()
771 @cached_property
772 def owned_projects(self):
773 return self.projects_with_perm("project.edit", explicit=True)
775 @cached_property
776 def managed_projects(self):
777 return self.projects_with_perm("project.edit")
779 @cached_property
780 def administered_group_ids(self):
781 return set(self.administered_group_set.values_list("id", flat=True))
783 @cached_property
784 def cached_groups(self) -> Iterable[Group]:
785 return self.groups.prefetch_related(
786 "roles__permissions",
787 Prefetch(
788 "componentlists__components",
789 queryset=Component.objects.only("id", "project_id"),
790 ),
791 # The name and slug are used when rendering the groups
792 Prefetch(
793 "components",
794 queryset=Component.objects.all().only(
795 "id", "project_id", "name", "slug"
796 ),
797 ),
798 # The name and slug are used when rendering the groups
799 Prefetch(
800 "projects",
801 queryset=Project.objects.only("id", "name", "slug"),
802 ),
803 # The name and code are used when rendering the groups
804 Prefetch("languages", queryset=Language.objects.only("id", "name", "code")),
805 )
807 def group_enforces_2fa(self) -> bool:
808 return any(group.enforced_2fa for group in self.cached_groups)
810 def _fetch_permissions(self) -> None:
811 """Fetch all user permissions into a dictionary."""
812 projects: PermissionCacheType = defaultdict(list)
813 components: SimplePermissionCacheType = defaultdict(list)
814 with sentry_sdk.start_span(op="auth.permissions", name=self.username):
815 for group in self.cached_groups:
816 # Skip permissions for not verified users
817 if group.enforced_2fa and not self.profile.has_2fa: 817 ↛ 818line 817 didn't jump to line 818 because the condition on line 817 was never true
818 continue
819 if group.language_selection == SELECTION_ALL: 819 ↛ 822line 819 didn't jump to line 822 because the condition on line 819 was always true
820 languages = None
821 else:
822 languages = {language.id for language in group.languages.all()}
823 permissions = {
824 permission.codename
825 for permission in chain.from_iterable(
826 role.permissions.all() for role in group.roles.all()
827 )
828 }
830 # Component list specific permissions
831 componentlist_values = {
832 (component.id, component.project_id)
833 for component in chain.from_iterable(
834 clist.components.all() for clist in group.componentlists.all()
835 )
836 }
837 if group.componentlists.exists(): 837 ↛ 838line 837 didn't jump to line 838 because the condition on line 837 was never true
838 for component, project in componentlist_values:
839 components[component].append((permissions, languages))
840 # Grant access to the project
841 projects[project].append((set(), languages))
842 continue
844 # Component specific permissions
845 component_values = {
846 (component.id, component.project_id)
847 for component in group.components.all()
848 }
849 if component_values: 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true
850 for component, project in component_values:
851 components[component].append((permissions, languages))
852 # Grant access to the project
853 projects[project].append((set(), languages))
854 continue
856 # Handle project selection
857 if group.project_selection in { 857 ↛ 865line 857 didn't jump to line 865 because the condition on line 857 was always true
858 SELECTION_ALL_PUBLIC,
859 SELECTION_ALL_PROTECTED,
860 SELECTION_ALL,
861 }:
862 projects[-group.project_selection].append((permissions, languages))
863 else:
864 # Project specific permissions
865 for project_obj in group.projects.all():
866 projects[project_obj.id].append((permissions, languages))
867 # Apply blocking
868 now = timezone.now()
869 for block in self.userblock_set.all(): 869 ↛ 870line 869 didn't jump to line 870 because the loop on line 869 never started
870 if block.expiry is not None and block.expiry <= now:
871 # Delete expired blocks
872 block.delete()
873 else:
874 # Remove all permissions for blocked user
875 projects[block.project_id] = [(None, None)]
877 self._permissions = {"projects": projects, "components": components}
879 @cached_property
880 def project_permissions(self) -> PermissionCacheType:
881 """List all project permissions."""
882 if not self._permissions: 882 ↛ 884line 882 didn't jump to line 884 because the condition on line 882 was always true
883 self._fetch_permissions()
884 return self._permissions["projects"]
886 @cached_property
887 def component_permissions(self) -> SimplePermissionCacheType:
888 """List all project permissions."""
889 if not self._permissions:
890 self._fetch_permissions()
891 return self._permissions["components"]
893 @cached_property
894 def global_permissions(self) -> set[str]:
895 return set(
896 Permission.objects.filter(
897 role__group__user=self, codename__in=GLOBAL_PERM_NAMES
898 ).values_list("codename", flat=True)
899 )
901 def projects_with_perm(self, perm: str, explicit: bool = False):
902 if not explicit and self.is_superuser:
903 return Project.objects.all().order()
904 # Explicit permissions
905 condition = Q(group__user=self) & Q(group__roles__permissions__codename=perm)
907 # Site-wide permissions
908 if not explicit: 908 ↛ 922line 908 didn't jump to line 922 because the condition on line 908 was always true
909 for access, selection in (
910 (Project.ACCESS_PUBLIC, -SELECTION_ALL_PUBLIC),
911 (Project.ACCESS_PROTECTED, -SELECTION_ALL_PROTECTED),
912 (None, -SELECTION_ALL),
913 ):
914 if any( 914 ↛ 918line 914 didn't jump to line 918 because the condition on line 914 was never true
915 perm in cast("set[str]", permissions)
916 for permissions, _langs in self.project_permissions[selection]
917 ):
918 if access is None:
919 condition = Q()
920 break
921 condition |= Q(access_control=access)
922 return Project.objects.filter(condition).distinct().order()
924 def get_visible_name(self) -> str:
925 """Get full name from database or username."""
926 if not self.full_name or CRUD_RE.match(self.full_name): 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 return self.username
928 return self.full_name
930 def get_author_name(self, address: str | None = None) -> str:
931 """Return formatted author name with e-mail."""
932 return format_address(
933 self.get_visible_name(), address or self.profile.get_commit_email()
934 )
936 def add_team(
937 self,
938 request: AuthenticatedHttpRequest | None,
939 team: Group,
940 *,
941 user: User | None = None,
942 ) -> None:
943 from weblate.accounts.models import AuditLog
945 self.groups.add(team)
947 username: str | None
948 if user is not None: 948 ↛ 949line 948 didn't jump to line 949 because the condition on line 948 was never true
949 username = user.username
950 elif request is not None: 950 ↛ 951line 950 didn't jump to line 951 because the condition on line 950 was never true
951 username = request.user.username
952 else:
953 username = None
955 AuditLog.objects.create(
956 user=self,
957 request=request if request is not None and request.user == self else None,
958 activity="team-add",
959 username=username,
960 team=team.name,
961 )
963 def remove_team(
964 self, request: AuthenticatedHttpRequest | None, team: Group
965 ) -> None:
966 from weblate.accounts.models import AuditLog
968 self.groups.remove(team)
969 AuditLog.objects.create(
970 user=self,
971 request=request if request is not None and request.user == self else None,
972 activity="team-remove",
973 username=request.user.username
974 if request is not None and request.user
975 else None,
976 team=team.name,
977 )
979 def has_email(self, email: str) -> bool:
980 return (
981 email == self.email
982 or User.objects.filter(
983 pk=self.pk, social_auth__verifiedemail__email=email
984 ).exists()
985 )
988class AutoGroup(models.Model):
989 match = RegexField(
990 verbose_name=gettext_lazy("Regular expression for e-mail address"),
991 max_length=200,
992 default="^$",
993 help_text=gettext_lazy(
994 "Users with e-mail addresses found to match will be added to this team."
995 ),
996 )
997 group = models.ForeignKey(
998 Group,
999 verbose_name=gettext_lazy("Team to assign"),
1000 on_delete=models.deletion.CASCADE,
1001 )
1003 class Meta:
1004 verbose_name = "Automatic team assignment"
1005 verbose_name_plural = "Automatic team assignments"
1007 def __str__(self) -> str:
1008 return f"Automatic rule for {self.group}"
1011class UserBlock(models.Model):
1012 user = models.ForeignKey(
1013 User,
1014 verbose_name=gettext_lazy("User to block"),
1015 on_delete=models.deletion.CASCADE,
1016 db_index=False,
1017 )
1018 project = models.ForeignKey(
1019 Project, verbose_name=gettext_lazy("Project"), on_delete=models.deletion.CASCADE
1020 )
1021 expiry = models.DateTimeField(gettext_lazy("Block expiry"), null=True)
1023 class Meta:
1024 verbose_name = "Blocked user"
1025 verbose_name_plural = "Blocked users"
1026 unique_together = [ # noqa: RUF012
1027 ("user", "project"),
1028 ]
1030 def __str__(self) -> str:
1031 return f"{self.user} blocked for {self.project}"
1034def create_groups(update) -> None:
1035 """Create standard groups and gives them permissions."""
1036 # Create permissions and roles
1037 migrate_permissions(Permission)
1038 new_roles = migrate_roles(Role, Permission)
1039 builtin_groups = migrate_groups(Group, Role, update)
1041 # Create anonymous user
1042 create_anonymous(User, Group, update)
1044 # Automatic assignment to the users group
1045 group = builtin_groups["Users"]
1046 if not AutoGroup.objects.filter(group=group).exists(): 1046 ↛ 1048line 1046 didn't jump to line 1048 because the condition on line 1046 was always true
1047 AutoGroup.objects.create(group=group, match="^.*$")
1048 group = builtin_groups["Viewers"]
1049 if not AutoGroup.objects.filter(group=group).exists(): 1049 ↛ 1053line 1049 didn't jump to line 1053 because the condition on line 1049 was always true
1050 AutoGroup.objects.create(group=group, match="^.*$")
1052 # Create new per project groups
1053 if new_roles: 1053 ↛ exitline 1053 didn't return from function 'create_groups' because the condition on line 1053 was always true
1054 for project in Project.objects.iterator(): 1054 ↛ 1055line 1054 didn't jump to line 1055 because the loop on line 1054 never started
1055 setup_project_groups(Project, project, new_roles=new_roles)
1058def sync_create_groups(sender, **kwargs) -> None:
1059 """Create default groups."""
1060 create_groups(False)
1063def auto_assign_group(user: User) -> None:
1064 """Automatic group assignment based on user e-mail address."""
1065 if user.username == settings.ANONYMOUS_USER_NAME:
1066 return
1067 # Add user to automatic groups
1068 for auto in AutoGroup.objects.prefetch_related("group"):
1069 if re.match(auto.match, user.email or ""): 1069 ↛ 1068line 1069 didn't jump to line 1068 because the condition on line 1069 was always true
1070 user.add_team(None, auto.group)
1073@receiver(m2m_changed, sender=ComponentList.components.through)
1074@disable_for_loaddata
1075def change_componentlist(sender, instance, action, **kwargs) -> None:
1076 if not action.startswith("post_"):
1077 return
1078 groups = Group.objects.filter(
1079 componentlists=instance, project_selection=SELECTION_COMPONENT_LIST
1080 )
1081 for group in groups:
1082 group.projects.set(
1083 Project.objects.filter(component__componentlist=instance), clear=True
1084 )
1087@receiver(m2m_changed, sender=User.groups.through)
1088def remove_group_admin(sender, instance, action, pk_set, reverse, **kwargs) -> None:
1089 if action != "post_remove": 1089 ↛ 1091line 1089 didn't jump to line 1091 because the condition on line 1089 was always true
1090 return
1091 for pk in pk_set:
1092 if reverse:
1093 group = instance
1094 user = User.objects.get(pk=pk)
1095 else:
1096 group = Group.objects.get(pk=pk)
1097 user = instance
1098 group.admins.remove(user)
1101@receiver(post_save, sender=User)
1102@disable_for_loaddata
1103def auto_group_upon_save(sender, instance, created=False, **kwargs) -> None:
1104 """Apply automatic group assignment rules."""
1105 if created:
1106 auto_assign_group(instance)
1109@receiver(post_save, sender=Project)
1110@disable_for_loaddata
1111def setup_project_groups(
1112 sender,
1113 instance,
1114 created: bool = False,
1115 new_roles: set[str] | None = None,
1116 **kwargs,
1117) -> None:
1118 """Set up group objects upon saving project."""
1119 old_access_control = instance.old_access_control
1120 instance.old_access_control = instance.access_control
1122 changed_review = (
1123 instance.old_translation_review != instance.translation_review
1124 or instance.old_source_review != instance.source_review
1125 )
1126 # Handle no groups as newly created project
1127 if not created and not instance.defined_groups.exists(): 1127 ↛ 1128line 1127 didn't jump to line 1128 because the condition on line 1127 was never true
1128 created = True
1130 # No changes needed
1131 if (
1132 old_access_control == instance.access_control
1133 and not changed_review
1134 and not created
1135 and not new_roles
1136 ):
1137 return
1139 # Do not perform anything with custom ACL
1140 if instance.access_control == Project.ACCESS_CUSTOM: 1140 ↛ 1141line 1140 didn't jump to line 1141 because the condition on line 1140 was never true
1141 return
1143 # Choose groups to configure
1144 if instance.access_control == Project.ACCESS_PUBLIC: 1144 ↛ 1147line 1144 didn't jump to line 1147 because the condition on line 1144 was always true
1145 groups = {"Administration", "Review"}
1146 else:
1147 groups = set(ACL_GROUPS.keys())
1149 # Remove review group if review is not enabled
1150 if not instance.source_review and not instance.translation_review:
1151 groups.remove("Review")
1153 # Remove billing if billing is not installed
1154 if "weblate.billing" not in settings.INSTALLED_APPS: 1154 ↛ 1158line 1154 didn't jump to line 1158 because the condition on line 1154 was always true
1155 groups.discard("Billing")
1157 # Filter only newly introduced groups
1158 if new_roles: 1158 ↛ 1159line 1158 didn't jump to line 1159 because the condition on line 1158 was never true
1159 groups = {group for group in groups if ACL_GROUPS[group] in new_roles}
1161 # Access control changed
1162 elif ( 1162 ↛ 1173line 1162 didn't jump to line 1173 because the condition on line 1162 was never true
1163 not created
1164 and (
1165 instance.access_control == Project.ACCESS_PUBLIC
1166 or old_access_control in {Project.ACCESS_PROTECTED, Project.ACCESS_PRIVATE}
1167 )
1168 and not changed_review
1169 ):
1170 # Avoid changing groups on some access control changes:
1171 # - Public groups are always present, so skip change on changing to public
1172 # - Change between protected/private means no change in groups
1173 return
1175 # Create role specific groups
1176 for group_name in groups:
1177 group, created = instance.defined_groups.get_or_create(
1178 internal=True,
1179 name=group_name,
1180 project_selection=SELECTION_MANUAL,
1181 defining_project=instance,
1182 language_selection=SELECTION_ALL,
1183 )
1184 if not created:
1185 continue
1186 group.projects.add(instance)
1187 group.roles.add(Role.objects.get(name=ACL_GROUPS[group_name]))
1190class Invitation(models.Model):
1191 """
1192 User invitation store.
1194 Either user or e-mail attribute is set, this is to invite current and new users.
1195 """
1197 uuid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
1198 timestamp = models.DateTimeField(auto_now_add=True)
1199 author = models.ForeignKey(
1200 User, on_delete=models.deletion.CASCADE, related_name="created_invitation_set"
1201 )
1202 user = models.ForeignKey(
1203 User,
1204 on_delete=models.deletion.CASCADE,
1205 null=True,
1206 verbose_name=gettext_lazy("User to add"),
1207 help_text=gettext_lazy(
1208 "Please type in an existing Weblate account name or e-mail address."
1209 ),
1210 )
1211 username = UsernameField(
1212 gettext_lazy("Username"),
1213 max_length=USERNAME_LENGTH,
1214 blank=True,
1215 help_text=gettext_lazy(
1216 "Suggest username for the user. It can be changed later."
1217 ),
1218 validators=[validate_username],
1219 )
1220 full_name = models.CharField(
1221 gettext_lazy("Full name"),
1222 max_length=FULLNAME_LENGTH,
1223 blank=True,
1224 help_text=gettext_lazy(
1225 "Suggest full name for the user. It can be changed later."
1226 ),
1227 validators=[validate_fullname],
1228 )
1229 group = models.ForeignKey(
1230 Group,
1231 verbose_name=gettext_lazy("Team"),
1232 help_text=gettext_lazy(
1233 "The user is granted all permissions included in membership of these teams."
1234 ),
1235 on_delete=models.deletion.CASCADE,
1236 )
1237 email = EmailField(
1238 gettext_lazy("E-mail"),
1239 blank=True,
1240 )
1241 is_superuser = models.BooleanField(
1242 gettext_lazy("Superuser status"),
1243 default=False,
1244 help_text=gettext_lazy("User has all possible permissions."),
1245 )
1247 def __str__(self) -> str:
1248 return f"invitation {self.uuid} for {self.user or self.email} to {self.group}"
1250 def get_absolute_url(self) -> str:
1251 return reverse("invitation", kwargs={"pk": self.uuid})
1253 def send_email(self) -> None:
1254 from weblate.accounts.notifications import send_notification_email
1256 email: str
1257 if self.email:
1258 email = self.email
1259 elif self.user is not None:
1260 email = self.user.email
1261 else:
1262 msg = "Intiviation without an e-mail!"
1263 raise ValueError(msg)
1265 send_notification_email(
1266 None,
1267 [email],
1268 "invite",
1269 info=f"{self}",
1270 context={"invitation": self, "validity": settings.AUTH_TOKEN_VALID // 3600},
1271 )
1273 def accept(self, request: AuthenticatedHttpRequest, user: User) -> None:
1274 from weblate.accounts.models import AuditLog
1276 if self.user and self.user != user:
1277 msg = "User mismatch on accept!"
1278 raise ValueError(msg)
1280 if self.is_superuser:
1281 user.is_superuser = True
1282 user.save(update_fields=["is_superuser"])
1284 AuditLog.objects.create(
1285 user=user,
1286 request=request,
1287 activity="accepted",
1288 username=self.author.username,
1289 )
1291 user.add_team(request, self.group, user=self.author)
1293 self.delete()
1296class WeblateAuthConf(AppConf):
1297 """Authentication settings."""
1299 AUTH_RESTRICT_ADMINS: ClassVar[dict] = {}
1301 # Anonymous user name
1302 ANONYMOUS_USER_NAME = "anonymous"
1304 SESSION_COOKIE_AGE_AUTHENTICATED = 1209600
1305 SESSION_COOKIE_AGE_2FA = 180
1307 class Meta:
1308 prefix = ""
1311class AuthenticatedHttpRequest(HttpRequest):
1312 user: User
1313 # Added by weblate.accounts.AuthenticationMiddleware
1314 accepted_language: Language
1316 # type hint for social_auth
1317 social_strategy: WeblateStrategy
1319 # type hint for auth
1320 backend: BaseAuth | None
1322 # type hint for accounts middleware
1323 weblate_cached_user: User
1325 # type hint for wladmin
1326 weblate_support_status: SupportStatusDict
1328 # type hint for configuration module
1329 weblate_custom_css: str
1331 # Overrides django.http.request URL generating
1332 _current_scheme_host: str