Аутентификация: JWT, login endpoint, Depends

Resource Owner Password Credentials (ROPC) - это грант, в котором приложение собирает пароль пользователя от другого сервиса и обменивает его в чужом authorization server на токен. Именно он по RFC 9700 §2.4 - это актуальный Security Best Current Practice для OAuth 2.0 - MUST NOT be used. Формулировки RFC: грант «небезопасно раскрывает учётные данные владельца ресурса клиенту», расширяет поверхность атаки (пароль утекает не только из authorization server, но и из всех мест, куда его донёс клиент) и «приучает пользователей вводить свои учётные данные не на authorization server» - то есть готовит их к фишингу. Плюс он архитектурно несовместим с двухфакторной аутентификацией и любым входом в несколько шагов, а с WebAuthn не реализуем вовсе. В черновике OAuth 2.1 грант убран из спецификации.

Код выше - не ROPC. Здесь твой же сервис владеет паролями, сам их проверяет и выдаёт свой токен. Это обычный login endpoint; OAuth в нём - только имя класса OAuth2PasswordRequestForm, который читает username и password из формы, и схема Bearer в заголовке. Никакой делегации доступа нет, потому что делегировать некому.

Практический вывод из различия:

• Свой frontend к своему API - показанный endpoint подходит. Не называй его «password flow»: назовёшь так - и рано или поздно кто-то выставит его наружу как OAuth-эндпоинт для сторонних клиентов.
• Сторонние приложения к твоему API, или твоё приложение к чужому провайдеру (Google, GitLab, Keycloak) - только Authorization Code + PKCE. Пароль вводится на странице провайдера, приложение его никогда не видит.
• Никогда - не принимай в своём API пароль от чужого сервиса, чтобы сходить с ним куда-то за токеном. Это и есть запрещённый случай.

Защищённые 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:

import secrets

from fastapi import Cookie
from fastapi.responses import JSONResponse

def create_refresh_token(data: dict) -> tuple[str, str]:
    """Возвращает (токен, jti). jti нужен, чтобы токен можно было отозвать."""
    jti = secrets.token_urlsafe(16)
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=30)   # дольше живёт
    to_encode.update({"exp": expire, "type": "refresh", "jti": jti})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM), jti

@app.post("/refresh", response_model=Token)
async def refresh_token(
    # Cookie, а НЕ обычный параметр функции.
    #
    # `refresh_token: str` в FastAPI - это query-параметр, то есть токен
    # уехал бы в URL: access-логи, история браузера, заголовок Referer.
    # Для токена, живущего 30 дней, это равносильно его публикации.
    refresh_token: str = Cookie(...),
    db: AsyncSession = Depends(get_db),
):
    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")
        jti = payload.get("jti")
    except JWTError:
        raise HTTPException(401, "Invalid refresh token")

    # Ротация: старый refresh гасится, выдаётся новая пара.
    #
    # Именно поэтому refresh-токены приходится хранить на сервере -
    # хотя бы их идентификаторы (jti). Без этого «отозвать» токен
    # невозможно: подпись остаётся валидной до exp.
    stored = await db.get(RefreshToken, jti)
    if stored is None or stored.used_at is not None:
        # Токен неизвестен или уже использован. Второе - признак кражи:
        # легитимный клиент обменял бы его один раз. Гасим всю цепочку
        # сессий пользователя, а не только этот токен.
        await revoke_all_sessions(db, email)
        raise HTTPException(401, "Refresh token reuse detected")

    stored.used_at = datetime.now(timezone.utc)
    new_refresh, new_jti = create_refresh_token({"sub": email})
    db.add(RefreshToken(jti=new_jti, user_email=email))
    await db.commit()

    response = JSONResponse({
        "access_token": create_access_token({"sub": email}),
        "token_type": "bearer",
    })
    response.set_cookie(
        "refresh_token", new_refresh,
        httponly=True, secure=True, samesite="lax", path="/refresh",
    )
    return response

POST /logout
  → refresh token отозван, новую пару не получить
  → выданный access token работает до exp

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    # jwt.decode бросает JWTError на истёкшем, подделанном и просто
    # непарсящемся токене. Без перехвата это 500 вместо 401: клиент
    # решит, что сломался сервер, а в логи посыпятся трейсбеки
    # от обычного перебора.
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise HTTPException(401, "Invalid or expired token")

    user = get_user_from_db(payload["sub"])
    # Пользователя могли удалить, пока токен ещё валиден по подписи.
    # Без этой проверки следующая строка даёт AttributeError на None.
    if user is None:
        raise HTTPException(401, "Invalid or expired token")
    if user.token_version != payload.get("ver"):
        raise HTTPException(401, "Token revoked")
    return user

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)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        # 401, а не 500: токен невалиден - это ожидаемая ситуация,
        # а не сбой сервера.
        raise HTTPException(401, "Invalid or expired token")

    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():
    ...

import secrets

from fastapi import Cookie, Depends, Form, HTTPException

@app.post("/transfer")
async def transfer(
    user = Depends(get_user_from_cookie),
    csrf_token: str = Cookie(...),
    csrf_form: str = Form(...),
):
    # compare_digest, а не !=.
    #
    # Обычное сравнение строк прекращается на первом различающемся байте,
    # поэтому время ответа зависит от того, сколько символов совпало,
    # и токен можно подобрать побайтово. compare_digest сравнивает
    # за постоянное время.
    if not secrets.compare_digest(csrf_token, csrf_form):
        raise HTTPException(403, "CSRF token mismatch")

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

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

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

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

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

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

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

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))

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

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