Egor fureunoir Gorbunov
ad320235d6
- Refactored monetary fields across models to use `DecimalField` for improved precision. - Implemented two-factor authentication (2FA) for admin logins with OTP codes. - Added ability to generate admin OTP via management commands. - Updated Docker Compose override for dev-specific port bindings. - Included template for 2FA OTP verification to enhance security. Additional changes: - Upgraded and downgraded various dependencies (e.g., django-celery-beat and yarl). - Replaced float-based calculations with decimal for consistent rounding behavior. - Improved admin user management commands for activation and OTP generation.
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
from celery.app import shared_task
|
|
from constance import config
|
|
from django.conf import settings
|
|
from django.contrib.auth.tokens import PasswordResetTokenGenerator
|
|
from django.core.mail import EmailMessage
|
|
from django.template.loader import render_to_string
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.http import urlsafe_base64_encode
|
|
from django.utils.translation import activate
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from engine.core.utils import get_dynamic_email_connection
|
|
from engine.vibes_auth.models import User
|
|
|
|
|
|
@shared_task(queue="default")
|
|
def send_verification_email_task(user_pk: str) -> tuple[bool, str]:
|
|
try:
|
|
user = User.objects.get(pk=user_pk)
|
|
user.refresh_from_db()
|
|
|
|
raw_token = user.refresh_activation_token()
|
|
user.save(update_fields=["activation_token", "activation_token_created"])
|
|
|
|
activate(user.language)
|
|
|
|
email_subject = _(f"{settings.PROJECT_NAME} | Activate Account")
|
|
email_body = render_to_string(
|
|
"../templates/user_verification_email.html",
|
|
{
|
|
"user_first_name": user.first_name,
|
|
"activation_link": f"https://{settings.STOREFRONT_DOMAIN}/{user.language}/activate-user?uid={urlsafe_base64_encode(force_bytes(user.uuid))}"
|
|
f"&token={urlsafe_base64_encode(force_bytes(raw_token))}",
|
|
"project_name": settings.PROJECT_NAME,
|
|
},
|
|
)
|
|
|
|
email = EmailMessage(
|
|
subject=email_subject,
|
|
body=email_body,
|
|
from_email=f"{settings.PROJECT_NAME} <{config.EMAIL_FROM}>",
|
|
to=[user.email],
|
|
connection=get_dynamic_email_connection(),
|
|
)
|
|
email.content_subtype = "html"
|
|
email.send()
|
|
|
|
except User.DoesNotExist:
|
|
return False, f"User not found with the given pk: {user_pk}"
|
|
|
|
except Exception as e:
|
|
return False, f"Something went wrong while sending an email: {e!s}"
|
|
|
|
else:
|
|
return True, str(user.uuid)
|
|
|
|
|
|
@shared_task(queue="default")
|
|
def send_reset_password_email_task(user_pk: str) -> tuple[bool, str]:
|
|
try:
|
|
user = User.objects.get(pk=user_pk)
|
|
user.refresh_from_db()
|
|
|
|
activate(user.language)
|
|
|
|
email_subject = _(f"{settings.PROJECT_NAME} | Reset Password")
|
|
email_body = render_to_string(
|
|
"../templates/user_reset_password_email.html",
|
|
{
|
|
"user_first_name": user.first_name,
|
|
"reset_link": f"https://{settings.STOREFRONT_DOMAIN}/{user.language}/reset-password?uid="
|
|
f"{urlsafe_base64_encode(force_bytes(user.pk))}"
|
|
f"&token={PasswordResetTokenGenerator().make_token(user)}",
|
|
"project_name": settings.PROJECT_NAME,
|
|
},
|
|
)
|
|
|
|
email = EmailMessage(
|
|
subject=email_subject,
|
|
body=email_body,
|
|
from_email=f"{settings.PROJECT_NAME} <{config.EMAIL_FROM}>",
|
|
to=[user.email],
|
|
connection=get_dynamic_email_connection(),
|
|
)
|
|
email.content_subtype = "html"
|
|
email.send()
|
|
|
|
except User.DoesNotExist:
|
|
return False, f"User not found with the given pk: {user_pk}"
|
|
|
|
except Exception as e:
|
|
return False, f"Something went wrong while sending an email: {e!s}"
|
|
|
|
else:
|
|
return True, str(user.uuid)
|