mirror of
https://github.com/i701/sarlink-portal-api.git
synced 2025-02-21 17:51:59 +00:00
119 lines
5.8 KiB
Python
119 lines
5.8 KiB
Python
import logging
|
|
from django.contrib.auth import get_user_model
|
|
from django.dispatch import receiver
|
|
from django.db.models import signals
|
|
from djangopasswordlessknox.models import CallbackToken
|
|
from djangopasswordlessknox.models import generate_numeric_token
|
|
from djangopasswordlessknox.settings import api_settings
|
|
from djangopasswordlessknox.services import TokenService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@receiver(signals.pre_save, sender=CallbackToken)
|
|
def invalidate_previous_tokens(sender, instance, **kwargs):
|
|
"""
|
|
Invalidates all previously issued tokens as a post_save signal.
|
|
"""
|
|
active_tokens = None
|
|
if isinstance(instance, CallbackToken):
|
|
active_tokens = CallbackToken.objects.active().filter(user=instance.user).exclude(id=instance.id)
|
|
|
|
# Invalidate tokens
|
|
if active_tokens:
|
|
for token in active_tokens:
|
|
token.is_active = False
|
|
token.save()
|
|
|
|
|
|
@receiver(signals.pre_save, sender=CallbackToken)
|
|
def check_unique_tokens(sender, instance, **kwargs):
|
|
"""
|
|
Ensures that mobile and email tokens are unique or tries once more to generate.
|
|
"""
|
|
if isinstance(instance, CallbackToken):
|
|
if CallbackToken.objects.filter(key=instance.key, is_active=True).exists():
|
|
instance.key = generate_numeric_token()
|
|
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
@receiver(signals.pre_save, sender=User)
|
|
def update_alias_verification(sender, instance, **kwargs):
|
|
"""
|
|
Flags a user's email as unverified if they change it.
|
|
Optionally sends a verification token to the new endpoint.
|
|
"""
|
|
if isinstance(instance, User):
|
|
|
|
if instance.id:
|
|
|
|
if api_settings.PASSWORDLESS_USER_MARK_EMAIL_VERIFIED is True:
|
|
"""
|
|
For marking email aliases as not verified when a user changes it.
|
|
"""
|
|
email_field = api_settings.PASSWORDLESS_USER_EMAIL_FIELD_NAME
|
|
email_verified_field = api_settings.PASSWORDLESS_USER_EMAIL_VERIFIED_FIELD_NAME
|
|
|
|
# Verify that this is an existing instance and not a new one.
|
|
try:
|
|
user_old = User.objects.get(id=instance.id) # Pre-save object
|
|
instance_email = getattr(instance, email_field) # Incoming Email
|
|
old_email = getattr(user_old, email_field) # Pre-save object email
|
|
|
|
if instance_email != old_email and instance_email != "" and instance_email is not None:
|
|
# Email changed, verification should be flagged
|
|
setattr(instance, email_verified_field, False)
|
|
if api_settings.PASSWORDLESS_AUTO_SEND_VERIFICATION_TOKEN is True:
|
|
email_subject = api_settings.PASSWORDLESS_EMAIL_VERIFICATION_SUBJECT
|
|
email_plaintext = api_settings.PASSWORDLESS_EMAIL_VERIFICATION_PLAINTEXT_MESSAGE
|
|
email_html = api_settings.PASSWORDLESS_EMAIL_VERIFICATION_TOKEN_HTML_TEMPLATE_NAME
|
|
message_payload = {'email_subject': email_subject,
|
|
'email_plaintext': email_plaintext,
|
|
'email_html': email_html}
|
|
success = TokenService.send_token(instance, 'email', **message_payload)
|
|
|
|
if success:
|
|
logger.info('djangopasswordlessknox: Successfully sent email on updated address: %s'
|
|
% instance_email)
|
|
else:
|
|
logger.info('djangopasswordlessknox: Failed to send email to updated address: %s'
|
|
% instance_email)
|
|
|
|
except User.DoesNotExist:
|
|
# User probably is just initially being created
|
|
setattr(instance, email_verified_field, True)
|
|
|
|
if api_settings.PASSWORDLESS_USER_MARK_MOBILE_VERIFIED is True:
|
|
"""
|
|
For marking mobile aliases as not verified when a user changes it.
|
|
"""
|
|
mobile_field = api_settings.PASSWORDLESS_USER_MOBILE_FIELD_NAME
|
|
mobile_verified_field = api_settings.PASSWORDLESS_USER_MOBILE_VERIFIED_FIELD_NAME
|
|
|
|
# Verify that this is an existing instance and not a new one.
|
|
try:
|
|
user_old = User.objects.get(id=instance.id) # Pre-save object
|
|
instance_mobile = getattr(instance, mobile_field) # Incoming mobile
|
|
old_mobile = getattr(user_old, mobile_field) # Pre-save object mobile
|
|
|
|
if instance_mobile != old_mobile and instance_mobile != "" and instance_mobile is not None:
|
|
# Mobile changed, verification should be flagged
|
|
setattr(instance, mobile_verified_field, False)
|
|
if api_settings.PASSWORDLESS_AUTO_SEND_VERIFICATION_TOKEN is True:
|
|
mobile_message = api_settings.PASSWORDLESS_MOBILE_MESSAGE
|
|
message_payload = {'mobile_message': mobile_message}
|
|
success = TokenService.send_token(instance, 'mobile', **message_payload)
|
|
|
|
if success:
|
|
logger.info('djangopasswordlessknox: Successfully sent SMS on updated mobile: %s'
|
|
% instance_mobile)
|
|
else:
|
|
logger.info('djangopasswordlessknox: Failed to send SMS to updated mobile: %s'
|
|
% instance_mobile)
|
|
|
|
except User.DoesNotExist:
|
|
# User probably is just initially being created
|
|
setattr(instance, mobile_verified_field, True)
|