Egor fureunoir Gorbunov
adec5503b2
Add encryption for user PII fields (phone number, name, attributes) and address fields to enhance data security. Introduced timestamped activation tokens for improved validation. Included migrations to encrypt existing plaintext data. Refactored GraphQL settings to limit query depth and optionally disable introspection for enhanced API defense. Implemented throttling to safeguard API rates. Improved Dockerfiles for better user management and restored media migration tools for smooth instance upgrades.
255 lines
8.2 KiB
Python
255 lines
8.2 KiB
Python
from uuid import uuid4
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractUser
|
|
from django.contrib.auth.models import Group as BaseGroup
|
|
from django.core.cache import cache
|
|
from django.core.exceptions import ValidationError
|
|
from django.db.models import (
|
|
CASCADE,
|
|
SET_NULL,
|
|
BooleanField,
|
|
CharField,
|
|
DateTimeField,
|
|
EmailField,
|
|
ForeignKey,
|
|
ImageField,
|
|
Index,
|
|
JSONField,
|
|
TextField,
|
|
UUIDField,
|
|
)
|
|
from django.templatetags.static import static
|
|
from django.utils import timezone
|
|
from django.utils.functional import cached_property
|
|
from django.utils.translation import gettext_lazy as _
|
|
from encrypted_fields.fields import EncryptedCharField
|
|
from rest_framework_simplejwt.token_blacklist.models import (
|
|
BlacklistedToken as BaseBlacklistedToken,
|
|
)
|
|
from rest_framework_simplejwt.token_blacklist.models import (
|
|
OutstandingToken as BaseOutstandingToken,
|
|
)
|
|
|
|
from engine.core.abstract import NiceModel
|
|
from engine.payments.models import Balance
|
|
from engine.vibes_auth.choices import SenderType, ThreadStatus
|
|
from engine.vibes_auth.managers import UserManager
|
|
from engine.vibes_auth.validators import validate_phone_number
|
|
from schon.fields import EncryptedJSONTextField
|
|
|
|
|
|
class User(AbstractUser, NiceModel):
|
|
__doc__ = _(
|
|
"Represents a User entity with customized fields and methods for extended functionality. "
|
|
"This class extends the AbstractUser model and integrates additional features like "
|
|
"custom email login, validation methods, subscription status, verification, and "
|
|
"attributes storage. It also provides utilities for managing recently viewed items and "
|
|
"token-based activation for verifying accounts. The User model is designed to handle "
|
|
"specific use cases for enhanced user management."
|
|
)
|
|
|
|
def get_uuid_as_path(self, *args):
|
|
return "users/" + str(self.uuid) + "/" + args[0]
|
|
|
|
email = EmailField(_("email"), unique=True, help_text=_("user email address"))
|
|
phone_number = EncryptedCharField(
|
|
_("phone_number"),
|
|
max_length=20,
|
|
blank=True,
|
|
null=True,
|
|
help_text=_("user phone number"),
|
|
validators=[
|
|
validate_phone_number,
|
|
],
|
|
)
|
|
username: None = None
|
|
first_name = EncryptedCharField(
|
|
_("first_name"), max_length=150, blank=True, null=True
|
|
)
|
|
last_name = EncryptedCharField(
|
|
_("last_name"), max_length=150, blank=True, null=True
|
|
)
|
|
avatar = ImageField(
|
|
null=True,
|
|
verbose_name=_("avatar"),
|
|
upload_to=get_uuid_as_path,
|
|
blank=True,
|
|
help_text=_("user profile image"),
|
|
)
|
|
|
|
is_verified = BooleanField(
|
|
default=False,
|
|
verbose_name=_("is verified"),
|
|
help_text=_("user verification status"),
|
|
)
|
|
is_active = BooleanField(
|
|
_("is_active"),
|
|
default=False,
|
|
help_text=_("unselect this instead of deleting accounts"),
|
|
)
|
|
is_subscribed = BooleanField(
|
|
verbose_name=_("is_subscribed"),
|
|
help_text=_("user's newsletter subscription status"),
|
|
default=False,
|
|
)
|
|
|
|
activation_token = UUIDField(default=uuid4, verbose_name=_("activation token"))
|
|
activation_token_created = DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("activation token created"),
|
|
)
|
|
unsubscribe_token = UUIDField(
|
|
default=uuid4,
|
|
verbose_name=_("unsubscribe token"),
|
|
help_text=_("token for secure one-click unsubscribe from campaigns"),
|
|
)
|
|
language = CharField(
|
|
choices=settings.LANGUAGES,
|
|
default=settings.LANGUAGE_CODE,
|
|
null=False,
|
|
blank=False,
|
|
max_length=7,
|
|
)
|
|
attributes = EncryptedJSONTextField(
|
|
verbose_name=_("attributes"), default=dict, blank=True, null=True
|
|
)
|
|
|
|
payments_balance: "Balance"
|
|
|
|
USERNAME_FIELD = "email"
|
|
REQUIRED_FIELDS = []
|
|
# noinspection PyClassVar
|
|
objects = UserManager()
|
|
|
|
@cached_property
|
|
def avatar_url(self) -> str:
|
|
try:
|
|
return self.avatar.url
|
|
except ValueError:
|
|
return static("person.png")
|
|
|
|
def add_to_recently_viewed(self, product_uuid):
|
|
recently_viewed = self.recently_viewed
|
|
if product_uuid not in recently_viewed:
|
|
if not len(recently_viewed) >= 48:
|
|
recently_viewed.append(product_uuid)
|
|
cache.set(f"user_{self.uuid}_rv", recently_viewed)
|
|
else:
|
|
recently_viewed.pop(0)
|
|
recently_viewed.append(product_uuid)
|
|
cache.set(f"user_{self.uuid}_rv", recently_viewed)
|
|
|
|
@property
|
|
def recently_viewed(self):
|
|
return cache.get(f"user_{self.uuid}_rv", [])
|
|
|
|
def save(self, *args, **kwargs):
|
|
if self._state.adding and self.activation_token_created is None:
|
|
self.activation_token_created = timezone.now()
|
|
super().save(*args, **kwargs)
|
|
|
|
def refresh_activation_token(self) -> None:
|
|
"""Generate a fresh activation token and update its timestamp."""
|
|
self.activation_token = uuid4()
|
|
self.activation_token_created = timezone.now()
|
|
|
|
def check_token(self, token) -> bool:
|
|
from datetime import timedelta
|
|
|
|
if str(token) != str(self.activation_token):
|
|
return False
|
|
if self.activation_token_created:
|
|
if timezone.now() > self.activation_token_created + timedelta(hours=24):
|
|
return False
|
|
return True
|
|
|
|
def __str__(self):
|
|
return self.email
|
|
|
|
class Meta:
|
|
swappable = "AUTH_USER_MODEL"
|
|
verbose_name = _("user")
|
|
verbose_name_plural = _("users")
|
|
|
|
|
|
class ChatThread(NiceModel):
|
|
user = ForeignKey(
|
|
User, null=True, blank=True, on_delete=SET_NULL, related_name="chat_threads"
|
|
)
|
|
email = EmailField(blank=True, default="", help_text=_("For anonymous threads"))
|
|
assigned_to = ForeignKey(
|
|
User,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=SET_NULL,
|
|
related_name="assigned_chat_threads",
|
|
)
|
|
status = CharField(
|
|
max_length=16, choices=ThreadStatus.choices, default=ThreadStatus.OPEN
|
|
)
|
|
last_message_at = DateTimeField(default=timezone.now)
|
|
attributes = JSONField(default=dict, blank=True)
|
|
|
|
class Meta:
|
|
indexes = [
|
|
Index(fields=["status", "modified"], name="chatthread_status_mod_idx"),
|
|
Index(
|
|
fields=["assigned_to", "status"], name="chatthread_assigned_status_idx"
|
|
),
|
|
Index(fields=["user"], name="chatthread_user_idx"),
|
|
Index(fields=["email"], name="chatthread_email_idx"),
|
|
]
|
|
ordering = ("-modified",)
|
|
verbose_name = _("Chat thread")
|
|
verbose_name_plural = _("Chat threads")
|
|
|
|
def clean(self) -> None:
|
|
super().clean()
|
|
if not self.user and not self.email:
|
|
raise ValidationError(
|
|
{"email": _("provide user or email for anonymous thread.")}
|
|
)
|
|
if self.assigned_to and not self.assigned_to.is_staff:
|
|
raise ValidationError({"assigned_to": _("assignee must be a staff user.")})
|
|
|
|
|
|
class ChatMessage(NiceModel):
|
|
thread = ForeignKey(ChatThread, on_delete=CASCADE, related_name="messages")
|
|
sender_type = CharField(max_length=16, choices=SenderType.choices)
|
|
sender_user = ForeignKey(
|
|
User, null=True, blank=True, on_delete=SET_NULL, related_name="chat_messages"
|
|
)
|
|
text = TextField()
|
|
sent_at = DateTimeField(default=timezone.now)
|
|
attributes = JSONField(default=dict, blank=True)
|
|
|
|
class Meta:
|
|
indexes = [
|
|
Index(fields=["thread", "sent_at"], name="chatmessage_thread_sent_idx"),
|
|
]
|
|
ordering = ("sent_at", "uuid")
|
|
verbose_name = _("Chat message")
|
|
verbose_name_plural = _("Chat messages")
|
|
|
|
|
|
class Group(BaseGroup):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("group")
|
|
verbose_name_plural = _("groups")
|
|
|
|
|
|
class OutstandingToken(BaseOutstandingToken):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("outstanding token")
|
|
verbose_name_plural = _("outstanding tokens")
|
|
|
|
|
|
class BlacklistedToken(BaseBlacklistedToken):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("blacklisted token")
|
|
verbose_name_plural = _("blacklisted tokens")
|