Configured JWT Authentication using fastapi-jwt-auth

Author: s3m3dov

.gitignore

@@ -167,4 +167,3 @@ dmypy.json
 
 # Generated by Windows
 Thumbs.db
-

api/endpoints/items.py

@@ -1,8 +1,7 @@
 from typing import List
-from fastapi import (
-    APIRouter,
-    Depends,
-)
+
+from fastapi import APIRouter, Depends
+from fastapi_jwt_auth import AuthJWT
 
 import schemas
 import crud
@@ -21,6 +20,7 @@ def create_item_for_user(user_id: int, item: schemas.ItemCreate):
 
 
 @router.get("/items/", response_model=List[schemas.Item], dependencies=[Depends(get_db)])
-def read_items(skip: int = 0, limit: int = 100):
+def user_items(Authorize: AuthJWT = Depends(), skip: int = 0, limit: int = 100):
+    Authorize.jwt_required()
     items = crud.get_items(skip=skip, limit=limit)
     return items

api/endpoints/users.py

@@ -1,26 +1,83 @@
 import time
+from datetime import timedelta
 from typing import List
 
-from fastapi import (
-    APIRouter,
-    HTTPException,
-    Depends,
-)
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi_jwt_auth import AuthJWT
 
 import schemas
 import crud
+from core import settings
 from db.utils import get_db
 
 router = APIRouter()
 sleep_time = 10
 
 
+@router.post('/login/')
+def login(user_in: schemas.UserLogin, Authorize: AuthJWT = Depends()):
+    db_user = crud.authenticate_user(
+        email=user_in.email, password=user_in.password
+    )
+    if not db_user:
+        raise HTTPException(status_code=400, detail="Incorrect email or password!")
+
+    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    refresh_token_expires = timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES)
+
+    access_token = Authorize.create_access_token(
+        subject=db_user.email,
+        expires_time=access_token_expires,
+        algorithm=settings.AUTH_JWT_TOKEN_ALGORITHM
+    )
+    refresh_token = Authorize.create_refresh_token(
+        subject=db_user.email,
+        expires_time=refresh_token_expires,
+        algorithm=settings.AUTH_JWT_TOKEN_ALGORITHM
+    )
+
+    Authorize.set_access_cookies(access_token)
+    Authorize.set_refresh_cookies(refresh_token)
+
+    return {"msg": "Successfully logged in.", "access_token": access_token}
+
+
+@router.get('/user/', response_model=schemas.User, dependencies=[Depends(get_db)])
+def user(Authorize: AuthJWT = Depends()):
+    Authorize.jwt_required()
+
+    auth_user_email = Authorize.get_jwt_subject()
+    db_user = crud.get_user_by_email(auth_user_email)
+
+    return db_user
+
+
+@router.post('/refresh/')
+def refresh(Authorize: AuthJWT = Depends()):
+    Authorize.jwt_refresh_token_required()
+
+    auth_user_email = Authorize.get_jwt_subject()
+
+    new_access_token = Authorize.create_access_token(subject=auth_user_email)
+    Authorize.set_access_cookies(new_access_token)
+
+    return {"msg": "The token has been refreshed.", "access_token": new_access_token}
+
+
+@router.delete('/logout/')
+def logout(Authorize: AuthJWT = Depends()):
+    Authorize.jwt_required()
+
+    Authorize.unset_jwt_cookies()
+    return {"msg": "Successfully logged out."}
+
+
 @router.post("/users/", response_model=schemas.User, dependencies=[Depends(get_db)])
-def create_user(user: schemas.UserCreate):
-    db_user = crud.get_user_by_email(email=user.email)
+def create_user(user_in: schemas.UserCreate):
+    db_user = crud.get_user_by_email(email=user_in.email)
     if db_user:
-        raise HTTPException(status_code=400, detail="Email already registered")
-    return crud.create_user(user=user)
+        raise HTTPException(status_code=400, detail="The user with this email already exists in the system!", )
+    return crud.create_user(obj_in=user_in)
 
 
 @router.get("/users/", response_model=List[schemas.User], dependencies=[Depends(get_db)])

core/config.py

@@ -7,12 +7,19 @@
 
 class Settings:
     API_V1_STR: str = "/api/v1"
-    IPDATA_API_KEY: str = env("IPDATA_API_KEY", default="test")  # Your Key Here
+
+    IPDATA_API_KEY: str = env("IPDATA_API_KEY", default="test")
+    AUTH_JWT_SECRET_KEY: str = env("AUTH_JWT_SECRET_KEY", default="secret")
+    AUTH_JWT_TOKEN_ALGORITHM = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1
+    REFRESH_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7
+    # 60 minutes * 24 hours * x days = x days
 
     # Database Settings
     DATABASE_NAME: str = env("MYSQL_DATABASE", default="mysql_db")
     DATABASE_USER: str = env("MYSQL_USER", default="mysql")
     DATABASE_PASSWORD: str = env("MYSQL_PASSWORD", default="mysql")
+    DATABASE_ROOT_PASSWORD: str = env("MYSQL_ROOT_PASSWORD", default="mysql")
     DATABASE_HOST: str = "db"
     DATABASE_PORT: int = env.int("MYSQL_PORT", default=3306)
 

core/security.py

@@ -0,0 +1,23 @@
+from typing import Optional
+
+from fastapi import Cookie
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str) -> str:
+    return pwd_context.hash(password)
+
+
+async def get_access_token_from_cookie(csrf_access_token: Optional[str] = Cookie(None)):
+    return csrf_access_token
+
+
+async def get_refresh_token_from_cookie(csrf_refresh_token: Optional[str] = Cookie(None)):
+    return csrf_refresh_token
+

crud/__init__.py

@@ -7,5 +7,6 @@
     get_user,
     get_user_by_email,
     get_users,
-    create_user
+    create_user,
+    authenticate_user,
 )

crud/crud_user.py

@@ -1,5 +1,6 @@
 import schemas
 from db import models
+from core.security import pwd_context, verify_password, get_password_hash
 
 
 def get_user(user_id: int):
@@ -14,8 +15,21 @@ def get_users(skip: int = 0, limit: int = 100):
     return list(models.User.select().offset(skip).limit(limit))
 
 
-def create_user(user: schemas.UserCreate):
-    fake_hashed_password = user.password + "notreallyhashed"
-    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
+def create_user(obj_in: schemas.UserCreate):
+    db_user = models.User(
+        email=obj_in.email,
+        hashed_password=get_password_hash(obj_in.password),
+        fullname=obj_in.fullname,
+        is_superuser=obj_in.is_superuser,
+    )
     db_user.save()
     return db_user
+
+
+def authenticate_user(email: str, password: str):
+    user = get_user_by_email(email=email)
+    if not user:
+        return None
+    if not verify_password(password, user.hashed_password):
+        return None
+    return user

db/models.py

@@ -6,6 +6,8 @@ class User(peewee.Model):
     email = peewee.CharField(unique=True, index=True)
     hashed_password = peewee.CharField()
     is_active = peewee.BooleanField(default=True)
+    is_superuser = peewee.BooleanField(default=False)
+    fullname = peewee.CharField(null=True)
 
     class Meta:
         database = mysql_db
@@ -17,4 +19,4 @@ class Item(peewee.Model):
     owner = peewee.ForeignKeyField(User, backref="items")
 
     class Meta:
-        database = mysql_db
+        database = mysql_db

docker-compose.yml

@@ -6,6 +6,7 @@ services:
     command: bash -c "python main.py"
     environment:
       - IPDATA_API_KEY=test
+      - AUTH_JWT_SECRET_KEY=test
     volumes:
       - .:/code
     ports:

main.py

@@ -1,21 +1,39 @@
 import uvicorn
-from fastapi import FastAPI
 
-from core import settings
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from fastapi_jwt_auth import AuthJWT
+from fastapi_jwt_auth.exceptions import AuthJWTException
 
+from core import settings
 from db import models
 from db.init_db import mysql_db
-from api import api_router
 
+from api import api_router
+from schemas.auth_jwt_config import Settings
 
 mysql_db.connect()
 mysql_db.create_tables([models.User, models.Item])
 mysql_db.close()
 
-
 app = FastAPI(
     openapi_url=f"{settings.API_V1_STR}/openapi.json"
 )
+
+
+@AuthJWT.load_config
+def get_config():
+    return Settings()
+
+
+@app.exception_handler(AuthJWTException)
+def auth_jwt_exception_handler(request: Request, exc: AuthJWTException):
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"detail": exc.message}
+    )
+
+
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
 

requirements.txt

@@ -1,9 +1,11 @@
 fastapi~=0.73.0
 uvicorn~=0.17.3
 peewee~=3.14.8
-pydantic~=1.9.0
+pydantic[email]~=1.9.0
 contextvars~=2.4.0
 pymysql~=1.0.2
 cryptography~=36.0.1
 environs~=9.5.0
-aiohttp~=3.8.1
+aiohttp~=3.8.1
+fastapi-jwt-auth~=0.5.0
+passlib[bcrypt]~=1.7.4

schemas/__init__.py

@@ -4,6 +4,8 @@
 )
 
 from .user import (
+    UserBase,
+    UserLogin,
     UserCreate,
     User
 )

schemas/auth_jwt_config.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+from core import settings
+
+
+class Settings(BaseModel):
+    authjwt_secret_key: str = settings.AUTH_JWT_SECRET_KEY
+    authjwt_token_location: set = {"cookies"}
+    authjwt_cookie_secure: bool = False  # Only allow JWT cookies to be sent over https
+    authjwt_cookie_csrf_protect: bool = False  # Change it on production

schemas/user.py

@@ -1,23 +1,42 @@
-from typing import List
-from pydantic import BaseModel
+from typing import List, Optional
+from pydantic import BaseModel, EmailStr
 
 from .utils import PeeweeGetterDict
 from .item import Item
 
 
 class UserBase(BaseModel):
-    email: str
+    email: Optional[EmailStr] = None
+    is_active: Optional[bool] = True
+    is_superuser: bool = False
+    fullname: Optional[str] = None
+
+
+class UserLogin(BaseModel):
+    email: EmailStr
+    password: str
 
 
 class UserCreate(UserBase):
+    email: EmailStr
     password: str
+    fullname: Optional[str] = None
 
 
-class User(UserBase):
-    id: int
-    is_active: bool
+class UserInDBBase(UserBase):
+    id: Optional[int] = None
     items: List[Item] = []
 
     class Config:
         orm_mode = True
         getter_dict = PeeweeGetterDict
+
+
+# Additional properties to return via API
+class User(UserInDBBase):
+    pass
+
+
+# Additional properties stored in DB
+class UserInDB(UserInDBBase):
+    hashed_password: str