Building a Secure API with FastAPI, PostgreSQL, and Hanko Authentication

In the continuous exponential growth of internet and web application users, building robust, secure and user-friendly API is a necessity. In this blog post, we'll explore the development of a fastapi application using a powerful tech stack consisting of FastAPI, PostgreSQL, and integrating Hanko authentication for enhanced security.

Introduction to the Tech Stack:

FastAPI: Known for its high performance, simplicity, and ease of development, FastAPI is a modern Python web framework that enables the rapid creation of APIs with automatic interactive documentation.
PostgreSQL: A powerful open-source relational database known for its reliability and robustness, PostgreSQL is a popular choice for data storage in various applications.
Hanko Authentication: Hanko provides a secure, passwordless authentication solution that enhances user security and experience by utilizing biometric authentication methods, such as fingerprint, face or iris recognition.

Development Process

Project directory: Here is a brief overview of the project directory structure.

|---- project-directory/
| |---- backend/
| | |---- app/
| | | |---- alembic/
| | | | |---- env.py
| | | | |---- script.py.mako
| | | |---- db/
| | | | |---- base.py
| | | | |---- base_class.py
| | | | |---- session.py
| | | | |---- __init__.py
| | | |---- models/
| | | | |---- item.py
| | | | |---- __init__.py
| | | |------routers/
| | | | |---item.py
| | | |---- schemas/
| | | | |---- item.py
| | | | |---- __init__.py
| | | |---- alembic.ini
| | | |---- config.py
| | | |---- main.py
| | | |---- utils.py
| | | |---- __init__.py

Setting up the Backend with FastAPI and PostgreSQL: poetry will be used to manage the backend project. Visit the official website of poetry to download and install poetry then initialize poetry dependency management in the backend project.

cd backend
poetry init

follow the default options a pyproject.toml is created and poetry is setup for use. To install a package the command poetry add [package-name] is used a poetry.lock file is created, for subsequent package installations, the poetry.lock and pyproject.toml files are updated. You can open these files to inspect the contents.

installing the packages

poetry add fastapi[uvicorn] psycopg2-binary sqlalchemy pydantic-settings pyjwt alembic
uvicorn is the server that will be used to serve the fastapi application. pyscopg2-binary is a python-postgres interface that provides a means of connecting to postgresql. sqlalchemy is an orm mapper for relational databases. It is worth noting that pydantic is a core feature of the fastAPI framework.

main.py

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from app.config import settings
from app.routers import item

app = FastAPI(
    title=settings.PROJECT_NAME, openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(item.router, prefix=settings.API_V1_STR)

The main.py file is the entry point into the application. It defines the middlewares that the application needs and the routes associated with the application.
The CORSMiddleware in FastAPI is necessary to handle Cross-Origin Resource Sharing (CORS) policies effectively. CORS is a security feature implemented by web browsers to protect against cross-origin requests .When a frontend web application running in one domain (origin) wants to make requests to a backend API server in another domain, the browser enforces CORS to prevent potential security risks.

utils.py

import ssl
from typing import Generator
from fastapi import HTTPException, Request, status
import jwt
from pydantic import ValidationError
from app.config import settings


from app.db.session import SessionLocal


def get_db() -> Generator:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


def deny():
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Unauthorized",
    )


def extract_token_from_header(header: str):
    parts = header.split(" ")
    return parts[1] if len(parts) == 2 and parts[0].lower() == "bearer" else None


def get_current_user(request: Request) -> str:
    authorization = request.headers.get("authorization")

    if not (authorization):
        return deny()

    token = extract_token_from_header(authorization)

    if not token:
        return deny()
    try:
        ssl_context = ssl.create_default_context()
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE

        jwks_client = jwt.PyJWKClient(
            settings.HANKO_API_URL + "/.well-known/jwks.json", ssl_context=ssl_context
        )
        signing_key = jwks_client.get_signing_key_from_jwt(token)

        data = jwt.decode(
            token,
            signing_key.key,
            algorithms=[settings.ALGORITHM],
            audience="localhost",  # str(settings.SERVER_HOST),
        )
    except (jwt.PyJWTError, ValidationError):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Could not validate credentials",
        )
    user = data.get("sub")

    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

The get_current_user function is a critical part of the authentication process. It checks the request's authorization header from the client(frontend) application, extracts the token, attempts to retrieve the signing key from Hanko authorization API, validate and decode the JWT token. If successful, it returns the user id associated with the token; otherwise, it raises appropriate HTTP exceptions.

routers/item.py

from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, schemas
from app.utils import get_current_user, get_db

router = APIRouter(prefix="/item")


@router.get("/", response_model=List[schemas.Item])
def read_items(
    db: Session = Depends(get_db),
    skip: int = 0,
    limit: int = 100,
    current_user=Depends(get_current_user),
) -> Any:
    """
    Retrieve items.
    """
    if crud.user.is_superuser(current_user):
        items = crud.item.get_multi(db, skip=skip, limit=limit)
    else:
        items = crud.item.get_multi_by_owner(
            db=db, owner_id=current_user, skip=skip, limit=limit
        )
    return items


@router.post("/", response_model=schemas.Item)
def create_item(
    *,
    db: Session = Depends(get_db),
    item_in: schemas.ItemCreate,
    current_user=Depends(get_current_user),
) -> Any:
    """
    Create new item.
    """
    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=current_user)
    return item


@router.put("/{id}", response_model=schemas.Item)
def update_item(
    *,
    db: Session = Depends(get_db),
    id: int,
    item_in: schemas.ItemUpdate,
    current_user=Depends(get_current_user),
) -> Any:
    """
    Update an item.
    """
    item = crud.item.get(db=db, id=id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    if item.owner_id != current_user:
        raise HTTPException(status_code=400, detail="Not enough permissions")
    item = crud.item.update(db=db, db_obj=item, obj_in=item_in)
    return item


@router.get("/{id}", response_model=schemas.Item)
def read_item(
    *,
    db: Session = Depends(get_db),
    id: int,
    current_user=Depends(get_current_user),
) -> Any:
    """
    Get item by ID.
    """
    item = crud.item.get(db=db, id=id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    if item.owner_id != current_user:
        raise HTTPException(status_code=400, detail="Not enough permissions")
    return item


@router.delete("/{id}", response_model=schemas.Item)
def delete_item(
    *,
    db: Session = Depends(get_db),
    id: int,
    current_user=Depends(get_current_user),
) -> Any:
    """
    Delete an item.
    """
    item = crud.item.get(db=db, id=id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    if item.owner_id != current_user:
        raise HTTPException(status_code=400, detail="Not enough permissions")
    item = crud.item.remove(db=db, id=id)
    return item

in the routers/item.py file, each route depends on the get_current_user, if the current user is not available the instructions in the route will not be executed and as such the route is protected.

The complete Backend code is available on github fastapi-hanko and the production ready fullstack application with Vue2.js is available at authflowx. Also visit Hanko documentation on how you can integrate your favorite frontend framework.
Benefits and Use Cases:

Enhanced Security: Leveraging Hanko's biometric authentication adds an extra layer of security, eliminating the need for traditional passwords and reducing the risk of unauthorized access.
Scalability and Performance: The combination of FastAPI and PostgreSQL ensures high performance and scalability, allowing the application to handle a growing user base and data load efficiently.
User-Friendly Experience: With any frontend framework and Hanko for authentication, users can enjoy a seamless and intuitive experience while ensuring their data remains secure.

This project is a modification of the authentication flow of the awesome repository made by tiangolo at full-stack-fastapi-postgresql

Blog

Building a Secure API with FastAPI, PostgreSQL, and Hanko Authentication

Akachi Anabanti

Join Our Newsletter. No Spam, Only the good stuff.

Related