FastAPI Security and Authentication - Interview Questions

1. What are the main security features in FastAPI?

FastAPI provides several built-in security features:

• Automatic Data Validation: Prevents injection attacks through Pydantic validation
• Type Safety: Reduces security vulnerabilities through type checking
• OpenAPI Security Schemes: Built-in support for OAuth2, HTTP Basic, API Keys
• CORS Support: Built-in CORS middleware for cross-origin requests
• HTTPS Support: Easy integration with SSL/TLS
• Dependency Injection: Secure handling of authentication and authorization
• Request/Response Validation: Automatic validation of all inputs and outputs

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

2. How do you implement JWT authentication in FastAPI?

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional

app = FastAPI()
security = HTTPBearer()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Configuration
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password hashing
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

# Token creation
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

# Token verification
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = get_user(username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            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"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

3. How do you implement OAuth2 with Password flow in FastAPI?

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# User model
class User:
    def __init__(self, username: str, hashed_password: str):
        self.username = username
        self.hashed_password = hashed_password

# Simulated user database
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "hashed_password": pwd_context.hash("secret"),
        "email": "johndoe@example.com",
        "full_name": "John Doe",
        "disabled": False,
    }
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return User(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: 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

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/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=status.HTTP_401_UNAUTHORIZED,
            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"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

4. How do you implement role-based access control (RBAC) in FastAPI?

from fastapi import FastAPI, Depends, HTTPException, status
from enum import Enum
from typing import List

app = FastAPI()

class Role(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class User:
    def __init__(self, username: str, roles: List[Role]):
        self.username = username
        self.roles = roles

def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    # Token validation logic here
    return User(username="john", roles=[Role.USER])

def require_role(required_role: Role):
    def role_checker(current_user: User = Depends(get_current_user)):
        if required_role not in current_user.roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Not enough permissions"
            )
        return current_user
    return role_checker

def require_any_role(required_roles: List[Role]):
    def role_checker(current_user: User = Depends(get_current_user)):
        if not any(role in current_user.roles for role in required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Not enough permissions"
            )
        return current_user
    return role_checker

# Endpoints with role-based access
@app.get("/admin-only")
async def admin_only(current_user: User = Depends(require_role(Role.ADMIN))):
    return {"message": "Admin access granted", "user": current_user.username}

@app.get("/moderator-or-admin")
async def moderator_or_admin(
    current_user: User = Depends(require_any_role([Role.MODERATOR, Role.ADMIN]))
):
    return {"message": "Moderator or admin access granted", "user": current_user.username}

@app.get("/public")
async def public_endpoint():
    return {"message": "Public access"}

@app.get("/user-only")
async def user_only(current_user: User = Depends(require_role(Role.USER))):
    return {"message": "User access granted", "user": current_user.username}

5. How do you implement API key authentication in FastAPI?

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security.api_key import APIKeyHeader, APIKeyQuery
from typing import Optional

app = FastAPI()

# API Key in header
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

# API Key in query parameter
api_key_query = APIKeyQuery(name="api_key", auto_error=False)

# Simulated API key database
API_KEYS = {
    "test-api-key-1": {"user_id": 1, "permissions": ["read", "write"]},
    "test-api-key-2": {"user_id": 2, "permissions": ["read"]},
}

def get_api_key_user(api_key: str = Depends(api_key_header)):
    if api_key not in API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )
    return API_KEYS[api_key]

def require_permission(permission: str):
    def permission_checker(api_key_user: dict = Depends(get_api_key_user)):
        if permission not in api_key_user["permissions"]:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Permission '{permission}' required"
            )
        return api_key_user
    return permission_checker

@app.get("/protected")
async def protected_endpoint(api_key_user: dict = Depends(get_api_key_user)):
    return {"message": "Access granted", "user_id": api_key_user["user_id"]}

@app.get("/read-only")
async def read_only_endpoint(api_key_user: dict = Depends(require_permission("read"))):
    return {"message": "Read access granted", "user_id": api_key_user["user_id"]}

@app.post("/write-endpoint")
async def write_endpoint(api_key_user: dict = Depends(require_permission("write"))):
    return {"message": "Write access granted", "user_id": api_key_user["user_id"]}

# Alternative: API key in query parameter
@app.get("/query-key")
async def query_key_endpoint(api_key: str = Depends(api_key_query)):
    if api_key not in API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )
    return {"message": "Access granted via query parameter"}

6. How do you implement CORS (Cross-Origin Resource Sharing) in FastAPI?

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Basic CORS setup
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

# More restrictive CORS setup
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",
        "https://yourdomain.com",
        "https://api.yourdomain.com"
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=[
        "Accept",
        "Accept-Language",
        "Content-Language",
        "Content-Type",
        "Authorization",
        "X-Requested-With"
    ],
    expose_headers=["Content-Length", "X-Total-Count"],
    max_age=600,  # Cache preflight requests for 10 minutes
)

# Environment-based CORS
import os

origins = [
    "http://localhost:3000",
    "http://localhost:8080",
]

if os.getenv("ENVIRONMENT") == "production":
    origins = [
        "https://yourdomain.com",
        "https://www.yourdomain.com",
    ]

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

7. How do you implement rate limiting for security in FastAPI?

from fastapi import FastAPI, HTTPException, Depends, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from collections import defaultdict
import time

app = FastAPI()

# Using slowapi for rate limiting
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/api/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
    return {"data": "some data"}

@app.post("/api/login")
@limiter.limit("3/minute")
async def login(request: Request):
    return {"message": "Login attempt"}

# Custom rate limiting implementation
class SecurityRateLimiter:
    def __init__(self):
        self.requests = defaultdict(list)
        self.limits = {
            "login": {"requests": 5, "window": 300},  # 5 requests per 5 minutes
            "api": {"requests": 100, "window": 60},   # 100 requests per minute
            "upload": {"requests": 10, "window": 3600}  # 10 uploads per hour
        }
    
    def is_allowed(self, client_id: str, endpoint: str) -> bool:
        now = time.time()
        limit_config = self.limits.get(endpoint, {"requests": 60, "window": 60})
        
        # Clean old requests
        window_start = now - limit_config["window"]
        self.requests[f"{client_id}:{endpoint}"] = [
            req_time for req_time in self.requests[f"{client_id}:{endpoint}"]
            if req_time > window_start
        ]
        
        current_requests = len(self.requests[f"{client_id}:{endpoint}"])
        
        if current_requests >= limit_config["requests"]:
            return False
        
        self.requests[f"{client_id}:{endpoint}"].append(now)
        return True

security_limiter = SecurityRateLimiter()

def check_rate_limit(endpoint: str):
    def rate_limit_checker(request: Request):
        client_id = request.client.host
        if not security_limiter.is_allowed(client_id, endpoint):
            raise HTTPException(
                status_code=429,
                detail=f"Rate limit exceeded for {endpoint}"
            )
    return rate_limit_checker

@app.post("/login")
async def login_endpoint(request: Request):
    check_rate_limit("login")(request)
    return {"message": "Login successful"}

@app.get("/api/protected")
async def protected_api(request: Request):
    check_rate_limit("api")(request)
    return {"data": "protected data"}

8. How do you implement input validation and sanitization for security?

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, validator, Field
import re
from typing import Optional
import html

app = FastAPI()

class UserInput(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    password: str = Field(..., min_length=8)
    bio: Optional[str] = Field(None, max_length=500)
    
    @validator('username')
    def validate_username(cls, v):
        # Check for SQL injection patterns
        sql_patterns = [
            r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER)\b)",
            r"(\b(UNION|EXEC|EXECUTE)\b)",
            r"(--|/\*|\*/|xp_|sp_)",
        ]
        
        for pattern in sql_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("Invalid username format")
        
        # Check for XSS patterns
        xss_patterns = [
            r"<script.*?>.*?</script>",
            r"javascript:",
            r"on\w+\s*=",
        ]
        
        for pattern in xss_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("Invalid username format")
        
        return v.strip()
    
    @validator('bio')
    def sanitize_bio(cls, v):
        if v:
            # HTML escape to prevent XSS
            v = html.escape(v)
            # Remove potentially dangerous tags
            v = re.sub(r'<[^>]*>', '', v)
            return v
        return v

class CommentInput(BaseModel):
    content: str = Field(..., max_length=1000)
    
    @validator('content')
    def sanitize_content(cls, v):
        # Remove HTML tags
        v = re.sub(r'<[^>]*>', '', v)
        # HTML escape
        v = html.escape(v)
        # Remove excessive whitespace
        v = re.sub(r'\s+', ' ', v).strip()
        return v

@app.post("/users")
async def create_user(user: UserInput):
    # Additional server-side validation
    if user.username.lower() in ['admin', 'root', 'system']:
        raise HTTPException(status_code=400, detail="Username not allowed")
    
    return {"message": "User created", "username": user.username}

@app.post("/comments")
async def create_comment(comment: CommentInput):
    return {"message": "Comment created", "content": comment.content}

# Custom validation decorator
def validate_file_upload(file_size_limit: int = 5 * 1024 * 1024):  # 5MB
    def validator(request: Request):
        content_length = request.headers.get('content-length')
        if content_length and int(content_length) > file_size_limit:
            raise HTTPException(
                status_code=413,
                detail=f"File too large. Maximum size is {file_size_limit} bytes"
            )
    return validator

@app.post("/upload")
async def upload_file(request: Request):
    validate_file_upload()(request)
    return {"message": "File upload validated"}

9. How do you implement secure headers in FastAPI?

from fastapi import FastAPI, Request, Response
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import secrets

app = FastAPI()

# Trusted Host middleware
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
)

# Custom middleware for security headers
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    
    # Security headers
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    response.headers["Permissions-Policy"] = "geolocation=(), microphone=()"
    
    # Remove server information
    response.headers.pop("Server", None)
    
    return response

# CSRF protection
class CSRFMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # Add CSRF token to state
            scope["state"]["csrf_token"] = secrets.token_urlsafe(32)
        
        await self.app(scope, receive, send)

app = CSRFMiddleware(app)

# Session security
from fastapi import FastAPI, Request, Response
from starlette.middleware.sessions import SessionMiddleware

app.add_middleware(
    SessionMiddleware,
    secret_key="your-secret-key-here",
    max_age=3600,  # 1 hour
    same_site="lax",
    https_only=True  # In production
)

@app.post("/secure-action")
async def secure_action(request: Request):
    # Verify CSRF token
    csrf_token = request.state.get("csrf_token")
    form_token = request.form().get("csrf_token")
    
    if not csrf_token or csrf_token != form_token:
        raise HTTPException(status_code=403, detail="CSRF token invalid")
    
    return {"message": "Action completed securely"}

10. How do you implement secure file uploads in FastAPI?

from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
from fastapi.responses import FileResponse
import os
import hashlib
import magic
from typing import List
import aiofiles

app = FastAPI()

# Allowed file types
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".pdf", ".txt"}
ALLOWED_MIME_TYPES = {
    "image/jpeg", "image/png", "image/gif", 
    "application/pdf", "text/plain"
}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB

def validate_file(file: UploadFile):
    # Check file size
    if file.size and file.size > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail=f"File too large. Maximum size is {MAX_FILE_SIZE} bytes"
        )
    
    # Check file extension
    file_extension = os.path.splitext(file.filename)[1].lower()
    if file_extension not in ALLOWED_EXTENSIONS:
        raise HTTPException(
            status_code=400,
            detail=f"File type not allowed. Allowed types: {ALLOWED_EXTENSIONS}"
        )
    
    return file

async def validate_file_content(file: UploadFile):
    # Read first 2048 bytes to check MIME type
    content = await file.read(2048)
    await file.seek(0)  # Reset file pointer
    
    # Check MIME type using python-magic
    mime_type = magic.from_buffer(content, mime=True)
    if mime_type not in ALLOWED_MIME_TYPES:
        raise HTTPException(
            status_code=400,
            detail=f"File content type not allowed: {mime_type}"
        )
    
    return file

def secure_filename(filename: str) -> str:
    """Generate a secure filename"""
    # Remove path separators and dangerous characters
    filename = os.path.basename(filename)
    filename = re.sub(r'[^\w\-_\.]', '_', filename)
    
    # Add hash to prevent conflicts
    name, ext = os.path.splitext(filename)
    hash_suffix = hashlib.md5(name.encode()).hexdigest()[:8]
    return f"{name}_{hash_suffix}{ext}"

@app.post("/upload")
async def upload_file(
    file: UploadFile = Depends(validate_file),
    validated_file: UploadFile = Depends(validate_file_content)
):
    # Generate secure filename
    secure_name = secure_filename(file.filename)
    file_path = f"uploads/{secure_name}"
    
    # Ensure upload directory exists
    os.makedirs("uploads", exist_ok=True)
    
    # Save file securely
    async with aiofiles.open(file_path, 'wb') as f:
        content = await file.read()
        await f.write(content)
    
    return {
        "filename": secure_name,
        "size": len(content),
        "message": "File uploaded successfully"
    }

@app.get("/files/{filename}")
async def get_file(filename: str):
    file_path = f"uploads/{filename}"
    
    # Validate filename to prevent path traversal
    if not os.path.exists(file_path) or ".." in filename:
        raise HTTPException(status_code=404, detail="File not found")
    
    return FileResponse(file_path)

# Multiple file upload with validation
@app.post("/upload-multiple")
async def upload_multiple_files(
    files: List[UploadFile] = File(...)
):
    uploaded_files = []
    
    for file in files:
        try:
            validated_file = validate_file(file)
            validated_content = await validate_file_content(validated_file)
            
            secure_name = secure_filename(file.filename)
            file_path = f"uploads/{secure_name}"
            
            async with aiofiles.open(file_path, 'wb') as f:
                content = await file.read()
                await f.write(content)
            
            uploaded_files.append({
                "filename": secure_name,
                "size": len(content)
            })
            
        except HTTPException as e:
            return {"error": f"Error uploading {file.filename}: {e.detail}"}
    
    return {"uploaded_files": uploaded_files}

Interview angle

• “How do you implement auth in FastAPI?” - a dependency that extracts and validates the credential and returns the current user, applied per route or per router. OAuth2PasswordBearer wires the token extraction and the OpenAPI security scheme together.
• “Where do you store the token in a browser client?” - an httpOnly, Secure, SameSite cookie rather than localStorage, which is readable by any XSS. If you use cookies you must handle CSRF; with a bearer header you don’t, but you’ve taken on XSS exposure instead.
• “How do you revoke a JWT?” - you can’t directly, which is the trade. Use short-lived access tokens with refresh tokens, plus a denylist keyed on a token id for immediate revocation. See ../../11_authentication/jwt/05_revocation_logout.md.
• “What’s easy to get wrong?” - verifying the signature but not the algorithm, audience or expiry; accepting alg: none; and putting authorisation decisions in the token rather than checking them server-side at use time.