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')