forked from CGM_Public/pretix_original
* Improve subject lines for admin-facing emails
A few of the current subjects are ambiguous about the expected
action, and some omit context that would help in an inbox preview
(which event, which address). The rewrites bring them closer to
common conventions in modern transactional email (verb-led,
recipient-addressed, with recipient-meaningful variables). Two
themes:
- Action-required emails lead with the action verb. "Reset your
password", "Confirm event cancellation and bulk refund", and
"Confirm <address> as a sender address" tell the recipient up
front what's expected, where "Password recovery", "Bulk-refund
confirmation" and "Sender address verification" did not.
- Surface the relevant variable when the email is about something
specific. "Data shredding completed for <event>" is more useful
than the generic version when an admin manages several events.
"You've been invited to join <organizer>" names the inviting
organizer. "Confirm <address> as a sender address" names the
address.
The remaining rewrites are lighter rewordings. "New sign-in to
your account" replaces "Login from new source detected" because
"source" is jargon a non-technical recipient wouldn't recognise.
"Changes to your account" replaces "Account information changed"
because the possessive frames the email as being about the
recipient's own account.
Also fixes a hardcoded "pretix" in the confirmation-code subject.
* Standardise admin email sign-offs as "Thanks, The <instance> Team"
The current sign-offs ("Best regards, Your <instance> team") have
a formal tone. A review of the last ~20 transactional emails in
my inbox showed most senders use something friendlier:
- Thanks: Deliveroo, Starling Bank, GitHub, Cloudflare
- Thank you: AWS
- Sincerely: Google Workspace
A small minority (e.g., Sentry) had no sign-off at all. "Thanks"
was the most common, and among that group "The <instance> Team"
was the consistent phrasing rather than "Your <instance> team".
Two templates (cancel_confirm, export_failed) didn't have a
sign-off; they now get one for consistency. Notification emails
are deliberately excluded: they're system alerts rather than
direct correspondence.
* Add anti-phishing notice to admin emails containing confirmation codes
Three admin emails send the recipient a confirmation code to
enter back into a form: confirmation_code, email_setup, and
cancel_confirm. Only confirmation_code had an anti-phishing
warning, and its wording was awkward ("Please do never give this
code to another person. Our support team will never ask for this
code.").
This commit standardises the warning across all three:
> Don't share this code with anyone. The <instance> team will
> never ask you for it.
* Add structured details to login-notice email
The single-sentence body ("The login was performed using <agent>
on <os> from <country>.") is replaced with a labelled bullet list:
Time, Browser, Operating system, Device, Country.
Time and Device are new fields. Device is omitted when ua-parser
can't identify the device, Country when GeoIP isn't available,
so the user only sees fields with real values.
* Restructure notification.txt for clearer layout
- Attributes: bullet list instead of paragraph-per-attribute.
- Actions: label gets a colon, URL on its own paragraph (was
4-space-indented code block).
- Footer: separated by --- and bulleted (manage / disable
links). "Click here X" phrasing dropped (incidentally moots
a missing-"to" typo).
- Minor whitespace fix: detail-block endif now matches the
placement of the rest of the template.
notification.html's footer text is also updated, only to match
the new .txt wording (link labels and intro line). No
structural changes to the HTML template.
* Improve confirmation-code email reason strings
- Drop the redundant "to confirm" opener.
- Replace hardcoded "your pretix account" in email_verify
with "{instance}".
* Polish admin email body copy
A small wording and formatting pass on the admin email bodies,
in three loosely-grouped themes:
1. Sentence case for body text (previously lowercase after
"Hello,"), matching standard English convention.
2. Light restructuring where helpful: bullet lists for sets
of labelled facts; 4-space-indented code blocks for codes
the recipient is meant to type back.
3. Phrasing polish. Some sentences tightened or shortened.
Largely matters of taste, but generally read smoother.
---------
Co-authored-by: Raphael Michel <michel@pretix.eu>
803 lines
32 KiB
Python
803 lines
32 KiB
Python
#
|
|
# This file is part of pretix (Community Edition).
|
|
#
|
|
# Copyright (C) 2014-2020 Raphael Michel and contributors
|
|
# Copyright (C) 2020-today pretix GmbH and contributors
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General
|
|
# Public License as published by the Free Software Foundation in version 3 of the License.
|
|
#
|
|
# ADDITIONAL TERMS APPLY: Pursuant to Section 7 of the GNU Affero General Public License, additional terms are
|
|
# applicable granting you additional permissions and placing additional restrictions on your usage of this software.
|
|
# Please refer to the pretix LICENSE file to obtain the full terms applicable to this work. If you did not receive
|
|
# this file, see <https://pretix.eu/about/en/license>.
|
|
#
|
|
# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
|
|
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
|
|
# details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License along with this program. If not, see
|
|
# <https://www.gnu.org/licenses/>.
|
|
#
|
|
|
|
# This file is based on an earlier version of pretix which was released under the Apache License 2.0. The full text of
|
|
# the Apache License 2.0 can be obtained at <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# This file may have since been changed and any changes are released under the terms of AGPLv3 as described above. A
|
|
# full history of changes and contributors is available at <https://github.com/pretix/pretix>.
|
|
#
|
|
# This file contains Apache-licensed contributions copyrighted by: Jakob Schnell, Tobias Kunze
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software distributed under the Apache License 2.0 is
|
|
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations under the License.
|
|
|
|
import binascii
|
|
import json
|
|
import operator
|
|
import secrets
|
|
from datetime import timedelta
|
|
from functools import reduce
|
|
from typing import Protocol
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import (
|
|
AbstractBaseUser, BaseUserManager, PermissionsMixin,
|
|
)
|
|
from django.contrib.auth.tokens import default_token_generator
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.exceptions import BadRequest, PermissionDenied
|
|
from django.db import IntegrityError, models, transaction
|
|
from django.db.models import Q
|
|
from django.utils.crypto import get_random_string, salted_hmac
|
|
from django.utils.functional import cached_property
|
|
from django.utils.timezone import now
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django_otp.models import Device
|
|
from django_scopes import scopes_disabled
|
|
|
|
from pretix.base.i18n import language
|
|
from pretix.helpers.urls import mainreverse_absolute
|
|
|
|
from ...helpers.countries import FastCountryField
|
|
from ...helpers.u2f import pub_key_from_der, websafe_decode
|
|
from .base import LoggingMixin
|
|
|
|
|
|
class EmailAddressTakenError(IntegrityError):
|
|
pass
|
|
|
|
|
|
class PermissionHolder(Protocol):
|
|
def has_event_permission(self, organizer, event, perm_name=None, request=None, session_key=None) -> bool:
|
|
...
|
|
|
|
def has_organizer_permission(self, organizer, perm_name=None, request=None):
|
|
...
|
|
|
|
|
|
class UserManager(BaseUserManager):
|
|
"""
|
|
This is the user manager for our custom user model. See the User
|
|
model documentation to see what's so special about our user model.
|
|
"""
|
|
|
|
def create_user(self, email: str, password: str = None, **kwargs):
|
|
user = self.model(email=email, **kwargs)
|
|
user.set_password(password)
|
|
user.save()
|
|
return user
|
|
|
|
def create_superuser(self, email: str, password: str = None): # NOQA
|
|
# Not used in the software but required by Django
|
|
if password is None:
|
|
raise Exception("You must provide a password")
|
|
user = self.model(email=email)
|
|
user.is_staff = True
|
|
user.set_password(password)
|
|
user.save()
|
|
return user
|
|
|
|
def get_or_create_for_backend(self, backend, identifier, email, set_always, set_on_creation):
|
|
"""
|
|
This method should be used by third-party authentication backends to log in a user.
|
|
It either returns an already existing user or creates a new user.
|
|
|
|
In pretix 4.7 and earlier, email addresses were the only property to identify a user with.
|
|
Starting with pretix 4.8, backends SHOULD instead use a unique, immutable identifier
|
|
based on their backend data store to allow for changing email addresses.
|
|
|
|
This method transparently handles the conversion of old user accounts and adds the
|
|
backend identifier to their database record.
|
|
|
|
This method will never return users managed by a different authentication backend.
|
|
If you try to create an account with an email address already blocked by a different
|
|
authentication backend, :py:class:`EmailAddressTakenError` will be raised. In this case,
|
|
you should display a message to the user.
|
|
|
|
:param backend: The `identifier` attribute of the authentication backend
|
|
:param identifier: The unique, immutable identifier of this user, max. 190 characters
|
|
:param email: The user's email address
|
|
:param set_always: A dictionary of fields to update on the user model on every login
|
|
:param set_on_creation: A dictionary of fields to set on the user model if it's newly created
|
|
:return: A `User` instance.
|
|
"""
|
|
if identifier is None:
|
|
raise ValueError('You need to supply a custom, unique identifier for this user.')
|
|
if email is None:
|
|
raise ValueError('You need to supply an email address for this user.')
|
|
if 'auth_backend_identifier' in set_always or 'auth_backend_identifier' in set_on_creation or \
|
|
'auth_backend' in set_always or 'auth_backend' in set_on_creation:
|
|
raise ValueError('You may not update auth_backend/auth_backend_identifier.')
|
|
if len(identifier) > 190:
|
|
raise ValueError('The user identifier must not be more than 190 characters.')
|
|
|
|
# Always update the email address
|
|
set_always.update({'email': email})
|
|
|
|
# First, check if we find the user based on it's backend-specific authenticator
|
|
try:
|
|
u = self.get(
|
|
auth_backend=backend,
|
|
auth_backend_identifier=identifier,
|
|
)
|
|
dirty = False
|
|
for k, v in set_always.items():
|
|
if getattr(u, k) != v:
|
|
setattr(u, k, v)
|
|
dirty = True
|
|
if dirty:
|
|
try:
|
|
with transaction.atomic():
|
|
u.save(update_fields=set_always.keys())
|
|
except IntegrityError:
|
|
# This might only raise IntegrityError if the email address is used
|
|
# by someone else
|
|
raise EmailAddressTakenError()
|
|
return u
|
|
except self.model.DoesNotExist:
|
|
pass
|
|
|
|
# Second, check if we find the user based on their email address and this backend
|
|
try:
|
|
u = self.get(
|
|
auth_backend=backend,
|
|
auth_backend_identifier__isnull=True,
|
|
email=email,
|
|
)
|
|
u.auth_backend_identifier = identifier
|
|
for k, v in set_always.items():
|
|
setattr(u, k, v)
|
|
try:
|
|
with transaction.atomic():
|
|
u.save(update_fields=['auth_backend_identifier'] + list(set_always.keys()))
|
|
return u
|
|
except IntegrityError:
|
|
# This might only raise IntegrityError if this code is being executed twice
|
|
# and runs into a race condition, this mechanism is taken from Django's
|
|
# get_or_create
|
|
try:
|
|
return self.get(
|
|
auth_backend=backend,
|
|
auth_backend_identifier=identifier,
|
|
)
|
|
except self.model.DoesNotExist:
|
|
pass
|
|
raise
|
|
except self.model.DoesNotExist:
|
|
pass
|
|
|
|
# Third, create a new user
|
|
u = User(
|
|
auth_backend=backend,
|
|
auth_backend_identifier=identifier,
|
|
**set_on_creation,
|
|
**set_always,
|
|
)
|
|
try:
|
|
u.save(force_insert=True)
|
|
return u
|
|
except IntegrityError:
|
|
# This might either be a race condition or the email address is taken
|
|
# by a different backend
|
|
try:
|
|
return self.get(
|
|
auth_backend=backend,
|
|
auth_backend_identifier=identifier,
|
|
)
|
|
except self.model.DoesNotExist:
|
|
raise EmailAddressTakenError()
|
|
|
|
|
|
def generate_notifications_token():
|
|
return get_random_string(length=32)
|
|
|
|
|
|
def generate_session_token():
|
|
return get_random_string(length=32)
|
|
|
|
|
|
class SuperuserPermissionSet:
|
|
def __contains__(self, item):
|
|
return True
|
|
|
|
|
|
class EventPermissionSet(set):
|
|
def __contains__(self, item):
|
|
from pretix.base.permissions import assert_valid_event_permission
|
|
|
|
if super().__contains__(item):
|
|
return True
|
|
|
|
assert_valid_event_permission(item, allow_tuple=False)
|
|
return False
|
|
|
|
|
|
class OrganizerPermissionSet(set):
|
|
def __contains__(self, item):
|
|
from pretix.base.permissions import assert_valid_organizer_permission
|
|
|
|
if super().__contains__(item):
|
|
return True
|
|
|
|
assert_valid_organizer_permission(item, allow_tuple=False)
|
|
return False
|
|
|
|
|
|
class User(AbstractBaseUser, PermissionsMixin, LoggingMixin):
|
|
"""
|
|
This is the user model used by pretix for authentication.
|
|
|
|
:param email: The user's email address, used for identification.
|
|
:type email: str
|
|
:param fullname: The user's full name. May be empty or null.
|
|
:type fullname: str
|
|
:param is_active: Whether this user account is activated.
|
|
:type is_active: bool
|
|
:param is_staff: ``True`` for system operators.
|
|
:type is_staff: bool
|
|
:param date_joined: The datetime of the user's registration.
|
|
:type date_joined: datetime
|
|
:param locale: The user's preferred locale code.
|
|
:type locale: str
|
|
:param needs_password_change: Whether this user's password needs to be changed.
|
|
:type needs_password_change: bool
|
|
:param timezone: The user's preferred timezone.
|
|
:type timezone: str
|
|
:param auth_backend: The identifier of the authentication backend plugin responsible for managing this user.
|
|
:type auth_backend: str
|
|
:param auth_backend_identifier: The native identifier of the user provided by a non-native authentication backend.
|
|
:type auth_backend_identifier: str
|
|
"""
|
|
|
|
USERNAME_FIELD = 'email'
|
|
REQUIRED_FIELDS = []
|
|
MAX_CONFIRMATION_CODE_ATTEMPTS = 10
|
|
|
|
email = models.EmailField(unique=True, db_index=True, null=True, blank=True,
|
|
verbose_name=_('Email'), max_length=190)
|
|
is_verified = models.BooleanField(default=False, verbose_name=_('Verified email address'))
|
|
fullname = models.CharField(max_length=255, blank=True, null=True,
|
|
verbose_name=_('Full name'))
|
|
is_active = models.BooleanField(default=True,
|
|
verbose_name=_('Is active'))
|
|
is_staff = models.BooleanField(default=False,
|
|
verbose_name=_('Is site admin'))
|
|
date_joined = models.DateTimeField(auto_now_add=True,
|
|
verbose_name=_('Date joined'))
|
|
needs_password_change = models.BooleanField(default=False,
|
|
verbose_name=_('Force user to select a new password'))
|
|
locale = models.CharField(max_length=50,
|
|
choices=settings.LANGUAGES,
|
|
default=settings.LANGUAGE_CODE,
|
|
verbose_name=_('Language'))
|
|
timezone = models.CharField(max_length=100,
|
|
default=settings.TIME_ZONE,
|
|
verbose_name=_('Timezone'))
|
|
require_2fa = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_('Two-factor authentication is required to log in')
|
|
)
|
|
notifications_send = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_('Receive notifications according to my settings below'),
|
|
help_text=_('If turned off, you will not get any notifications.')
|
|
)
|
|
notifications_token = models.CharField(max_length=255, default=generate_notifications_token)
|
|
auth_backend = models.CharField(max_length=255, default='native')
|
|
auth_backend_identifier = models.CharField(max_length=190, db_index=True, null=True, blank=True)
|
|
session_token = models.CharField(max_length=32, default=generate_session_token)
|
|
|
|
objects = UserManager()
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._teamcache = {}
|
|
|
|
class Meta:
|
|
verbose_name = _("User")
|
|
verbose_name_plural = _("Users")
|
|
ordering = ('email',)
|
|
unique_together = (('auth_backend', 'auth_backend_identifier'),)
|
|
|
|
def save(self, *args, **kwargs):
|
|
self.email = self.email.lower()
|
|
is_new = not self.pk
|
|
super().save(*args, **kwargs)
|
|
if is_new:
|
|
self.notification_settings.create(
|
|
action_type='pretix.event.order.refund.requested',
|
|
event=None,
|
|
method='mail',
|
|
enabled=True
|
|
)
|
|
|
|
def __str__(self):
|
|
return self.email
|
|
|
|
@property
|
|
def is_superuser(self):
|
|
return False
|
|
|
|
def get_short_name(self) -> str:
|
|
"""
|
|
Returns the first of the following user properties that is found to exist:
|
|
|
|
* Full name
|
|
* Email address
|
|
|
|
Only present for backwards compatibility
|
|
"""
|
|
if self.fullname:
|
|
return self.fullname
|
|
else:
|
|
return self.email
|
|
|
|
def get_full_name(self) -> str:
|
|
"""
|
|
Returns the first of the following user properties that is found to exist:
|
|
|
|
* Full name
|
|
* Email address
|
|
"""
|
|
if self.fullname:
|
|
return self.fullname
|
|
else:
|
|
return self.email
|
|
|
|
def send_security_notice(self, messages, email=None):
|
|
from pretix.base.services.mail import mail
|
|
|
|
with language(self.locale):
|
|
msg = '- ' + '\n- '.join(str(m) for m in messages)
|
|
|
|
mail(
|
|
email or self.email,
|
|
_('Changes to your account'),
|
|
'pretixcontrol/email/security_notice.txt',
|
|
{
|
|
'user': self,
|
|
'messages': msg,
|
|
'url': mainreverse_absolute('control:user.settings'),
|
|
'instance': settings.PRETIX_INSTANCE_NAME,
|
|
},
|
|
event=None,
|
|
user=self,
|
|
locale=self.locale
|
|
)
|
|
|
|
def send_confirmation_code(self, session, reason, email=None, state=None):
|
|
"""
|
|
Sends a confirmation code via email to the user. The code is only valid for the action specified by `reason`.
|
|
The email is either sent to the email address currently on file for the user, or to the one given in the optional `email` parameter.
|
|
A `state` value can be provided which is bound to this confirmation code, and returned on successfully checking the code.
|
|
:param session: the user's request session
|
|
:param reason: the action which should be confirmed using this confirmation code (currently, only `email_change` is allowed)
|
|
:param email: optional, the email address to send the confirmation code to
|
|
:param state: optional
|
|
"""
|
|
from pretix.base.services.mail import mail
|
|
|
|
with language(self.locale):
|
|
if reason == 'email_change':
|
|
msg = str(_('To change your email address from {old_email} to {new_email}, use the following code:').format(
|
|
old_email=self.email, new_email=email,
|
|
))
|
|
elif reason == 'email_verify':
|
|
msg = str(_('To verify your email address {email} on {instance}, use the following code:').format(
|
|
email=self.email,
|
|
instance=settings.PRETIX_INSTANCE_NAME,
|
|
))
|
|
else:
|
|
raise Exception('Invalid confirmation code reason')
|
|
|
|
code = "%07d" % secrets.SystemRandom().randint(0, 9999999)
|
|
session['user_confirmation_code:' + reason] = {
|
|
'code': code,
|
|
'state': state,
|
|
'attempts': 0,
|
|
}
|
|
mail(
|
|
email or self.email,
|
|
_('Your confirmation code'),
|
|
'pretixcontrol/email/confirmation_code.txt',
|
|
{
|
|
'user': self,
|
|
'reason': msg,
|
|
'code': code,
|
|
'instance': settings.PRETIX_INSTANCE_NAME,
|
|
},
|
|
event=None,
|
|
user=self,
|
|
locale=self.locale
|
|
)
|
|
|
|
def check_confirmation_code(self, session, reason, code):
|
|
"""
|
|
Checks a confirmation code entered by the user against the valid code stored in the session.
|
|
If the code is correct, an optional state bound to the code is returned.
|
|
If the code is incorrect, PermissionDenied is raised. If the code could not be validated, either because no
|
|
code for the given reason is stored, or the number of input attempts is exceeded, BadRequest is raised.
|
|
|
|
:param session: the user's request session
|
|
:param reason: the action which should be confirmed using this confirmation code
|
|
:param code: the code entered by the user
|
|
:return: optional state bound to this code using the state parameter of send_confirmation_code, None otherwise
|
|
"""
|
|
stored = session.get('user_confirmation_code:' + reason)
|
|
if not stored:
|
|
raise BadRequest
|
|
|
|
if stored['attempts'] > User.MAX_CONFIRMATION_CODE_ATTEMPTS:
|
|
raise BadRequest
|
|
|
|
if int(stored['code']) == int(code):
|
|
del session['user_confirmation_code:' + reason]
|
|
return stored['state']
|
|
else:
|
|
stored['attempts'] += 1
|
|
session['user_confirmation_code:' + reason] = stored
|
|
raise PermissionDenied
|
|
|
|
def send_password_reset(self):
|
|
from pretix.base.services.mail import mail
|
|
|
|
mail(
|
|
self.email,
|
|
_('Reset your password'),
|
|
'pretixcontrol/email/forgot.txt',
|
|
{
|
|
'instance': settings.PRETIX_INSTANCE_NAME,
|
|
'user': self,
|
|
'url': (mainreverse_absolute('control:auth.forgot.recover')
|
|
+ '?id=%d&token=%s' % (self.id, default_token_generator.make_token(self)))
|
|
},
|
|
None, locale=self.locale, user=self
|
|
)
|
|
|
|
@property
|
|
def top_logentries(self):
|
|
return self.all_logentries
|
|
|
|
@property
|
|
def all_logentries(self):
|
|
from pretix.base.models import LogEntry
|
|
|
|
return LogEntry.objects.filter(content_type=ContentType.objects.get_for_model(User),
|
|
object_id=self.pk)
|
|
|
|
def _get_teams_for_organizer(self, organizer):
|
|
if 'o{}'.format(organizer.pk) not in self._teamcache:
|
|
self._teamcache['o{}'.format(organizer.pk)] = list(self.teams.filter(organizer=organizer))
|
|
return self._teamcache['o{}'.format(organizer.pk)]
|
|
|
|
def _get_teams_for_event(self, organizer, event):
|
|
if 'e{}'.format(event.pk) not in self._teamcache:
|
|
self._teamcache['e{}'.format(event.pk)] = list(self.teams.filter(organizer=organizer).filter(
|
|
Q(all_events=True) | Q(limit_events=event)
|
|
))
|
|
return self._teamcache['e{}'.format(event.pk)]
|
|
|
|
def get_event_permission_set(self, organizer, event) -> set:
|
|
"""
|
|
Gets a set of permissions (as strings) that a user holds for a particular event
|
|
|
|
:param organizer: The organizer of the event
|
|
:param event: The event to check
|
|
:return: set
|
|
"""
|
|
teams = self._get_teams_for_event(organizer, event)
|
|
sets = [t.event_permission_set() for t in teams]
|
|
if sets:
|
|
return set.union(*sets)
|
|
else:
|
|
return set()
|
|
|
|
def get_organizer_permission_set(self, organizer) -> set:
|
|
"""
|
|
Gets a set of permissions (as strings) that a user holds for a particular organizer
|
|
|
|
:param organizer: The organizer of the event
|
|
:return: set
|
|
"""
|
|
teams = self._get_teams_for_organizer(organizer)
|
|
sets = [t.organizer_permission_set() for t in teams]
|
|
if sets:
|
|
return set.union(*sets)
|
|
else:
|
|
return set()
|
|
|
|
def has_event_permission(self, organizer, event, perm_name=None, request=None, session_key=None) -> bool:
|
|
"""
|
|
Checks if this user is part of any team that grants access of type ``perm_name``
|
|
to the event ``event``.
|
|
|
|
Either ``request`` or ``session_key`` are required to detect staff sessions properly.
|
|
|
|
:param organizer: The organizer of the event
|
|
:param event: The event to check
|
|
:param perm_name: The permission, e.g. ``event.orders:read``
|
|
:param request: The current request (optional)
|
|
:param session_key: The current session key (optional)
|
|
:return: bool
|
|
"""
|
|
assert not (session_key and request)
|
|
if (session_key or request) and self.has_active_staff_session(session_key or request.session.session_key):
|
|
return True
|
|
teams = self._get_teams_for_event(organizer, event)
|
|
if teams:
|
|
self._teamcache['e{}'.format(event.pk)] = teams
|
|
if isinstance(perm_name, (tuple, list)):
|
|
return any([any(team.has_event_permission(p) for team in teams) for p in perm_name])
|
|
if not perm_name or any([team.has_event_permission(perm_name) for team in teams]):
|
|
return True
|
|
return False
|
|
|
|
def has_organizer_permission(self, organizer, perm_name=None, request=None):
|
|
"""
|
|
Checks if this user is part of any team that grants access of type ``perm_name``
|
|
to the organizer ``organizer``.
|
|
|
|
:param organizer: The organizer to check
|
|
:param perm_name: The permission, e.g. ``organizer.events:create``
|
|
:param request: The current request (optional). Required to detect staff sessions properly.
|
|
:return: bool
|
|
"""
|
|
if request and self.has_active_staff_session(request.session.session_key):
|
|
return True
|
|
teams = self._get_teams_for_organizer(organizer)
|
|
if teams:
|
|
if isinstance(perm_name, (tuple, list)):
|
|
return any([any(team.has_organizer_permission(p) for team in teams) for p in perm_name])
|
|
if not perm_name or any([team.has_organizer_permission(perm_name) for team in teams]):
|
|
return True
|
|
return False
|
|
|
|
@scopes_disabled()
|
|
def get_events_with_any_permission(self, request=None):
|
|
"""
|
|
Returns a queryset of events the user has any permissions to.
|
|
|
|
:param request: The current request (optional). Required to detect staff sessions properly.
|
|
:return: Iterable of Events
|
|
"""
|
|
from .event import Event
|
|
|
|
if request and self.has_active_staff_session(request.session.session_key):
|
|
return Event.objects.all()
|
|
|
|
return Event.objects.filter(
|
|
Q(organizer_id__in=self.teams.filter(all_events=True).values_list('organizer', flat=True))
|
|
| Q(id__in=self.teams.values_list('limit_events__id', flat=True))
|
|
)
|
|
|
|
@scopes_disabled()
|
|
def get_events_with_permission(self, permission, request=None):
|
|
"""
|
|
Returns a queryset of events the user has a specific permissions to.
|
|
|
|
:param request: The current request (optional). Required to detect staff sessions properly.
|
|
:return: Iterable of Events
|
|
"""
|
|
from .event import Event
|
|
from .organizer import TeamQuerySet
|
|
|
|
if request and self.has_active_staff_session(request.session.session_key):
|
|
return Event.objects.all()
|
|
|
|
if isinstance(permission, (tuple, list)):
|
|
q = reduce(operator.or_, [TeamQuerySet.event_permission_q(p) for p in permission])
|
|
else:
|
|
q = TeamQuerySet.event_permission_q(permission)
|
|
|
|
return Event.objects.filter(
|
|
Q(organizer_id__in=self.teams.filter(q, all_events=True).values_list('organizer', flat=True))
|
|
| Q(id__in=self.teams.filter(q).values_list('limit_events__id', flat=True))
|
|
)
|
|
|
|
@scopes_disabled()
|
|
def get_organizers_with_any_permission(self, request=None):
|
|
"""
|
|
Returns a queryset of organizers the user has any permissions to.
|
|
|
|
:param request: The current request (optional). Required to detect staff sessions properly.
|
|
:return: Iterable of Organizers
|
|
"""
|
|
from .event import Organizer
|
|
|
|
if request and self.has_active_staff_session(request.session.session_key):
|
|
return Organizer.objects.all()
|
|
|
|
return Organizer.objects.filter(
|
|
id__in=self.teams.values_list('organizer', flat=True)
|
|
)
|
|
|
|
@scopes_disabled()
|
|
def get_organizers_with_permission(self, permission, request=None):
|
|
"""
|
|
Returns a queryset of organizers the user has a specific permissions to.
|
|
|
|
:param request: The current request (optional). Required to detect staff sessions properly.
|
|
:return: Iterable of Organizers
|
|
"""
|
|
from .event import Organizer
|
|
from .organizer import TeamQuerySet
|
|
|
|
if request and self.has_active_staff_session(request.session.session_key):
|
|
return Organizer.objects.all()
|
|
|
|
return Organizer.objects.filter(
|
|
id__in=self.teams.filter(TeamQuerySet.organizer_permission_q(permission)).values_list('organizer', flat=True)
|
|
)
|
|
|
|
def has_active_staff_session(self, session_key):
|
|
"""
|
|
Returns whether or not a user has an active staff session (formerly known as superuser session)
|
|
with the given session key.
|
|
"""
|
|
return self.get_active_staff_session(session_key) is not None
|
|
|
|
def get_active_staff_session(self, session_key):
|
|
if not self.is_staff or not session_key:
|
|
return None
|
|
if not hasattr(self, '_staff_session_cache'):
|
|
self._staff_session_cache = {}
|
|
if session_key not in self._staff_session_cache:
|
|
sess = StaffSession.objects.filter(
|
|
user=self, date_end__isnull=True, session_key=session_key
|
|
).first()
|
|
if sess:
|
|
if sess.date_start < now() - timedelta(seconds=settings.PRETIX_SESSION_TIMEOUT_ABSOLUTE):
|
|
sess.date_end = now()
|
|
sess.save()
|
|
sess = None
|
|
|
|
self._staff_session_cache[session_key] = sess
|
|
return self._staff_session_cache[session_key]
|
|
|
|
def get_session_auth_hash(self):
|
|
"""
|
|
Return an HMAC that needs to be the same throughout the session, used e.g. for forced
|
|
logout after every password change.
|
|
"""
|
|
return self._get_session_auth_hash(secret=settings.SECRET_KEY)
|
|
|
|
def get_session_auth_fallback_hash(self):
|
|
for fallback_secret in settings.SECRET_KEY_FALLBACKS:
|
|
yield self._get_session_auth_hash(secret=fallback_secret)
|
|
|
|
def _get_session_auth_hash(self, secret):
|
|
"""
|
|
"""
|
|
key_salt = "pretix.base.models.User.get_session_auth_hash"
|
|
payload = self.password
|
|
payload += self.email
|
|
payload += self.session_token
|
|
return salted_hmac(key_salt, payload, secret=secret).hexdigest()
|
|
|
|
def update_session_token(self):
|
|
self.session_token = generate_session_token()
|
|
self.save(update_fields=['session_token'])
|
|
|
|
@cached_property
|
|
@scopes_disabled()
|
|
def is_in_any_teams(self):
|
|
return self.teams.exists()
|
|
|
|
|
|
class UserWithStaffSession:
|
|
# Wrapper around a User object with a staff session, implementing the PermissionHolder Protocol
|
|
def __init__(self, user):
|
|
self.user = user
|
|
|
|
def has_event_permission(self, organizer, event, perm_name=None, request=None, session_key=None) -> bool:
|
|
return True
|
|
|
|
def has_organizer_permission(self, organizer, perm_name=None, request=None):
|
|
return True
|
|
|
|
|
|
class UserKnownLoginSource(models.Model):
|
|
user = models.ForeignKey('User', on_delete=models.CASCADE, related_name="known_login_sources")
|
|
agent_type = models.CharField(max_length=255, null=True, blank=True)
|
|
device_type = models.CharField(max_length=255, null=True, blank=True)
|
|
os_type = models.CharField(max_length=255, null=True, blank=True)
|
|
country = FastCountryField(null=True, blank=True)
|
|
last_seen = models.DateTimeField()
|
|
|
|
class Meta:
|
|
unique_together = ('user', 'agent_type', 'device_type', 'os_type', 'country')
|
|
|
|
|
|
class StaffSession(models.Model):
|
|
user = models.ForeignKey('User', on_delete=models.PROTECT)
|
|
date_start = models.DateTimeField(auto_now_add=True)
|
|
date_end = models.DateTimeField(null=True, blank=True)
|
|
session_key = models.CharField(max_length=255)
|
|
comment = models.TextField()
|
|
|
|
class Meta:
|
|
ordering = ('date_start',)
|
|
|
|
|
|
class StaffSessionAuditLog(models.Model):
|
|
session = models.ForeignKey('StaffSession', related_name='logs', on_delete=models.PROTECT)
|
|
datetime = models.DateTimeField(auto_now_add=True)
|
|
url = models.CharField(max_length=255)
|
|
method = models.CharField(max_length=255)
|
|
impersonating = models.ForeignKey('User', null=True, blank=True, on_delete=models.PROTECT)
|
|
|
|
class Meta:
|
|
ordering = ('datetime',)
|
|
|
|
|
|
class U2FDevice(Device):
|
|
json_data = models.TextField()
|
|
|
|
@property
|
|
def webauthndevice(self):
|
|
from webauthn.helpers.structs import PublicKeyCredentialDescriptor
|
|
|
|
d = json.loads(self.json_data)
|
|
return PublicKeyCredentialDescriptor(websafe_decode(d['keyHandle']))
|
|
|
|
@property
|
|
def webauthnpubkey(self):
|
|
d = json.loads(self.json_data)
|
|
# We manually need to convert the pubkey from DER format (used in our
|
|
# former U2F implementation) to the format required by webauthn. This
|
|
# is based on the following example:
|
|
# https://www.w3.org/TR/webauthn/#sctn-encoded-credPubKey-examples
|
|
pub_key = pub_key_from_der(websafe_decode(d['publicKey'].replace('+', '-').replace('/', '_')))
|
|
pub_key = binascii.unhexlify(
|
|
'A5010203262001215820{:064x}225820{:064x}'.format(
|
|
pub_key.public_numbers().x, pub_key.public_numbers().y
|
|
)
|
|
)
|
|
return pub_key
|
|
|
|
|
|
class WebAuthnDevice(Device):
|
|
credential_id = models.CharField(max_length=255, null=True, blank=True)
|
|
rp_id = models.CharField(max_length=255, null=True, blank=True)
|
|
icon_url = models.CharField(max_length=255, null=True, blank=True)
|
|
ukey = models.TextField(null=True)
|
|
pub_key = models.TextField(null=True)
|
|
sign_count = models.IntegerField(default=0)
|
|
|
|
@property
|
|
def webauthndevice(self):
|
|
from webauthn.helpers.structs import PublicKeyCredentialDescriptor
|
|
|
|
return PublicKeyCredentialDescriptor(websafe_decode(self.credential_id))
|
|
|
|
@property
|
|
def webauthnpubkey(self):
|
|
return websafe_decode(self.pub_key)
|
|
|
|
|
|
class HistoricPassword(models.Model):
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="historic_passwords")
|
|
created = models.DateTimeField(auto_now_add=True)
|
|
password = models.CharField(verbose_name=_("Password"), max_length=128)
|