1
\$\begingroup\$

I created a two steps changing email system. Firstly I show you the concept how it should work and then the code. I am using Django and React.

Main question: is it a good way to do it? Is it fine to use Redis in this situation? Can I improve something in this code?

Concept

When the user wants to change their email address, in the settings page they enter their accout password and click "send email changing link" button. After that, if the password was correct, Django generates a token using the PasswordResetTokenGenerator class (the token is saved in Redis like this: key: "email_changing_message_" + str(user.id), value: token) and sends a message to the current user's email address. In this message there is a link to set new email page. This link contains the token and their user.id like this: /new-email/:uidb64/:token.

On the /new-email page the user needs to enter their password again, the new email address and repeat the new email address. This data, uidb64 and token is sent to Django. Django checks if the password is correct, if new_email != old_email and if token is stored in Redis. If so, the new email address is set in Redis like this: key: "new_email_" + str(uid), value: new_email, the email_changing_message_ token is removed from Redis and a new token is generated (but now it is not saved in Redis, because the next step is activation and if the email field changes in User model record, the token will be deactivated). A new email activation message is sent to the user's new email address to verify if it is real. This message contains a link like this: /email-activation/:uidb64/:token.

On the "/email-activation" there is just information for the user that the new email has been activated (if the token was correct). The new_email_ is removed from Redis.

Code

@api_view(['POST'])
@permission_classes([IsAuthenticated])
def send_email_changing_message(request):
    password = request.data['password']
    user = request.user

    if not user.check_password(password):
        return Response(status=status.HTTP_401_UNAUTHORIZED)

    token = email_changing_token_generator.make_token(user)
    cache.set("email_changing_message_" + str(user.id), token, 
    settings.PASSWORD_RESET_TIMEOUT)
    
    email_subject = "..."
    email_body = render_to_string('.../....html', {
        'user': user,
        'domain': ...,
        'uidb64': urlsafe_base64_encode(force_bytes(user.pk)),
        'token': token
    })

    email=EmailMessage(subject=email_subject, body=email_body, 
    from_email=settings.EMAIL_FROM_USER, to=[user])
    email.content_subtype='html'
    email.send()
    return Response(status=status.HTTP_200_OK)
@api_view(['PUT'])
def set_new_email(request, uidb64, token):
    password = request.data['password']
    new_email = request.data['new_email']

    try:
        uid = force_str(urlsafe_b64decode(uidb64))
        user = User.object.get(pk=uid)
    except:
        user=None
    
    if user and cache.get("email_changing_message_" + str(user.id)) == token:
        if not user.check_password(password):
            return Response(status=status.HTTP_401_UNAUTHORIZED)
        
        if new_email == user.email:
            return Response(status=status.HTTP_403_FORBIDDEN)

        cache.set("new_email_" + str(uid), new_email, settings.PASSWORD_RESET_TIMEOUT)
        
        email_subject = "..."
        email_body = render_to_string('.../....html', {
            'user': user,
            'domain': ...,
            'uidb64': urlsafe_base64_encode(force_bytes(user.pk)),
            'token': email_changing_token_generator.make_token(user)
        })

        email=EmailMessage(subject=email_subject, body=email_body, 
        from_email=settings.EMAIL_FROM_USER, to=[new_email])
        email.content_subtype='html'
        email.send()
        cache.delete("email_changing_message_" + str(user.id))

        return Response(status=status.HTTP_200_OK)
    return Response(status=status.HTTP_400_BAD_REQUEST)
@api_view(['POST'])
def activate_new_email(request, uidb64, token):
    try:
        uid = force_str(urlsafe_b64decode(uidb64))
        user = User.object.get(pk=uid)
    except Exception as e:
        user=None

    if user and email_changing_token_generator.check_token(user,token):
        user.email = cache.get("new_email_" + str(user.id))
        user.save()
        cache.delete("new_email_" + str(user.id))
        return Response(status=status.HTTP_200_OK)
    return Response(status=status.HTTP_403_FORBIDDEN)
class TokenGenerator(PasswordResetTokenGenerator):
    def __init__(self, user_field) -> None:
        super().__init__()
        self.user_field = user_field

    def _make_hash_value(self, user, timestamp):
        return (six.text_type(user.pk) + six.text_type(timestamp) + 
                six.text_type(user[self.user_field]))

email_changing_token_generator = TokenGenerator('email')
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.