Developing Secure REST APIs Using FastAPI

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints. It provides automatic interactive API documentation and offers great support for asynchronous programming. One of the key concerns while developing APIs is security. In this blog post, we will explore how to develop secure REST APIs using FastAPI, with practical examples.

Table of Contents

1. Introduction to FastAPI
2. Setting Up FastAPI
3. Building a Simple REST API
4. Security Best Practices
5. Implementing Authentication and Authorization
6. Rate Limiting
7. Input Validation and Serialization
8. Logging and Monitoring
9. Conclusion

1. Introduction to FastAPI

FastAPI is known for its speed, simplicity, and the ability to use asynchronous functions. It also has automatic interactive documentation and type-checking which makes it a robust choice for API development.

2. Setting Up FastAPI

First, let’s set up FastAPI. You can install FastAPI and an ASGI server (e.g., uvicorn) using pip:

pip install fastapi
pip install uvicorn

3. Building a Simple REST API

Let’s start by building a simple CRUD API for managing items.

# main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

items = []

@app.get("/items", response_model=List[Item])
async def get_items():
    return items

@app.post("/items", response_model=Item)
async def create_item(item: Item):
    items.append(item)
    return item

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    for item in items:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, updated_item: Item):
    for idx, item in enumerate(items):
        if item.id == item_id:
            items[idx] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    for idx, item in enumerate(items):
        if item.id == item_id:
            del items[idx]
            return {"message": "Item deleted successfully"}
    raise HTTPException(status_code=404, detail="Item not found")

Run the app using uvicorn:

uvicorn main:app --reload

Now you have a basic CRUD API. Next, let’s focus on securing it.

4. Security Best Practices

HTTPS

Always use HTTPS to encrypt data between the client and the server. In a production environment, use a reverse proxy like Nginx to handle HTTPS.

CORS

Configure Cross-Origin Resource Sharing (CORS) to control which domains can access your API.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # Adjust this as needed
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

5. Implementing Authentication and Authorization

Basic Authentication

Implementing basic authentication using HTTP Basic Auth.

from fastapi import Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, "admin")
    correct_password = secrets.compare_digest(credentials.password, "password")
    if not (correct_username and correct_password):
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    return credentials.username

@app.get("/secure-items", response_model=List[Item])
async def get_secure_items(username: str = Depends(get_current_username)):
    return items

OAuth2 with JWT

Implement OAuth2 with JWT tokens for better security.

from datetime import datetime, timedelta
from jose import JWTError, jwt

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Token(BaseModel):
    access_token: str
    token_type: str

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    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"}

6. Rate Limiting

Rate limiting helps to prevent abuse by limiting the number of requests a user can make in a given period.

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/items", response_model=List[Item])
@limiter.limit("5/minute")
async def get_items():
    return items

7. Input Validation and Serialization

FastAPI uses Pydantic models for data validation. Ensure that all inputs are validated to prevent injection attacks.

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

8. Logging and Monitoring

Implement logging to keep track of the API activities and errors.

import logging

logging.basicConfig(level=logging.INFO)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    idem = str(uuid4())
    logger.info(f"rid={idem} start request path={request.url.path}")
    response = await call_next(request)
    logger.info(f"rid={idem} completed_in={response.headers['x-process-time']}")
    return response

Conclusion

Developing secure REST APIs using FastAPI involves several best practices, including using HTTPS, implementing authentication and authorization, rate limiting, input validation, and logging. By following these practices, you can ensure that your APIs are secure and robust. FastAPI makes it easy to implement these security measures, providing a solid foundation for building secure applications.

FastAPI’s extensive documentation and built-in features make it an excellent choice for developing secure and high-performance APIs. Whether you’re building simple or complex APIs, FastAPI’s flexibility and efficiency can help you achieve your goals while maintaining a strong focus on security.

About:

VerticalServe Inc — Niche Cloud, Data & AI/ML Premier Consulting Company, Partnered with Google Cloud, Confluent, AWS, Azure…60+ Customers and many success stories..

Website: http://www.VerticalServe.com

Contact: contact@verticalserve.com

Successful Case Studies: http://verticalserve.com/success-stories.html

InsightLake Solutions: Our pre built solutions — http://www.InsightLake.com