Securing Endpoints

Learn the fundamentals of securing endpoints in FastAPI.

🎯 Core Concept

Securing endpoints requires multiple layers: authentication (verifying who users are), authorization (verifying what they can do), and protection against common attacks.

📖 What You'll Learn

In this section, you'll understand:

• Multiple authentication methods

• Authorization and permission checking

• Common security patterns

• Best practices for production APIs

💡 Basic Endpoint Security

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthenticationCredentials

app = FastAPI()
security = HTTPBearer()

async def verify_token(credentials: HTTPAuthenticationCredentials = Depends(security)):
    if credentials.credentials != "valid-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return credentials.credentials

@app.get("/protected")
async def protected_endpoint(token: str = Depends(verify_token)):
    return {"message": "This is protected"}

🔍 Multi-Layer Security

from typing import Optional
from pydantic import BaseModel

class User(BaseModel):
    id: int
    username: str
    roles: list[str]

async def get_current_user(token: str = Depends(security)) -> User:
    # Verify token
    try:
        user_id = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])["user_id"]
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    # Get user from database
    user = db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")

    return user

async def require_role(role: str):
    async def role_checker(current_user: User = Depends(get_current_user)):
        if role not in current_user.roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"User does not have {role} role"
            )
        return current_user
    return role_checker

# Usage
@app.get("/admin")
async def admin_panel(user: User = Depends(require_role("admin"))):
    return {"message": "Welcome admin"}

@app.post("/users")
async def create_user(user: UserCreate, admin_user: User = Depends(require_role("admin"))):
    return {"id": 1, **user.dict()}

💼 Production-Grade Security

from datetime import datetime, timedelta
import secrets
from functools import lru_cache

class SecurityConfig:
    algorithm = "HS256"
    access_token_expire_minutes = 30
    refresh_token_expire_days = 7

@lru_cache()
def get_settings():
    return SecurityConfig()

async def verify_secure_token(
    token: str = Depends(HTTPBearer()),
    settings: SecurityConfig = Depends(get_settings)
):
    try:
        payload = jwt.decode(
            token.credentials,
            SECRET_KEY,
            algorithms=[settings.algorithm]
        )

        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")

        # Check token expiration
        exp = payload.get("exp")
        if exp and datetime.utcfromtimestamp(exp) < datetime.utcnow():
            raise HTTPException(status_code=401, detail="Token expired")

        # Check token is in whitelist (not revoked)
        if token.credentials in token_blacklist:
            raise HTTPException(status_code=401, detail="Token revoked")

    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = db.get_user(user_id)
    if not user or user.disabled:
        raise HTTPException(status_code=401, detail="User not found or disabled")

    return user

@app.post("/token")
async def login(credentials: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(credentials.username, credentials.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token = create_access_token({"sub": user.id})
    return {"access_token": access_token, "token_type": "bearer"}

@app.post("/logout")
async def logout(user: User = Depends(verify_secure_token), token: str = Depends(HTTPBearer())):
    # Add token to blacklist
    token_blacklist.add(token.credentials)
    logger.info(f"User {user.id} logged out")
    return {"message": "Logged out successfully"}

@app.get("/profile")
async def get_profile(user: User = Depends(verify_secure_token)):
    return {"id": user.id, "username": user.username}

How it applies to real-world API development

Securing endpoints is critical for:

• **Data privacy**: Protect sensitive user information

• **Compliance**: Meet GDPR, HIPAA, PCI-DSS requirements

• **Preventing attacks**: Protect against unauthorized access

• **Audit trails**: Track who accessed what and when

Best practices for securing endpoints

• ✅ Always use HTTPS in production

• ✅ Implement strong authentication (JWT, OAuth2)

• ✅ Use role-based or attribute-based access control

• ✅ Implement token expiration and refresh

• ✅ Check permissions on every protected endpoint

• ✅ Log authentication and authorization events

• ✅ Use secure password hashing (bcrypt)

• ✅ Implement rate limiting to prevent brute force

• ✅ Validate and sanitize all inputs

• ✅ Use CORS to restrict cross-origin requests

🔑 Key Takeaways

• ✅ Understand the purpose of securing endpoints

• ✅ Know when to apply this pattern

• ✅ Recognize its benefits in real-world scenarios

• ✅ Be prepared to use it in your projects

Ready to explore more? Check out the advanced section for production patterns and edge cases.