Security in FastAPI: Best practices to protect your application (Part I)

bcrypt==4.1.2
fastapi==0.109.2
passlib==1.7.4
PyJWT==2.8.0
python-multipart==0.0.7
uvicorn==0.27.0.post1

import jwt
import secrets
import bcrypt
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
from fastapi.openapi.utils import get_openapi
from fastapi.security import HTTPBasic, HTTPBasicCredentials, HTTPBearer
from pydantic import BaseModel
from passlib.context import CryptContext
from datetime import datetime, timedelta

# Declaration of the HTTP Basic Authentication method
security = HTTPBasic()

# Declaration of the Bearer schema for token-based authentication
protocol = HTTPBearer(auto_error=False, scheme_name="Bearer")

# Initialization of the FastAPI application
app = FastAPI(
    title="API SECURITY TEST",
    description='This is SECURITY TEST API',
    version=f"0.0.1",
    docs_url=None,
    redoc_url=None,
    openapi_url=None
)

# Function to get the current username (It's a basic http login)
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, "securityapi")
    correct_password = secrets.compare_digest(credentials.password, "testpassword")
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

# Route to get Swagger documentation
@app.get("/docs", include_in_schema=False)
async def get_swagger_documentation(username: str = Depends(get_current_username)):
    return get_swagger_ui_html(openapi_url="/openapi.json", title="docs")

# Route to get Redoc documentation
@app.get("/redoc", include_in_schema=False)
async def get_redoc_documentation(username: str = Depends(get_current_username)):
    return get_redoc_html(openapi_url="/openapi.json", title="docs")

# Route to get the OpenAPI JSON schema
@app.get("/openapi.json", include_in_schema=False)
async def openapi(username: str = Depends(get_current_username)):
    return get_openapi(title=app.title, version=app.version, routes=app.routes, description=app.description)

# Configuration of encryption parameters
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Example of a registered user (for testing)
fake_users_db = {
    "user": {
        "username": "user",
        "full_name": "Test User",
        "email": "user@example.com",
        "hashed_password": "$2b$12$vJGN.aAs5mV.KWC2Czt3PujQHXy.SVeC1UIINSGBLD0vmTfVuiGgC",  # Password: testpassword
        "disabled": False,
    }
}

# Class to represent access token data
class Token(BaseModel):
    access_token: str
    token_type: str

# Class to represent user data
class User(BaseModel):
    username: str
    email: str
    full_name: str

# Class to represent user data stored in the database
class UserInDB(User):
    hashed_password: str

# Class to represent user login data
class UserInLogin(BaseModel):
    username: str
    password: str

# Object for password encryption and hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Function to verify if the password is correct
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8"))

# Function to hash the password
def get_password_hash(password: str) -> str:
    salt = bcrypt.gensalt()
    return (bcrypt.hashpw(password.encode('utf-8'), salt)).decode('utf-8')

# Function to generate a JWT access token
def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# Function to authenticate the user
def authenticate_user(fake_db, username: str, password: str):
    user = fake_db.get(username)
    if not user or not verify_password(password, user.get("hashed_password")):
        return False
    return user

# Function to get the current user
async def get_current_user(token: any = Depends(protocol)):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Unable to validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except Exception as e:
        raise credentials_exception
    return username

# Route for authentication and access token generation
@app.post("/login", response_model=Token)
async def login_for_access_token(username: str, password: str):
    user = authenticate_user(fake_users_db, username, password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "Bearer"}

# Route for current user information
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: str = Depends(get_current_user)):
    user = fake_users_db.get(current_user)
    return user