JWT Implimentation

JWT

What is a JWT (JSON Web Token)?

JSON Web Tokens (JWT) serve as a secure means of transmitting information in JSON format, providing a way to establish trust between different parties in a web application.

Parts of a JWT:

Parts

Each section is divided by a dot

{
  # For example:
  "alg": "HS256",
  "typ": "JWT"
}
{
  # For Example:
  "sub": "1234567890",
  "name": "John Doe",
  "admin": true
}
# Using the HMAC SHA256 algorithm the signature will be created as such:
HMACSHA256(
  base64UrlEncode(header) + "." +
  base64UrlEncode(payload),
  secret)

HMAC and RSA or ECDSA Keys:

Used for signing JWT tokens

Rest APIs and JSON Web Tokens

Rest APIs:

JWT

Principals of REST-

REST APIs and JWT:

A JSON Web Token (JWT) is often used in the context of RESTful APIs to secure and authenticate communication between clients and servers.

Implimenting JWT in Flask Using PyJWT

1) Add PyJWT to requirements.txt

PyJWT will be used to encode and decode JWT tokens Flask_SqlAlchemy Flask_Migrate Flask_Restful Flask_Cors PyJWT

2) Add a Secret Key as App Config

Modify the app.config [‘SECRET_KEY’] to try and get secret key for encoding and decoding the JWT from the variable SECRET_KEY. It defaults to “SECRET_KEY” for the key.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
import os

"""
These objects can be used throughout project.
"""
@@ -14,7 +15,8 @@
dbURI = 'sqlite:///volumes/sqlite.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = dbURI
SECRET_KEY = os.environ.get('SECRET_KEY') or 'SECRET_KEY'
app.config['SECRET_KEY'] = SECRET_KEY
db = SQLAlchemy()
Migrate(app, db)

3) Migration Script Hashes Passwords and New Valid Hash Method

We need to prevent storing passwords in plaintext (for security reasons) so we need to hash passwords when initializing users in the database (in migration sript). All the user set_password methods are set to the same hash method.

Migration Script Hashes Passwords
from alembic import op
import sqlalchemy as sa
from datetime import date
from werkzeug.security import generate_password_hash


# revision identifiers, used by Alembic.
revision = '5ac11951f352'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    players_table = op.create_table('players',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('_name', sa.String(length=255), nullable=False),
    sa.Column('_uid', sa.String(length=255), nullable=False),
    sa.Column('_password', sa.String(length=255), nullable=False),
    sa.Column('_tokens', sa.Integer(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('_uid')
    )
    users_table = op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('_name', sa.String(length=255), nullable=False),
    sa.Column('_uid', sa.String(length=255), nullable=False),
    sa.Column('_password', sa.String(length=255), nullable=False),
    sa.Column('_dob', sa.Date(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('_uid')
    )
    posts_table = op.create_table('posts',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('note', sa.Text(), nullable=False),
    sa.Column('image', sa.String(), nullable=True),
    sa.Column('userID', sa.Integer(), nullable=True),
    sa.ForeignKeyConstraint(['userID'], ['users.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    op.bulk_insert(players_table,
    [
        {'id': 1,
        '_name': "Azeem Khan",
        '_uid': "azeemK",
        '_password': generate_password_hash("123qwerty", "pbkdf2:sha256", salt_length=10),
        "_tokens": 45},
        {'id': 2,
        '_name': "Ahad Biabani",
        '_uid': "ahadB",
        '_password': generate_password_hash("123qwerty", "pbkdf2:sha256", salt_length=10),
        "_tokens": 41},
        {'id': 3,
        '_name': "Akshat Parikh",
        '_uid': "akshatP",
        '_password': generate_password_hash("123qwerty", "pbkdf2:sha256", salt_length=10),
        "_tokens": 40},
        {'id': 4,
        '_name': "Josh Williams",
        '_uid': "joshW",
        '_password': generate_password_hash("123qwerty", "pbkdf2:sha256", salt_length=10),
        "_tokens": 38},
    ]
    )
    op.bulk_insert(users_table,
    [
        {'id': 1,
        '_name': "Thomas Edison",
        '_uid': "toby",
        '_password': generate_password_hash("123toby", "pbkdf2:sha256", salt_length=10),
        "_dob": date(1847, 2, 11)},
        {'id': 2,
        '_name': "Nicholas Tesla",
        '_uid': "niko",
        '_password': generate_password_hash("123niko", "pbkdf2:sha256", salt_length=10),
        "_dob": date(1856, 7, 10)},
        {'id': 3,
        '_name': "Alexander Graham Bell",
        '_uid': "lex",
        '_password': generate_password_hash("123qwerty", "pbkdf2:sha256", salt_length=10),
        "_dob": date.today()},
        {'id': 4,
        '_name': "Grace Hopper",
        '_uid': "hop",
        '_password': generate_password_hash("123hop", "pbkdf2:sha256", salt_length=10),
        "_dob": date(1906, 12, 9)},
    ]
    )
    op.bulk_insert(posts_table,
    [
        {'id': 1,
        'note': "#### Thomas Edison Test Note ####",
        'image': "ncs_logo.png",
        'userID': 1
        },
        {'id': 2,
        'note': "#### Nicholas Tesla Test Note ####",
        'image': "ncs_logo.png",
        'userID': 2
        },
        {'id': 3,
        'note': "#### Alexander Graham Bell Test Note ####",
        'image': "ncs_logo.png",
        'userID': 3
        },
        {'id': 4,
        'note': "#### Grace Hopper Test Note ####",
        'image': "ncs_logo.png",
        'userID': 4},
    ]
    )
    # ### end Alembic commands ###
def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('posts')
    op.drop_table('users')
    op.drop_table('players')
    # ### end Alembic commands ###
New Valid Hash Method
    # update password, this is conventional setter
    def set_password(self, password):
        """Create a hashed password."""
        self._password = generate_password_hash(password, "pbkdf2:sha256", salt_length=10)

    # check password parameter versus stored/encrypted password
    def is_password(self, password):

4) Checking JWT Token and Authentication

This authentication middleware functions as a checkpoint for all incoming requests to the web server before reaching the API. Its primary task is to check for the presence of a JWT cookie in the incoming request. In cases where the cookie is absent, the middleware responds with a 401 Unauthorized Error, indicating that the required token is missing. If a cookie is found, the middleware proceeds to decode the token, extracting the user ID (uid). Following this, it validates the existence of a user associated with that ID.

from functools import wraps
import jwt
from flask import request, abort
from flask import current_app
from model.users import User

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.cookies.get("jwt")
        print(token)
        if not token:
            return {
                "message": "Authentication Token is missing!",
                "data": None,
                "error": "Unauthorized"
            }, 401
        try:
            data=jwt.decode(token, current_app.config["SECRET_KEY"], algorithms=["HS256"])
            print(data["_uid"])
            current_user=User.query.filter_by(_uid=data["_uid"]).first()
            if current_user is None:
                return {
                "message": "Invalid Authentication token!",
                "data": None,
                "error": "Unauthorized"
            }, 401
        except Exception as e:
            return {
                "message": "Something went wrong",
                "data": None,
                "error": str(e)
            }, 500

        return f(current_user, *args, **kwargs)

    return decorated

5) Authenticate Returns from Generaated JWT Tokens

The endpoint /api/users/authenticate now provides a JWT token based on the request’s JSON data, which includes the uid and password. Several checks are performed, ensuring the presence of the request body, existence of the uid, presence of a user with the specified uid, and correctness of the provided password. If all checks pass, the uid is encoded into a JWT token, which is then returned as a cookie.

Additionally, the @token_required decorator is applied to the get and post methods so that only authenticated users can reach and access resources.

import json, jwt
from flask import Blueprint, request, jsonify, current_app, Response
from flask_restful import Api, Resource # used for REST API building
from datetime import datetime
from auth_middleware import token_required

from model.users import User

user_api = Blueprint('user_api', __name__,
                   url_prefix='/api/users')
# API docs https://flask-restful.readthedocs.io/en/latest/api.html
api = Api(user_api)

class UserAPI:        
    class _CRUD(Resource):  # User API operation for Create, Read.  THe Update, Delete methods need to be implemeented
        @token_required
        def post(self, current_user): # Create method
            ''' Read data for json body '''
            body = request.get_json()

            ''' Avoid garbage in, error checking '''
            # validate name
            name = body.get('name')
            if name is None or len(name) < 2:
                return {'message': f'Name is missing, or is less than 2 characters'}, 400
            # validate uid
            uid = body.get('uid')
            if uid is None or len(uid) < 2:
                return {'message': f'User ID is missing, or is less than 2 characters'}, 400
            # look for password and dob
            password = body.get('password')
            dob = body.get('dob')
            ''' #1: Key code block, setup USER OBJECT '''
            uo = User(name=name, 
                      uid=uid)
            
            ''' Additional garbage error checking '''
            # set password if provided
            if password is not None:
                uo.set_password(password)
            # convert to date type
            if dob is not None:
                try:
                    uo.dob = datetime.strptime(dob, '%Y-%m-%d').date()
                except:
                    return {'message': f'Date of birth format error {dob}, must be mm-dd-yyyy'}, 400
            
            ''' #2: Key Code block to add user to database '''
            # create user in database
            user = uo.create()
            # success returns json of user
            if user:
                return jsonify(user.read())
            # failure returns error
            return {'message': f'Processed {name}, either a format error or User ID {uid} is duplicate'}, 400
        
        @token_required
        def get(self, current_user): # Read Method
            users = User.query.all()    # read/extract all users from database
            json_ready = [user.read() for user in users]  # prepare output in json
            return jsonify(json_ready)  # jsonify creates Flask response object, more specific to APIs than json.dumps

    class _Security(Resource):
        def post(self):
            try:
                body = request.get_json()
                if not body:
                    return {
                        "message": "Please provide user details",
                        "data": None,
                        "error": "Bad request"
                    }, 400
                ''' Get Data '''
                uid = body.get('uid')
                if uid is None:
                    return {'message': f'User ID is missing'}, 400
                password = body.get('password')

                ''' Find user '''
                user = User.query.filter_by(_uid=uid).first()
                print(user._uid,user._password)
                if user is None or not user.is_password(password):
                    return {'message': f"Invalid user id or password"}, 400
                if user:
                    try:
                        token = jwt.encode(
                            {"_uid": user._uid},
                            current_app.config["SECRET_KEY"],
                            algorithm="HS256"
                        )
                        resp = Response("Authentication for %s successful" % (user._uid))
                        resp.set_cookie("jwt", token,
                                max_age=3600,
                                secure=True,
                                httponly=True,
                                path='/'
                                # domain="frontend.com"
                                )
                        return resp
                    except Exception as e:
                        return {
                            "error": "Something went wrong",
                            "message": str(e)
                        }, 500
                return {
                    "message": "Error fetching auth token!, invalid email or password",
                    "data": None,
                    "error": "Unauthorized"
                },
            except Exception as e:
                return {
                        "message": "Something went wrong!",
                        "error": str(e),
                        "data": None
                }, 500



    # building RESTapi endpoint
    api.add_resource(_CRUD, '/')
    api.add_resource(_Security, '/authenticate')

Implimenting JWT in Flask Using Flask-JWT-Extended

1) Install Flask-JWT-Extended

pip install Flask-JWT-Extended

2) Include JWT Library in Flask App Setup

from flask_jwt_extended import JWTManager

# You don't have to add this again if you already have it.
app = Flask(__name__)

# Setup the Flask-JWT-Extended extension
app.config["JWT_SECRET_KEY"] = "SECRET_KEY"  # Remember to change "secret" to a more complex key
jwt = JWTManager(app)

3) Create an Endpoint for Generating Tokens

Using POST because it is creating tokens (POST is for creation)

POST /token
Content-type: application/json
Body:
{
     "username": "alesanchezr",
     "password": "12341234"
}
from flask_jwt_extended import create_access_token

# Create a route to authenticate your users and return JWT Token
# The create_access_token() function is used to actually generate the JWT
@app.route("/token", methods=["POST"])
def create_token():
    username = request.json.get("username", None)
    password = request.json.get("password", None)

    # Query your database for username and password
    user = User.query.filter_by(username=username, password=password).first()

    if user is None:
        # The user was not found on the database
        return jsonify({"msg": "Bad username or password"}), 401
    
    # Create a new token with the user id inside
    access_token = create_access_token(identity=user.id)
    return jsonify({ "token": access_token, "user_id": user.id })

4) Use @jwt_required() Decorator on Private Routes

Endpoints that require authorization (private endpoints) should use the @jwt_required() decorator. You can retrieve valid, authenticated user information using the get_jwt_identity function.

from flask_jwt_extended import jwt_required, get_jwt_identity

# Protect a route with jwt_required, which will kick out requests without a valid JWT
@app.route("/protected", methods=["GET"])
@jwt_required()
def protected():
    # Access the identity of the current user with get_jwt_identity
    current_user_id = get_jwt_identity()
    user = User.query.get(current_user_id)
    
    return jsonify({"id": user.id, "username": user.username }), 200

Implimenting JWT in Front-End

1) Create a New Token

Based on earlier endpoints, we have to POST /token with username + password information in request body.

const login = async (username, password) => {
     const resp = await fetch(`https://your_api.com/token`, { 
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ username, password }) 
     })

     if(!resp.ok) throw Error("There was a problem in the login request")

     if(resp.status === 401){
          throw("Invalid credentials")
     }
     else if(resp.status === 400){
          throw ("Invalid email or password format")
     }
     const data = await resp.json()
     // Save your token in the localStorage
     // Also you should set your user into the store using the setItem function
     localStorage.setItem("jwt-token", data.token);

     return data
}

2) Retrieving Information

Now, lets say I’ve logged in to the front-end application and I want to retrieve sensitive data:

// Assuming "/protected" is a private endpoint
const getMyTasks = async () => {
     // Retrieve token from localStorage
     const token = localStorage.getItem('jwt-token');

     const resp = await fetch(`https://your_api.com/protected`, {
        method: 'GET',
        headers: { 
          "Content-Type": "application/json",
          'Authorization': 'Bearer ' + token // ⬅⬅⬅ authorization token
        } 
     });

     if(!resp.ok) {
          throw Error("There was a problem in the login request")
     } else if(resp.status === 403) {
          throw Error("Missing or invalid token");
     } else {
         throw Error("Unknown error");
     }

     const data = await resp.json();
     console.log("This is the data you requested", data);
     return data
}

CORS Configuration:

Cross-Origin Resource Sharing (CORS) is a crucial aspect of web development, allowing or restricting requests from different origins. This configuration is essential to ensure that your Flask application can seamlessly handle requests from various sources, promoting security and compatibility.

Change: Added CORS support for deployed and local runs.

cors = CORS(app, supports_credentials=True)

Dynamic Allowed Origins:

The concept of “Dynamic Allowed Origins” refers to the ability to dynamically set the allowed origins for Cross-Origin Resource Sharing (CORS) based on the incoming request’s ‘Origin’ header. In CORS, the ‘Origin’ header indicates the origin (domain) of the requesting client.

Change: A before_request hook is implemented to dynamically set the allowed origin based on the incoming request’s ‘Origin’ header.

@app.before_request
def before_request():
    allowed_origin = request.headers.get('Origin')
    if allowed_origin in ['http://localhost:4100', 'http://127.0.0.1:4100', 'https://nighthawkcoders.github.io']:
        cors._origins = allowed_origin

The SameSite attribute in a cookie is a security feature that controls whether a cookie should be sent with cross-site requests. It is set within the Set-Cookie header when a server instructs a browser to store a cookie on the user’s device. The SameSite attribute helps mitigate certain types of cross-site request forgery (CSRF) attacks.

Change: The SameSite attribute is set to ‘None’ in the set_cookie method to allow cross-site requests.

resp.set_cookie(
    "jwt",
    token,
    max_age=3600,
    secure=True,
    httponly=True,
    path='/',
    samesite='None'
)

Nginx Preflighted Requests Configuration:

The Nginx preflighted requests configuration is designed to handle preflighted requests. Preflighted requests are HTTP OPTIONS requests sent by the browser before the actual request to check whether the server will accept the actual request.

Change: Nginx configuration is added to handle preflighted requests, ensuring that only the specified frontend server is allowed.

if ($request_method = OPTIONS) {
    add_header "Access-Control-Allow-Credentials" "true" always;
    add_header "Access-Control-Allow-Origin"  "https://nighthawkcoders.github.io" always;
    add_header "Access-Control-Allow-Methods" "GET, POST, PUT, OPTIONS, HEAD" always;
    add_header "Access-Control-Allow-MaxAge" 600 always;
    add_header "Access-Control-Allow-Headers" "Authorization, Origin, X-Requested-With, Content-Type, Accept" always;
    return 204;
}

Cookies

Cookie

Cookies are small pieces of data stored on the user’s browser by website

Cookies and JWT

Cookies can be used to store JWTs.

The process of maintaining tokens in cookies and transmitting them in HTTP requests involves several steps (we’ve gone over this earlier, but lets go through it again in the context of cookies):

  1. Token Generation and Storage:
  2. Cookie Storage and Retrieval:
  3. Transmission in HTTP Request:
  4. Token Verification:

Postman

Postman

Postman serves as a valuable tool in the realm of computer science, specifically designed for testing and interacting with Application Programming Interfaces (APIs). Postman can be likened to a sophisticated testing environment that facilitates the exploration of API functionalities.

Within Postman, users can craft and dispatch HTTP requests to APIs, allowing for a comprehensive examination of how different software components communicate and respond. This tool is instrumental in ensuring the seamless integration of APIs into software applications. Its intuitive interface provides a user-friendly platform to engage in the essential practice of testing and validating API behavior.

Postman and JWT:

Postman allows you to include JWTs in your requests easily. Here’s different ways of how you can use Postman to send a request with a JWT:

  1. Authorization Tab
  2. Headers:

This way, Postman enables you to simulate API requests with JWTs, ensuring that your authentication mechanisms are working as expected. If everything is working as expected, you will recieve a status code in the 2xx range (e.g., 200 OK) which typically indicates success. If you do not recieve a 200 OK message, Postman will let you know what is wrong so you can catch errors in your code.

Postman and Cookies:

Similar to JWT, Postman allows you to manage cookies as efficently. Here’s different ways you can work with cookies in Postman:

  1. Cookies Tab:
  2. Headers:

Anatomy of Python Flask Repo

(Teacher Flask Repo)

Part1 Part1