Безопасность и аутентификация FastAPI

from datetime import datetime, timedelta
from typing import Annotated

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

# ===== Конфигурация =====
SECRET_KEY = "your-secret-key-keep-it-secret"  # Используйте переменные окружения в продакшене
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ===== Хеширование паролей =====
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# ===== Схема OAuth2 =====
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ===== Модели данных =====
class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


# ===== Фиктивная база данных =====
fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Алиса Уондерсон",
        "email": "[email protected]",
        "hashed_password": pwd_context.hash("secret"),  # Пароль: secret
        "disabled": False,
    }
}


# ===== Вспомогательные функции =====
def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Проверка пароля"""
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    """Генерация хеша пароля"""
    return pwd_context.hash(password)


def get_user(db: dict, username: str) -> UserInDB | None:
    """Получить пользователя из базы данных"""
    if username in db:
        return UserInDB(**db[username])
    return None


def authenticate_user(db: dict, username: str, password: str):
    """Проверить учетные данные пользователя"""
    user = get_user(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 = Отсутствует):
    """Создать JWT-токен доступа"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    """Получить текущего пользователя из токена (функция зависимости)"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=Не удалось проверить учетные данные,
        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)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    """Получить текущего активного пользователя"""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail=Пользователь был отключён)
    return current_user


# ===== Маршруты =====
app = FastAPI()


@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    """Вход для получения токена"""
    user = authenticate_user(fake_users_db, form_data.username, form_data.пароль)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверное имя пользователя или пароль",
            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: Annotated[User, Depends(get_current_active_user)],
):
    """Получить информацию о текущем пользователе (требуется аутентификация)"""
    return current_user


@app.get("/users/me/items")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    """Получить элементы текущего пользователя (требуется аутентификация)"""
    return [{"item_id": "Foo", "owner": current_user.username}]