Email Set Up and Confirmation

Hello, welcome back to the 6th post of my Flask series, this post would be a continuation of my previous post, please click on any link below to view any post you missed

OUTLINE OF FLASK SERIES

  1. Structuring a Flask-Restful API for Production
  2. Creating Custom Error Pages and Handling Exceptions for Flask
  3. Flask authentication with JWT
  4. CRUD Operations with Flask
  5. Using Marshmallow to Simplify Parameter Validation in APIs.
  6. Email Set up and Confirmation( this article)

 

At the end of this post, you will learn the following

  • Send out emails using the Sengrid Simple Mail Transfer Protocol (SMTP) server.
  • We would create a new user workflow before a user can log in to the application, we confirm the mail is a valid one, send an activation link to the email.
  • Allow user to change their password if they forget their login credentials.

We would install new dependencies for our new post

  • Flask-Mail:  it provides a simple interface to set up SMTP with your Flask application and to send messages from your views and scripts.
  • isdangerous: the itsdangerous package to create and validate tokens sent to an email address.

 

 

FIRST STEP: ADD EMAIL CONFIRMATION

We would modify our user model, add a field to confirm if the user email is valid and the user has clicked the confirmation link and date the confirmation happened.

#-----existing code base

class User(BaseModel):
    __tablename__ = 'user'

    username = db.Column(db.String(80), nullable=False, unique=True)
    email = db.Column(db.String(200), nullable=False, unique=True)
    password = db.Column(db.String(200),nullable=False)
    note = db.relationship('Note', backref='user')
    #new field
    confirmed = db.Column(db.Boolean(), default=False)
    confirmed_on = db.Column(db.DateTime, nullable=True)

#------existing code base


flask db migrate -m "confirmed feature added"
flask db upgrade


Update your config.py and add the following mail set, also update your .env file and add the following.

class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_ERROR_MESSAGE_KEY = 'message'
    JWT_BLACKLIST_ENABLED = True   
    JWT_BLACKLIST_TOKEN_CHECKS = ['access', 'refresh']
    JWT_ACCESS_TOKEN_EXPIRES = int(os.getenv("JWT_ACCESS_TOKEN_EXPIRES", 
                                                     14400))
    #mail setup
    MAIL_SERVER = os.environ.get('MAIL_SERVER')    
    MAIL_PORT = int(os.environ.get('MAIL_PORT')) 
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS')
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')   
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER')
    SECURITY_PASSWORD_SALT = os.environ.get('SECURITY_PASSWORD_SALT')

Update your environment variables.

DEV_DATABASE_URL='******'
FLASK_APP=main.py 
FLASK_DEBUG=1 
FLASK_ENV=development
SECRET_KEY='GGFGFJ77T54ERRErreeeeeeeeeeee88jbbb&'
MAIL_SERVER = 'smtp.sendgrid.net'  
MAIL_PORT = 587
MAIL_USE_TLS = True
MAIL_USERNAME = 'apikey'   
MAIL_PASSWORD = ''
SENDGRID_API_KEY = ''
SECURITY_PASSWORD_SALT='gfghtt6884@@%68848@$$@yygb'
MAIL_DEFAULT_SENDER='Topnotch@diary.com'

Please follow the instruction here on how to set up a free SendGrid account to get your API keys, which you will fill in the MAIL_PASSWORD and SENDGRID_API_KEY variables.

Update your app.py file and the following

#----existing code

from flask_mail import Mail

#------------existing code

mail = Mail()


def create_app(config_name):

    import resources

    app = Flask(__name__)

    app.config.from_object(env_config[config_name])

    db.init_app(app)  # Add db session, new code here

    Migrate(app, db) 
    # register api
    api.init_app(app)
    mail.init_app(app) #new code here
Generating Account Confirmation Tokens

Create a new file in your utils folder, name it email_token.py, and add the code below.

from flask import current_app
from itsdangerous import URLSafeTimedSerializer



def generate_confirmation_token(email):
    serializer = URLSafeTimedSerializer(current_app.config['SECRET_KEY'])
    return serializer.dumps(email, 
   salt=current_app.config['SECURITY_PASSWORD_SALT'])


def confirm_token(token, expiration=3600):
    serializer = URLSafeTimedSerializer(current_app.config['SECRET_KEY'])
    try:
        email = serializer.loads(
            token,
            salt=current_app.config['SECURITY_PASSWORD_SALT'],
            max_age=expiration
        )
    except:
        return False
    return email

 

So, in the generate_confirmation_token() function, we use the URLSafeTimedSerializer to generate a token using the email address obtained during user registration. The actual email is encoded in the token. Then to confirm the token, within the confirm_token() function, we can use the loads() method, which takes the token and expiration – valid for one hour (3,600 seconds) – as arguments. As long as the token has not expired, then it will return an email.

Create a basic function for sending emails, create a new file send_emails.py in the util folder, and add this code.

from flask_mail import Mail, Message




mail = Mail()


def send_email(to_email, subject, body):
  msg = Message(subject, recipients=[to_email])
  msg.html = body
  mail.send(msg)

 

Next step, update our user resource

import datetime
import re
from http import HTTPStatus

from flask import current_app, render_template, request, url_for
from flask_jwt_extended import (
    create_access_token,
    create_refresh_token,
    get_jwt_identity,
    get_raw_jwt,
    jwt_optional,
    jwt_refresh_token_required,
    jwt_required,
)
from flask_restful import Api, Resource
from marshmallow import ValidationError
from webargs import validate
from webargs.fields import Email, Str
from webargs.flaskparser import use_args, use_kwargs
from werkzeug.security import check_password_hash, generate_password_hash

from api.models import User
from api.schemas import UserSchema
from utils.email_token import confirm_token, generate_confirmation_token
from utils.send_emails import send_email
from flask_mail import Mail, Message


api = Api()

user_schema = UserSchema()

black_list = set()

mail = Mail()


PASSWORD_VALIDATION = validate.Regexp(
    "^(?=.*[0-9]+.*)(?=.*[a-zA-Z]+.*).{7,16}$",
    error="Password must contain at least one letter, at"
    " least one number, be longer than six charaters "
    "and shorter than 16.",
)


class UserRegistrationResource(Resource):
    """Define endpoints for user registration."""

    def post(self):
        """Create new  user."""
        json_input = request.get_json()

        try:
            data = user_schema.load(json_input)
        except ValidationError as err:
            return {"errors": err.messages}, 422

        # Check if use and email exist before creation

        if User.get_by_username(data["username"]):

            return {"message": "username already exist"}, 
             HTTPStatus.BAD_REQUEST

        if User.get_by_email(data["email"]):
            return {"message": "email already exist"}, HTTPStatus.BAD_REQUEST

        user = User(**data)
        user.save()

        token = generate_confirmation_token(user.email)

        # mail requirements
        subject = "Please confirm your email to be able to use our app."
        # Reverse routing
        link = url_for("useractivateresource", token=token, _external=True)

        body = f"Hi, Thanks for using our app! Please confirm your registration 
        by clicking on the link: {link} . Welcome to our family"

        send_email(user.email, subject, body)

        data = user_schema.dump(user)
        data["message"] = "Successfully created a new user"
        return data, HTTPStatus.CREATED



class UserActivateResource(Resource):
    def get(self, token):

        email = confirm_token(token)

        if email is False:
            return {"message": "Invalid token or token expired"}, 
          HTTPStatus.BAD_REQUEST

        user = User.get_by_email(email=email)

        if not user:
            return {"message": "User not found"}, HTTPStatus.NOT_FOUND

        if user.confirmed is True:
            return (
                {"message": "The user account is already activated"},
                HTTPStatus.BAD_REQUEST,
            )

        user.confirmed = True
        user.confirmed_on = datetime.datetime.now()

        user.save()

        return {}, HTTPStatus.NO_CONTENT



 

Note that a user.save() had to be added before the confirmation email is sent out. The problem is that new users get assigned an id when they are committed to the database, and this id is needed to generate the confirmation token.

We create the activation link using the url_for function. It will require UserActivateResource. This endpoint will need a token as well.

The _external=True argument is added to the url_for() call to request a fully qualified URL that includes the scheme (http:// or https://), hostname, and port.

The _external=True parameter is used to convert the default relative URL,       /users/activate/<string:token>, to an absolute URL, http://localhost:5000/users/activate/<string:token>

Update your app.py, register your UserActivateResource

#---------EXISTING CODEBASE

from resources.user import (RefreshAccessTokenResource,
                            RevokeAccessTokenResource, UserActivateResource,
                            UserInfoResource, UserLoginResource,
                            UserRegistrationResource, black_list)

#---------EXISTING CODEBASE
api.add_resource(UserActivateResource, '/users/activate/<string:token>')

Finally, we would like to make sure the user cannot log in to the application before their account is activated. Update the UserLoginResource handling the user login resource.

class UserLoginResource(Resource):
    """Define endpoints for user login."""

    user_login = {
        "email":Email(required=True,location="json"),
        "password":Str(required=True, location="json")
        
    }

    @use_kwargs(user_login)
    def post(self, email, password):
        """Create new  user."""
        
        user = User.get_by_email(email)

        if not user or not check_password_hash(user.password, password):
            return {'message': 'username or password is incorrect'}, 
         HTTPStatus.UNAUTHORIZED

        if user.confirmed is False:
            return {'message': 'The user account is not activated yet'}, 
            HTTPStatus.FORBIDDEN

        access_token = create_access_token(identity=user.id, fresh=True)
        refresh_token = create_refresh_token(identity=user.id)

        return {'access_token': access_token, 'refresh_token': refresh_token}, 
        HTTPStatus.OK

 

second step: password reset feature

#======================existing code

class ForgotPasswordResource(Resource):
    """Define endpoints for resetting user password."""

    user_reset = { "email": Email(required=True, location="json")}

    @use_kwargs(user_reset)
    def post(self, email):
        user = User.get_by_email(email)

        if not user:
            return {"message": "email is invalid"}, HTTPStatus.UNAUTHORIZED

        subject = "Password reset requested"

        # Here we use the URLSafeTimedSerializer
        token = generate_confirmation_token(user.email)

        recover_url = url_for("resetpasswordresource", token=token, 
              _external=True)

        text = f"Hi {user.username}, Thanks for using our app! Please reset 
        your password by clicking on the link: {recover_url} .\
        If you didn't ask for a password reset, ignore the mail."

        send_email(to_email=user.email, subject=subject, body=text)

        return { "msg": "succesfully sent the reset mail to your email"} ,   
        HTTPStatus.OK

    


class ResetPasswordResource(Resource):
    @use_kwargs(
        {"password": Str(location="json", required=True, 
        validate=PASSWORD_VALIDATION)}
    )
    def patch(self, token, password):

        email = confirm_token(token)

        if email is False:
            return {"message": "Invalid token or token expired"}, 
        HTTPStatus.BAD_REQUEST

        user = User.get_by_email(email=email)

        if not user:
            return {"message": "User not found"}, HTTPStatus.NOT_FOUND

        if user.confirmed is True:

            user.password = generate_password_hash(password)

            user.save()

            return (
                {
                    "status": "success",
                    "data": {"msg": "New password was successfully set"},
                },
                HTTPStatus.OK,
            )
        else:
            return {"message": "An error occured"},   
    HTTPStatus.BAD_REQUEST

Update your app.py and register the following resources.

from resources.user import (RefreshAccessTokenResource, ResetPasswordResource,
                            RevokeAccessTokenResource, UserActivateResource,
                            UserInfoResource, UserLoginResource, ForgotPasswordResource,
                            UserRegistrationResource, black_list)


api.add_resource(ResetPasswordResource, "/users/reset_password/<string:token>")
api.add_resource(ForgotPasswordResource, "/users/forgot_password/")

 

We have come to the end of our tutorial.

The link to Github tutorial is here

Until then happy coding.

 

 

 

Leave a Reply

Your email address will not be published. Required fields are marked *