from .email_handler import *



verify_email/__init__.py
pyguru-dev/django-veify-email
from django.contrib import admin
from .models import LinkCounter

admin.site.register(LinkCounter)

verify_email/admin.py
pyguru-dev/django-veify-email
from django.conf import settings


class GetFieldFromSettings:
    """
    This class fetches the attributes that are defined in settings.py of your project by user OR Django itself.
    self.default_configs : is a dict with keys as the names used in this app and values being a tuple of 
                           attributes defined in settings.py and their corresponding default values if not found.
    
    There is a special case in "get" method, if you set "VERIFICATION_SUCCESS_TEMPLATE" as None is settings.py, it 
    will skip the intermediate page where success information is displayed. (This is better explained in docs.)

    The "get" method takes the name of the attributes as input, checks for it in settings.py, 
            if found:
                returns the corresponding value.
            else:
                returns the default value from "self.defaults_configs".
    """

    def __init__(self):

        self.defaults_configs = {
            'debug_settings': (
                'DEBUG',
                False
            ),

            'subject': (
                "SUBJECT",
                "Email Verification Mail"
            ),

            'email_field_name': (
                "EMAIL_FIELD_NAME",
                "email",
            ),

            'html_message_template': (
                "HTML_MESSAGE_TEMPLATE",
                'verify_email/email_verification_msg.html'
            ),

            'from_alias': (
                "DEFAULT_FROM_EMAIL",
                'noreply<noreply@gmail.com>',
            ),

            'login_page': (
                'LOGIN_URL',
                'accounts_login'
            ),

            'verification_success_template': (
                'VERIFICATION_SUCCESS_TEMPLATE',
                'verify_email/email_verification_successful.html'
            ),

            'verification_success_msg': (
                'VERIFICATION_SUCCESS_MSG',
                "Your Email is verified successfully and account has been activated. "
                "You can login with the credentials now..."
            ),

            'verification_failed_template': (
                'VERIFICATION_FAILED_TEMPLATE',
                'verify_email/email_verification_failed.html'
            ),

            'link_expired_template': (
                'LINK_EXPIRED_TEMPLATE',
                'verify_email/link_expired.html'
            ),

            'verification_failed_msg': (
                'VERIFICATION_FAILED_MSG',
                "There is something wrong with this link, can't verify the user..."
            ),
            'request_new_email_template': (
                'REQUEST_NEW_EMAIL_TEMPLATE',
                'verify_email/request_new_email.html'
            ),
            'new_email_sent_template': (
                'NEW_EMAIL_SENT_TEMPLATE',
                'verify_email/new_email_sent.html'
            ),

            'salt': (
                'HASH_SALT',
                None
            ),

            'sep': (
                'SEPARATOR',
                ':'
            ),

            'key': (
                'HASHING_KEY',
                None
            ),

            'max_age': (
                'EXPIRE_AFTER',
                None
            ),

            'max_retries': (
                'MAX_RETRIES',
                2
            )
        }

    def get(self, field_name, raise_exception=True, default_type=str):
        attr = getattr(
            settings,
            self.defaults_configs[field_name][0],  # get field from settings
            self.defaults_configs[field_name][1]  # get default value if field not defined
        )
        if (attr == '' or attr is None or not isinstance(field_name, default_type)) and raise_exception:
            if field_name == 'verification_success_template' and attr is None:
                return None
            raise AttributeError
        return attr

verify_email/app_configurations.py
pyguru-dev/django-veify-email
import logging

from django.apps import AppConfig

logger = logging.getLogger(__name__)


class VerifyEmailConfig(AppConfig):
    name = 'verify_email'

    def ready(self):
        logger.info('[Email Verification] : importing signals    - OK.')
        import verify_email.signals

verify_email/apps.py
pyguru-dev/django-veify-email
from django.utils import timezone
from .token_manager import TokenManager


class _UserActivationProcess:
    """
    This class is pretty self.explanatory...
    Exceptions Raised
    ------------------
        - MaxRetriesExceeded    (by) verify_user()
        - BadSignature          (by) verify_user()
        - SignatureExpired      (by) verify_user()
        - ValueError            (by) verify_user()
        - TypeError             (by) verify_user()
        - InvalidToken          (by) verify_user()
    """

    def __init__(self):
        self.token_manager = TokenManager()

    @staticmethod
    def __activate_user(user):
        user.is_active = True
        user.last_login = timezone.now()
        user.save()

    def verify_token(self, useremail, usertoken):
        user = self.token_manager.decrypt_link(useremail, usertoken)
        if user:
            self.__activate_user(user)
            return True
        return False


def verify_user(useremail, usertoken):
    return _UserActivationProcess().verify_token(useremail, usertoken)

verify_email/confirm.py
pyguru-dev/django-veify-email
from django.core.mail import send_mail
from django.template.loader import render_to_string
from django.utils.html import strip_tags

from .app_configurations import GetFieldFromSettings
from .errors import InvalidTokenOrEmail
from .token_manager import TokenManager


class _VerifyEmail:
    """
    This class does two things:
    1. set each user as inactive and saves it
    2. sends the email to user with that link.
    """

    def __init__(self):
        self.settings = GetFieldFromSettings()
        self.token_manager = TokenManager()

    # Private :
    def __send_email(self, msg, useremail):
        subject = self.settings.get('subject')
        send_mail(
            subject, strip_tags(msg),
            from_email=self.settings.get('from_alias'),
            recipient_list=[useremail], html_message=msg
        )

    # Public :
    def send_verification_link(self, request, inactive_user=None, form=None):
        
        if form:
            inactive_user = form.save(commit=False)
        
        inactive_user.is_active = False
        inactive_user.save()

        try:
            
            useremail = form.cleaned_data.get(self.settings.get('email_field_name')) if form else inactive_user.email
            if not useremail:
                raise KeyError(
                    'No key named "email" in your form. Your field should be named as email in form OR set a variable'
                    ' "EMAIL_FIELD_NAME" with the name of current field in settings.py if you want to use current name '
                    'as email field.'
                )

            verification_url = self.token_manager.generate_link(request, inactive_user, useremail)
            msg = render_to_string(
                self.settings.get('html_message_template', raise_exception=True),
                {"link": verification_url, "inactive_user": inactive_user}, 
                request=request
            )

            self.__send_email(msg, useremail)
            return inactive_user
        except Exception:
            inactive_user.delete()
            raise

    def resend_verification_link(self, request, email, **kwargs):
        """
        This method needs the previously sent link to get encoded email and token from that.
        Exceptions Raised
        -----------------
            - UserAlreadyActive (by) get_user_by_token()
            - MaxRetryExceeded  (by) request_new_link()
            - InvalidTokenOrEmail
        
        These exception should be handled in caller function.
        """
        inactive_user = kwargs.get('user')
        user_encoded_token = kwargs.get('token')
        encoded = kwargs.get('encoded', True)

        if encoded:
            decoded_encrypted_user_token = self.token_manager.perform_decoding(user_encoded_token)
            email = self.token_manager.perform_decoding(email)
            inactive_user = self.token_manager.get_user_by_token(email, decoded_encrypted_user_token)

        if not inactive_user or not email:
            raise InvalidTokenOrEmail(f'Either token or email is invalid. user: {inactive_user}, email: {email}')

        # At this point, we have decoded email(if it was encoded), and inactive_user, and we can request new link
        link = self.token_manager.request_new_link(request, inactive_user, email)
        msg = render_to_string(
            self.settings.get('html_message_template', raise_exception=True),
            {"link": link}, request=request
        )
        self.__send_email(msg, email)
        return True



#  These is supposed to be called outside of this module
def send_verification_email(request, form):
    return _VerifyEmail().send_verification_link(request, form)


#  These is supposed to be called outside of this module
def resend_verification_email(request, email, **kwargs):
    return _VerifyEmail().resend_verification_link(request, email, **kwargs)

verify_email/email_handler.py
pyguru-dev/django-veify-email
class UserAlreadyActive(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)


class MaxRetriesExceeded(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)


class UserNotFound(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)


class InvalidToken(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)


class InvalidTokenOrEmail(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)


class WrongTimeInterval(Exception):
    def __init__(self, *args: object) -> None:
        super().__init__(*args)

verify_email/errors.py
pyguru-dev/django-veify-email
from django import forms
from django.contrib.auth import get_user_model


class RequestNewVerificationEmail(forms.Form):
    email = forms.EmailField(
        label=get_user_model()._meta.get_field("email").verbose_name,  # noqa
        help_text=get_user_model()._meta.get_field("email").help_text,  # noqa
    )

verify_email/forms.py
pyguru-dev/django-veify-email
# Generated by Django 4.0.3 on 2022-04-11 07:00

from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion


class Migration(migrations.Migration):

    initial = True

    dependencies = [
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
    ]

    operations = [
        migrations.CreateModel(
            name='LinkCounter',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('sent_count', models.IntegerField()),
                ('requester', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
            ],
        ),
    ]

verify_email/migrations/0001_initial.py
pyguru-dev/django-veify-email
from django.db import models
from django.contrib.auth import get_user_model

USER = get_user_model()


class LinkCounter(models.Model):
    requester = models.OneToOneField(USER, on_delete=models.CASCADE)
    sent_count = models.IntegerField()

    def __str__(self):
        return str(self.requester.get_username())

    def __repr__(self):
        return str(self.requester.get_username())


verify_email/models.py
pyguru-dev/django-veify-email
import logging

from django.db.models.signals import post_save
from django.dispatch import receiver

from .models import LinkCounter, USER

logger = logging.getLogger(__name__)


@receiver(post_save, sender=USER)
def increase_count(sender, instance, created, **kwargs):
    if created:
        LinkCounter.objects.create(requester=instance, sent_count=1)


@receiver(post_save, sender=USER)
def save_count(sender, instance, **kwargs):
    try:
        instance.linkcounter.save()
    except Exception as err:
        logger.error(err)

verify_email/signals.py
pyguru-dev/django-veify-email
import logging
from datetime import timedelta
from binascii import Error as BASE64ERROR
from base64 import urlsafe_b64encode, urlsafe_b64decode

from django.core import signing
from django.contrib.auth import get_user_model
from django.contrib.auth.tokens import default_token_generator

from .app_configurations import GetFieldFromSettings
from .errors import UserAlreadyActive, MaxRetriesExceeded, UserNotFound, WrongTimeInterval, InvalidToken

__all__ = [
    "TokenManager"
]

logger = logging.getLogger(__name__)


class TokenManager(signing.TimestampSigner):
    """
    This class is responsible for creating encrypted links / verifying them / applying several checks for token lifetime
    and generating new verification links on request.

    ENCRYPTION KEY :
        The link is encrypted using the key variable from django settings.
            - If you want to provide a custom key (other than default secret key of django project), then
              you can set a variable name "HASHING_KEY" in settings.py. (by default it will use your project's secret key
              which if fine in most cases)

    ENCRYPTION SALT :
        The salt value which will be used in algo along with the key to generate hash digest. (by default None)

    SEPARATOR :
        A separator used to separate plain and encrypted text. (by default is ":")
            - If you decide to change this, keep in mind that separator cannot be in
              URL safe base64 alphabet. read : <https://tools.ietf.org/html/rfc4648.html#section-5>

    """

    def __init__(self):
        self.settings = GetFieldFromSettings()
        self.max_age = self.settings.get('max_age', raise_exception=False)
        self.max_reties = self.settings.get('max_retries') + 1
        self._time_units = ['s', 'm', 'h', 'd']

        assert self.max_reties >= 1 if self.max_reties else None

        key = self.settings.get('key', raise_exception=False)
        salt = self.settings.get('salt', raise_exception=False)
        sep = self.settings.get('sep', raise_exception=False)

        super().__init__(key, sep, salt)

    # Private :
    def __get_seconds(self, interval):
        """
        converts the time specified in settings.py "EXPIRE_AFTER" into seconds.
            By default the time will be considered in seconds, to specify days/minutes/hours
            suffix the time with relevant unit.
                - for example:
                    for 1 day, it'll be : "1d"
                    and so on.

            If integer is specified, that will be considered in seconds
        """
        if isinstance(interval, int):
            return interval
        if isinstance(interval, str):
            unit = [i for i in self._time_units if interval.endswith(i)]
            if not unit:
                unit = 's'
                interval += unit
            else:
                unit = unit[0]
            try:
                digit_time = int(interval[:-1])
                if digit_time <= 0:
                    raise WrongTimeInterval('Time must be greater than 0')

                if unit == 's':
                    return digit_time
                if unit == 'm':
                    return timedelta(minutes=digit_time).total_seconds()
                if unit == 'h':
                    return timedelta(hours=digit_time).total_seconds()
                if unit == 'd':
                    return timedelta(days=digit_time).total_seconds()
                else:
                    return WrongTimeInterval(f'Time unit must be from : {self._time_units}')

            except ValueError:
                raise WrongTimeInterval(f'Time unit must be from : {self._time_units}')
        else:
            raise WrongTimeInterval(f'Time unit must be from : {self._time_units}')

    @staticmethod
    def __get_sent_count(user):
        """
        Returns the no. of times email has already been sent to a user.
        """
        try:
            return int(user.linkcounter.sent_count)
        except Exception as e:
            logger.error(e)
            return False

    @staticmethod
    def __increment_sent_counter(user):
        """
        Increment count by one after resending the verification link.
        """
        user.linkcounter.sent_count += 1
        user.linkcounter.save()

    def __generate_token(self, user):
        """
        If "EXPIRE_AFTER" is specified in settings.py, will generate a timestamped signed encrypted token for
        user, otherwise will generate encrypted token without timestamp.
        """
        user_token = default_token_generator.make_token(user)
        if self.max_age is None:
            return self.perform_encoding(user_token)

        signed_token = self.sign(user_token)
        return self.perform_encoding(signed_token)

    def __verify_attempts(self, user):
        """
        Compares the no. of already sent and maximum no. of times that a user is permitted to request new link.
        """
        attempts = self.__get_sent_count(user)
        if attempts and attempts >= self.max_reties:
            return False
        return True

    @staticmethod
    def get_user_by_token(plain_email, encrypted_token):
        """
        returns either a bool or user itself which fits the token and is not active.
        Exceptions Raised
        -----------------
            - UserAlreadyActive
            - InvalidToken
            - UserNotFound
        """
        inactive_users = get_user_model().objects.filter(email=plain_email)
        encrypted_token = encrypted_token.split(':')[0]
        for unique_user in inactive_users:
            valid = default_token_generator.check_token(unique_user, encrypted_token)
            if valid:
                if unique_user.is_active:
                    raise UserAlreadyActive(f'The user with email: {plain_email} is already active')
                return unique_user
            else:
                raise InvalidToken('Token is invalid')
        else:
            raise UserNotFound(f'User with {plain_email} not found')

    def __decrypt_expired_user(self, expired_token):
        """ After the link expires, it decrypt the token without a time stamp to parse out the user token. """
        return self.unsign(expired_token)

    # Public :
    @staticmethod
    def perform_encoding(plain_entity):
        return urlsafe_b64encode(str(plain_entity).encode('UTF-8')).decode('UTF-8')

    @staticmethod
    def perform_decoding(encoded_entity):
        try:
            return urlsafe_b64decode(encoded_entity).decode('UTF-8')
        except BASE64ERROR:
            return False

    def generate_link(self, request, inactive_user, user_email):
        """
        Generates link for the first time.
        """
        token = self.__generate_token(inactive_user)
        encoded_email = urlsafe_b64encode(str(user_email).encode('utf-8')).decode('utf-8')

        link = f"/verification/user/verify-email/{encoded_email}/{token}/"

        absolute_link = request.build_absolute_uri(link)
        return absolute_link

    def request_new_link(self, request, inactive_user, user_email):
        """
        generate link when user clicks on request new link. Perform several checks and returns either a link or bool
        """
        if self.__verify_attempts(inactive_user):  # noqa
            link = self.generate_link(request, inactive_user, user_email)
            self.__increment_sent_counter(inactive_user)  # noqa
            return link
        else:
            raise MaxRetriesExceeded(f'Maximum retries for user with email: {user_email} has been exceeded.')

    def decrypt_link(self, encoded_email, encoded_token):
        """
        main verification and decryption happens here.
        Exceptions Raised
        ------------------
            - signing.SignatureExpired
            - MaxRetriesExceeded
            - signing.BadSignature
            - UserAlreadyActive
            - InvalidToken
        """
        decoded_email = self.perform_decoding(encoded_email)
        decoded_token = self.perform_decoding(encoded_token)

        if decoded_email and decoded_token:
            if self.max_age:
                alive_time = self.__get_seconds(self.max_age)
                try:
                    user_token = self.unsign(decoded_token, alive_time)
                    user = self.get_user_by_token(decoded_email, user_token)
                    if user:
                        return user
                    return False

                except signing.SignatureExpired:
                    logger.warning(f'\n{"~" * 40}\n[WARNING] : The link is Expired!\n{"~" * 40}\n')
                    user = self.get_user_by_token(decoded_email, self.__decrypt_expired_user(decoded_token))
                    if not self.__verify_attempts(user):
                        raise MaxRetriesExceeded()
                    raise

                except signing.BadSignature:
                    logger.critical(
                        f'\n{"~" * 40}\n[CRITICAL] : X_x --> CAUTION : LINK SIGNATURE ALTERED! <-- x_X\n{"~" * 40}\n'
                    )
                    raise
            else:
                user = self.get_user_by_token(decoded_email, decoded_token)
                return user if user else False
        else:
            logger.error(f'\n{"~" * 40}\nError occurred in decoding the link!\n{"~" * 40}\n')
            return False

verify_email/token_manager.py
pyguru-dev/django-veify-email
from django.urls import path
from .views import verify_user_and_activate, request_new_link

urlpatterns = [
    path(f'user/verify-email/<useremail>/<usertoken>/', verify_user_and_activate, name='verify-email'),
    path(f'user/verify-email/request-new-link/<useremail>/<usertoken>/', request_new_link, name='request-new-link-from-token'),
    path(f'user/verify-email/request-new-link/', request_new_link, name='request-new-link-from-email'),
]

verify_email/urls.py
pyguru-dev/django-veify-email
import logging

from django.http import Http404, HttpResponse
from django.urls import reverse
from django.shortcuts import render, redirect
from django.contrib.auth import get_user_model
from django.contrib import messages
from django.core.exceptions import ObjectDoesNotExist, MultipleObjectsReturned
from django.core.signing import SignatureExpired, BadSignature

from .app_configurations import GetFieldFromSettings
from .confirm import verify_user
from .email_handler import resend_verification_email
from .forms import RequestNewVerificationEmail
from .errors import (
    InvalidToken,
    MaxRetriesExceeded,
    UserAlreadyActive,
    UserNotFound,
)

logger = logging.getLogger(__name__)

pkg_configs = GetFieldFromSettings()

login_page = pkg_configs.get('login_page')

success_msg = pkg_configs.get('verification_success_msg')
failed_msg = pkg_configs.get('verification_failed_msg')

failed_template = pkg_configs.get('verification_failed_template')
success_template = pkg_configs.get('verification_success_template')
link_expired_template = pkg_configs.get('link_expired_template')
request_new_email_template = pkg_configs.get('request_new_email_template')
new_email_sent_template = pkg_configs.get('new_email_sent_template')


def verify_user_and_activate(request, useremail, usertoken):
    """
    A view function already implemented for you so you don't have to implement any function for verification
    as this function will be automatically be called when user clicks on verification link.

    verify the user's email and token and redirect'em accordingly.
    """
    if request.method == 'GET':
        try:
            verified = verify_user(useremail, usertoken)
            if verified is True:
                if login_page and not success_template:
                    messages.success(request, success_msg)
                    return redirect(to=login_page)
                return render(
                    request,
                    template_name=success_template,
                    context={
                        'msg': success_msg,
                        'status': 'Verification Successful!',
                        'link': reverse(login_page)
                    }
                )
            else:
                # we dont know what went wrong...
                raise ValueError
        except (ValueError, TypeError) as error:
            logger.error(f'[ERROR]: Something went wrong while verifying user, exception: {error}')
            return render(
                request,
                template_name=failed_template,
                context={
                    'msg': failed_msg,
                    'minor_msg': 'There is something wrong with this link...',
                    'status': 'Verification Failed!',
                }
            )
        except SignatureExpired:
            return render(
                request,
                template_name=link_expired_template,
                context={
                    'msg': 'The link has lived its life :( Request a new one!',
                    'status': 'Expired!',
                    'encoded_email': useremail,
                    'encoded_token': usertoken
                }
            )
        except BadSignature:
            return render(
                request,
                template_name=failed_template,
                context={
                    'msg': 'This link was modified before verification.',
                    'minor_msg': 'Cannot request another verification link with faulty link.',
                    'status': 'Faulty Link Detected!',
                }
            )
        except MaxRetriesExceeded:
            return render(
                request,
                template_name=failed_template,
                context={
                    'msg': 'You have exceeded the maximum verification requests! Contact admin.',
                    'status': 'Maxed out!',
                }
            )
        except InvalidToken:
            return render(
                request,
                template_name=failed_template,
                context={
                    'msg': 'This link is invalid or been used already, we cannot verify using this link.',
                    'status': 'Invalid Link',
                }
            )
        except UserNotFound:
            raise Http404("404 User not found")


def request_new_link(request, useremail=None, usertoken=None):
    try:
        if useremail is None or usertoken is None:
            # request came from re-request email page
            if request.method == 'POST':
                form = RequestNewVerificationEmail(request.POST)  # do not inflate data
                if form.is_valid():
                    form_data: dict = form.cleaned_data
                    email = form_data['email']

                    inactive_user = get_user_model().objects.get(email=email)
                    if inactive_user.is_active:
                        raise UserAlreadyActive('User is already active')
                    else:
                        # resend email
                        status = resend_verification_email(request, email, user=inactive_user, encoded=False)
                        if status:
                            return render(
                                request,
                                template_name=new_email_sent_template,
                                context={
                                    'msg': "You have requested another verification email!",
                                    'minor_msg': 'Your verification link has been sent',
                                    'status': 'Email Sent!',
                                }
                            )
                        else:
                            logger.error('something went wrong during sending email')
            else:
                form = RequestNewVerificationEmail()
            return render(
                request,
                template_name=request_new_email_template,
                context={'form': form}
            )
        else:
            # request came from  previously sent link
            status = resend_verification_email(request, useremail, token=usertoken)

        if status:
            return render(
                request,
                template_name=new_email_sent_template,
                context={
                    'msg': "You have requested another verification email!",
                    'minor_msg': 'Your verification link has been sent',
                    'status': 'Email Sent!',
                }
            )
        else:
            messages.info(request, 'Something went wrong during sending email :(')
            logger.error('something went wrong during sending email')

    except ObjectDoesNotExist as error:
        messages.warning(request, 'User not found associated with given email!')
        logger.error(f'[ERROR]: User not found. exception: {error}')
        return HttpResponse(b"User Not Found", status=404)

    except MultipleObjectsReturned as error:
        logger.error(f'[ERROR]: Multiple users found. exception: {error}')
        return HttpResponse(b"Internal server error!", status=500)

    except KeyError as error:
        logger.error(f'[ERROR]: Key error for email in your form: {error}')
        return HttpResponse(b"Internal server error!", status=500)

    except MaxRetriesExceeded as error:
        logger.error(f'[ERROR]: Maximum retries for link has been reached. exception: {error}')
        return render(
            request,
            template_name=failed_template,
            context={
                'msg': 'You have exceeded the maximum verification requests! Contact admin.',
                'status': 'Maxed out!',
            }
        )
    except InvalidToken:
        return render(
            request,
            template_name=failed_template,
            context={
                'msg': 'This link is invalid or been used already, we cannot verify using this link.',
                'status': 'Invalid Link',
            }
        )
    except UserAlreadyActive:
        return render(
            request,
            template_name=failed_template,
            context={
                'msg': "This user's account is already active",
                'status': 'Already Verified!',
            }
        )

verify_email/views.py
pyguru-dev/django-veify-email