Приклад налаштування системи автентифікації з OAuth2 та JWT

Після цієї лекції ви створите систему автентифікації з використанням OAuth2 та JWT, яка включатиме реєстрацію, вхід користувачів, оновлення токенів і захист ендпоінтів. Ми зберемо воєдино попередні знання і реалізуємо все це в FastAPI. Готуйте каву ☕ і голову 🧠 – буде цікаво!

---

1. Початкова структура проєкту

Припустимо, що ми вже створили проєкт FastAPI з простою структурою. У нас є файл main.py, де підключено наш сервер. Робити все з нуля не будемо, бо ви вже профі, тож краще зосередимось на створенні системи автентифікації.

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   ├── models.py
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── routes.py
│   │   ├── jwt.py
│   │   ├── schemas.py
│   └── database.py

Для простоти в цій лекції ми будемо використовувати SQLite як базу даних. У реальному проєкті ви можете замінити її на PostgreSQL або будь-яку іншу СУБД.

---

2. Налаштовуємо базу даних і моделі

Спочатку додамо базу даних і створимо модель користувача. Будемо використовувати sqlalchemy для взаємодії з базою, а pydantic для валідації даних.

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

models.py

from sqlalchemy import Column, Integer, String, Boolean
from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean, default=False)

Тепер створимо таблицю:

>>> from app.database import Base, engine
>>> Base.metadata.create_all(bind=engine)  # Генерація таблиць з моделі

---

3. Створюємо Pydantic-схеми для даних

Тепер додамо schemas.py для роботи з вхідними і вихідними даними.

auth/schemas.py

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    is_active: bool

    class Config:
        orm_mode = True

---

4. Реалізація JWT: генерація та перевірка токенів

Наша наступна ціль — робота з JWT. Додаємо окремий модуль для генерації та перевірки токенів.

auth/jwt.py

from datetime import datetime, timedelta
from jose import JWTError, jwt

SECRET_KEY = "your_secret_key"  # Замініть на більш надійний ключ
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        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

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

---

5. Реалізація маршрутів автентифікації

Тепер займемося додаванням маршрутів для реєстрації, входу й отримання токенів.

auth/routes.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import SessionLocal
from app.models import User
from app.auth.schemas import UserCreate
from app.auth.jwt import create_access_token
from passlib.context import CryptContext

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

def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

@router.post("/register")
def register(user: UserCreate, db: Session = Depends(get_db)):
    existing_user = get_user_by_email(db, user.email)
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    hashed_password = pwd_context.hash(user.password)
    new_user = User(email=user.email, hashed_password=hashed_password)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return {"msg": "User created successfully"}

@router.post("/login")
def login(user: UserCreate, db: Session = Depends(get_db)):
    db_user = get_user_by_email(db, user.email)
    if not db_user or not pwd_context.verify(user.password, db_user.hashed_password):
        raise HTTPException(status_code=400, detail="Invalid credentials")
    
    access_token = create_access_token({"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

---

6. Захищаємо ендпоінти

Тепер додамо захист ендпоінтів із використанням токена доступу. Для цього створимо залежність, що перевіряє валідність токена.

dependencies.py

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from app.auth.jwt import verify_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = verify_token(token)
    if payload is None:
        raise HTTPException(status_code=401, detail="Invalid authentication token")
    return payload

Приклад захищеного ендпоінта

@router.get("/protected")
def protected_route(current_user=Depends(get_current_user)):
    return {"msg": f"Hello, {current_user['sub']}!"}

---

7. Тестуємо нашу систему

Запустіть сервер і протестуйте:

1. Зареєструйте користувача:

   
   curl -X POST "http://127.0.0.1:8000/register" -H "Content-Type: application/json" -d '{"email":"user@test.com", "password":"password123"}'
   

2. Отримайте токен:

   
   curl -X POST "http://127.0.0.1:8000/login" -H "Content-Type: application/json" -d '{"email":"user@test.com", "password":"password123"}'
   

3. Використайте токен для доступу до захищеного ендпоінта:

   
   curl -X GET "http://127.0.0.1:8000/protected" -H "Authorization: Bearer <ВАШ_ТОКЕН>"
   

---

Браво! Ви створили працюючу систему автентифікації з використанням OAuth2 і JWT, готову до реального боєвого застосування!