Как реализована безопасность в FastAPI?

# Как реализована безопасность в FastAPI?

FastAPI имеет встроенные механизмы безопасности, которые позволяют легко реализовать аутентификацию, авторизацию и защиту от распространённых атак.

Основные механизмы безопасности

1. CORS (Cross-Origin Resource Sharing)

Zащита от несанкционированного доступа с других доменов:

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

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://example.com",
        "https://subdomain.example.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

2. HTTPS/TLS

Шифрование трафика между клиентом и сервером:

# Генерируем самоподписанный сертификат
openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365

# Запускаем FastAPI с HTTPS
uvicorn main:app --ssl-keyfile=key.pem --ssl-certfile=cert.pem

Аутентификация и авторизация

HTTP Basic Auth

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

app = FastAPI()
security = HTTPBasic()

@app.get("/users/me")
async def read_current_user(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, "admin")
    correct_password = secrets.compare_digest(credentials.password, "secret")
    
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    return {"username": credentials.username}

Bearer Token (JWT)

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    
    encoded_jwt = jwt.encode(
        to_encode,
        SECRET_KEY,
        algorithm=ALGORITHM
    )
    return encoded_jwt

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(
            credentials.credentials,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
    
    return {"username": username}

@app.post("/login")
async def login(username: str, password: str):
    # В реальности проверяй пароль в БД
    if username != "admin" or password != "password":
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    access_token = create_access_token(data={"sub": username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
    return {"message": f"Hello, {current_user['username']}"}

OAuth2

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    token = create_access_token(data={"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/protected")
async def protected(token: str = Depends(oauth2_scheme)):
    return {"message": f"Token: {token}"}

Валидация данных

FastAPI автоматически валидирует входные данные:

from pydantic import BaseModel, Field, EmailStr

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr  # Валидация email
    age: int = Field(..., ge=0, le=150)  # от 0 до 150
    password: str = Field(..., min_length=8)

@app.post("/users")
async def create_user(user: User):
    # user уже валидирован и типизирован
    return user

SQL Injection Protection

FastAPI с SQLAlchemy ORM защищает от SQL инъекций:

from sqlalchemy import select
from sqlalchemy.orm import Session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    # ✅ Безопасно: ORM экранирует параметры
    user = db.query(User).filter(User.id == user_id).first()
    
    # ❌ Опасно: прямые SQL запросы
    # user = db.execute(f"SELECT * FROM users WHERE id = {user_id}")
    
    return user

Защита от XSS (Cross-Site Scripting)

from fastapi.responses import HTMLResponse
from markupsafe import escape

@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(id: str):
    # ✅ Экранируем пользовательский ввод
    return f"<h1>{escape(id)}</h1>"
    
    # ❌ Опасно: без экранирования
    # return f"<h1>{id}</h1>"

Защита от CSRF (Cross-Site Request Forgery)

from fastapi.middleware.csrf import CsrfProtectMiddleware

app.add_middleware(
    CsrfProtectMiddleware,
    cookie_key="csrf-protect",
)

Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/limited")
@limiter.limit("5/minute")
async def rate_limited_route(request):
    return {"message": "This endpoint is rate limited"}

Логирование и мониторинг

import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

Переменные окружения для секретов

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    secret_key: str  # Из .env файла
    database_url: str
    debug: bool = False
    
    class Config:
        env_file = ".env"

settings = Settings()

SECRET_KEY = settings.secret_key  # Никогда не в коде!

Безопасность заголовков

from fastapi.middleware import Middleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI(
    middleware=[
        Middleware(
            TrustedHostMiddleware,
            allowed_hosts=["example.com", "*.example.com"]
        )
    ]
)

# Добавь безопасные заголовки
@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    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"
    return response

Лучшие практики

1. Никогда не коммить секреты — используй .env файлы
2. Всегда используй HTTPS в production
3. Валидируй все входные данные — используй Pydantic
4. Логируй попытки неавторизованного доступа
5. Используй strong passwords и хешируй их (bcrypt, argon2)
6. Регулярно обновляй зависимости для патчей безопасности
7. Используй rate limiting для защиты от brute-force атак
8. Тестируй безопасность — пентесты и статический анализ

Заключение

FastAPI предоставляет мощные встроенные инструменты для реализации безопасности. Главное — правильно их использовать и следовать лучшим практикам.