How JWT (JSON Web Token) authentication works?

https://www.idiotinside.com/2019/09/26/how-jwt-authentication-works/

What is JWT (JSON Web Token)?

JSON Web Token (JWT) is an open standard (RFC 7519) for securely transmitting information between parties as JSON object.

It is compact, readable and digitally signed using a private key/ or a public key pair by the Identity Provider(IdP). So the integrity and authenticity of the token can be verified by other parties involved.

The purpose of using JWT is not to hide data but to ensure the authenticity of the data. JWT is signed and encoded, not encrypted. 

JWT is a token based stateless authentication mechanism. Since it is a client-side based stateless session, server doesn’t have to completely rely on a datastore(database) to save session information.

How it works?

Basically the identity provider(IdP) generates a JWT certifying user identity and Resource server decodes and verifies the authenticity of the token using secret salt / public key.

1. User sign-in using username and password or google/facebook.
2. Authentication server verifies the credentials and issues a jwt signed using either a secret salt or a private key.
3. User’s Client uses the JWT to access protected resources by passing the JWT in HTTP Authorization header.
4. Resource server then verifies the authenticity of the token using the secret salt/ public key.

 

reference

https://jwt.io/introduction/

DEMO

https://github.com/amirhosss/FastAPI-RS256-MongoDB-Redis

私钥签发 token

https://github.com/amirhosss/FastAPI-RS256-MongoDB-Redis/blob/d3f484b542bb96be5d74b499a9f3ab586b77d4d1/core/security.py#L24

import uuid
from datetime import datetime, timedelta

from jose import jwt
from passlib.hash import bcrypt_sha256

from .config import settings


def get_password(password):
    return bcrypt_sha256.hash(password)


def verify_password(plain_password, hashed_password):
    return bcrypt_sha256.verify(plain_password, hashed_password)


with open('core/private_key.pem', 'rb') as private_file:
    private_key = private_file.read()
with open('core/public_key.pem', 'rb') as public_file:
    public_key = public_file.read()
    

def create_jwt_token(sub: str, aud: str, expires_delta: timedelta = None):
    expires_delta_condition = {
        'refresh': timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS),
        'access': timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
        'verification': timedelta(minutes=settings.VERIFICATION_TOKEN_EXPIRE_MINUTES)
    }

    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + expires_delta_condition[aud]

    to_encode = {'sub': sub, 'aud': aud, 'exp': expire, 'jti': str(uuid.uuid4())}
    encoded_jwt = jwt.encode(
        to_encode,
        private_key,
        algorithm=settings.JWT_TOKEN_ALGORITHM
    )
    return 

公钥验证和打开token

https://github.com/amirhosss/FastAPI-RS256-MongoDB-Redis/blob/master/api/dependencies/get_user.py

from typing import Optional

from fastapi import Depends, HTTPException, status, Cookie
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic.error_wrappers import ValidationError
from jose import jwt
from jose.exceptions import JWTError

import crud, models, schemas
from core.config import settings
from core.security import public_key
from utils.redis import redis

security = HTTPBearer(auto_error=settings.HTTP_BEARER_AUTO_ERROR)


class TokenRequired:
    def __init__(self, token_type) -> None:
        self.token_type = token_type

    def __call__(
        self,
        credentials: HTTPAuthorizationCredentials = Depends(security),
        access_token: Optional[str] = Cookie(None),
        refresh_token: Optional[str] = Cookie(None)

    ) -> Optional[str]:
        if credentials:
            return (token := credentials.credentials)
        elif access_token and self.token_type == 'access':
            return (token := access_token)
        elif refresh_token and self.token_type == 'refresh':
            return (token := refresh_token)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Not authenticated',
            headers={'WWW-Authenticate': 'Bearer'}
        )


access_token_required = TokenRequired(token_type='access')
refresh_token_required = TokenRequired(token_type='refresh')


class GetUser:
    def __init__(self, token: str, audience: str) -> None:
        self.token = token
        self.audience = audience

    async def decode(self) -> None:
        try:
            payload = jwt.decode(
                self.token,
                public_key,
                algorithms=[settings.JWT_TOKEN_ALGORITHM],
                audience=self.audience
            )
            token_data = schemas.TokenPayload(**payload)
        except (JWTError, ValidationError):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail='Could not validate credentials'
            )

        jti_status = await redis.get(str(token_data.jti))
        if jti_status and jti_status == 'true':
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail='Token is invalidated'
            )

        self.token_data = token_data

    async def get_current_user(self) -> models.User:
        await self.decode()
        user = await crud.user.read_by_id(self.token_data.sub)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail='User not found'
            )
        return user