Аутентификация: JWT, OAuth2 password flow, Depends

Auth - один из самых важных и easy-to-mess-up аспектов backend. В этом уроке - современный подход для REST API: JWT токены, password hashing через bcrypt, OAuth2 password flow в FastAPI, защита endpoints через Depends. Используем стандартные библиотеки и best practices.

Что такое JWT

JWT (JSON Web Token) - формат токена с тремя частями:

header.payload.signature

• header - алгоритм подписи (HS256, RS256)
• payload - данные (user_id, expiry, custom claims)
• signature - HMAC подпись для верификации

Пример:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzM1MzE1MjAwfQ.SUm3yKjY8M1...

Сервер выпускает token при login, клиент шлёт в каждом запросе. Сервер верифицирует подпись и читает payload.

Преимущества JWT:

• Stateless - не нужен server-side session storage
• Самодостаточен - все данные в токене
• Подписан - нельзя tampering без знания secret

Недостатки:

• Нельзя отозвать (revoke) без дополнительных механизмов
• Размер больше session-id

Установка библиотек

pip install "python-jose[cryptography]"      # JWT operations
pip install "passlib[bcrypt]"                 # password hashing
pip install "fastapi[all]"                    # FastAPI с form parsing

Password hashing - bcrypt

Никогда не храни пароли в plain text. Используй bcrypt:

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

# При регистрации
hashed = hash_password("user_password")
# Сохраняем hashed в БД

# При login
if verify_password(form_password, db_user.password_hash):
    # password correct

bcrypt:

• One-way (нельзя расшифровать)
• Slow by design (защита от brute force)
• Включает salt (защита от rainbow tables)
• Configurable cost factor

Создание JWT токена

from datetime import datetime, timedelta, timezone
from jose import jwt

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

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

token = create_access_token({"sub": "user@example.com"})
# eyJhbGc...

sub (subject) - идентификатор юзера. exp (expiration) - когда токен истечёт. Эти claims стандартизированы в RFC 7519.

SECRET_KEY критичен - кто знает, тот может выпускать valid токены. Должен быть длинной (256 bit для HS256), храниться в env (удобно подтягивать через Pydantic Settings), никогда в коде/git.

Декодирование и валидация

from jose import jwt, JWTError

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

payload = decode_token(token)
# {"sub": "user@example.com", "exp": 1735315200}

jwt.decode автоматически проверяет:

• Signature (через SECRET_KEY)
• Expiration (exp)
• Issuer/audience если заданы

Невалидная подпись или expired token → JWTError.

FastAPI security setup

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

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

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

class User(BaseModel):
    email: str
    name: str

OAuth2PasswordBearer - dependency, которая читает Authorization: Bearer <token> header.

Login endpoint

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # OAuth2PasswordRequestForm reads username/password from form
    user = get_user_from_db(form_data.username)
    if not user or not verify_password(form_data.password, user.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

Это OAuth2 password flow - простой, подходит для first-party clients (твой собственный frontend). Не для third-party (для них Authorization Code flow).

Защищённые endpoints

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    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])
        email = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user_from_db(email)
    if user is None:
        raise credentials_exception
    return user

@app.get("/me", response_model=User)
async def read_me(user: User = Depends(get_current_user)):
    return user

@app.get("/protected")
async def protected(user: User = Depends(get_current_user)):
    return {"message": f"Hello, {user.name}"}

get_current_user - dependency, которая:

1. Извлекает token из Authorization header
2. Декодирует JWT
3. Находит юзера в БД
4. Возвращает User объект или 401

Любой endpoint с Depends(get_current_user) защищён.

Refresh tokens

Access токены живут коротко (15-30 мин). При истечении нужен новый - но без повторного login. Решение: refresh tokens:

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=30)   # дольше живёт
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/refresh", response_model=Token)
async def refresh_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(401, "Invalid token type")
        email = payload.get("sub")
        # ... validate user existence
        new_access_token = create_access_token({"sub": email})
        return {"access_token": new_access_token, "token_type": "bearer"}
    except JWTError:
        raise HTTPException(401, "Invalid refresh token")

Refresh token хранится клиентом в secure storage (httpOnly cookie). Access token в memory или sessionStorage.

Token revocation

JWT нельзя revoke напрямую - подпись остаётся валидной. Решения:

1. Короткое время жизни access token (15 мин) Если accounts compromised - max 15 минут вреда.

2. Blocklist скомпрометированных tokens в Redis Проверять каждый request - есть ли token в blocklist. До expiry хранить.

3. JWT с versioning В payload user_version. Login инкрементирует версию в таблице пользователей. При logout - тоже. Проверять что версия в token == версия в БД.

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    user = get_user_from_db(payload["sub"])
    if user.token_version != payload.get("ver"):
        raise HTTPException(401, "Token revoked")
    return user

Полностью stateless с revocation несовместим - tradeoff.

Scopes - permissions

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"read": "Read access", "write": "Write access", "admin": "Admin"},
)

def create_access_token(data: dict, scopes: list[str]):
    to_encode = data.copy()
    to_encode.update({"scopes": scopes, "exp": ...})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(403, f"Missing scope: {scope}")
    return get_user_from_db(payload["sub"])

@app.get("/items", dependencies=[Security(get_current_user, scopes=["read"])])
def list_items():
    ...

@app.post("/items", dependencies=[Security(get_current_user, scopes=["write"])])
def create_item():
    ...

Scopes - стандарт OAuth2 для тонкого контроля разрешений.

CSRF и cookies

Если используешь cookies для auth, нужна CSRF защита:

from fastapi import Cookie

@app.post("/transfer")
async def transfer(
    user = Depends(get_user_from_cookie),
    csrf_token: str = Cookie(...),
    csrf_form: str = Form(...),
):
    if csrf_token != csrf_form:
        raise HTTPException(403, "CSRF token mismatch")

С header-based auth (Authorization: Bearer) CSRF меньше проблема - browser не шлёт custom headers automatically.

Безопасное хранение JWT на клиенте

Best: HttpOnly cookie с Secure + SameSite=Strict для access token, либо memory с refresh через httpOnly cookie.

Распространённые ошибки

1. Слабый SECRET_KEY

SECRET_KEY = "secret"   # КАТАСТРОФА

256+ битный random. Хранить в env var, не в коде. Регулярная ротация.

2. Algorithm = None атака

payload = jwt.decode(token, SECRET_KEY, algorithms=None)   # принимает любой

Atttacker может подделать token с alg=none. Всегда явный list: algorithms=["HS256"].

3. Нет expiration

to_encode = {"sub": email}   # нет exp
jwt.encode(to_encode, ...)

Token живёт вечно. ALWAYS set exp.

4. Логирование токенов

logger.info(f"User authenticated: {token}")   # ТОКЕН В ЛОГАХ

Никогда. Tokens это credentials. Если попадут в логи - breach.

5. Plain text passwords

if user.password == form.password:   # ПЛОХО - plain text сравнение

Always hash. Always verify против hash. Никогда не храни pure text.

6. JWT для session с large data

to_encode = {"sub": email, "all_permissions": [...100 items...], "user_data": {...}}

Token становится огромным (16KB), каждый request тащит. Используй JWT для identity, остальное - server-side через user lookup.

Полный пример

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

# Config
SECRET_KEY = os.getenv("JWT_SECRET")   # 256+ bit random
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Fake DB
users_db = {
    "alice@example.com": {
        "email": "alice@example.com",
        "name": "Alice",
        "password_hash": "$2b$12$KIX0nWiTczE3ZdyDmZcMxe...",   # hashed "secret"
    }
}

# Auth setup
app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class User(BaseModel):
    email: str
    name: str

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def authenticate(email: str, password: str):
    user = users_db.get(email)
    if not user or not verify_password(password, user["password_hash"]):
        return None
    return user

def create_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid token",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email = payload.get("sub")
        if not email:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user_data = users_db.get(email)
    if not user_data:
        raise credentials_exception
    return User(**user_data)

# Endpoints
@app.post("/token", response_model=Token)
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(401, "Incorrect username or password")
    token = create_token({"sub": user["email"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me", response_model=User)
async def me(user: User = Depends(get_current_user)):
    return user

Сравнение с Go и PHP

В Go популярны golang-jwt/jwt для JWT, crypto/bcrypt для password:

import "github.com/golang-jwt/jwt/v5"

token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
    "sub": userEmail,
    "exp": time.Now().Add(15 * time.Minute).Unix(),
})
tokenString, _ := token.SignedString([]byte(secretKey))

В PHP - firebase/php-jwt популярный, password_hash() для bcrypt built-in.

Концепции одинаковые - JWT универсальный стандарт, bcrypt де-факто для паролей. Различаются конкретные библиотеки и framework integration.

Мини-задание

Минимальный auth с FastAPI - смотри полный пример выше. Запусти, попробуй:

# Получить токен
curl -X POST -d "username=alice@example.com&password=secret" http://localhost:8000/token

# Использовать токен
curl -H "Authorization: Bearer <TOKEN>" http://localhost:8000/me

Что дальше

• Web - Cookie, session, localStorage - как клиент хранит токены: httpOnly cookies vs localStorage, атрибуты Secure и SameSite

Освоили auth. В следующем уроке - Docker и CI/CD для Python приложений. Это инфраструктурная часть production deployment.