Как работает авторизация на сайте?

Как работает авторизация на сайте

Авторизация — это процесс подтверждения того, что пользователь имеет доступ к определённым ресурсам. Это отличается от аутентификации (проверка личности) и аутентификации (выдача прав доступа).

Жизненный цикл авторизации

1. Регистрация

Пользователь создает учетную запись:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from passlib.context import CryptContext
import secrets

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

class UserRegister(BaseModel):
    email: EmailStr
    password: str
    name: str

@app.post("/api/v1/auth/register")
async def register(user_data: UserRegister):
    # Проверяем, не зарегистрирован ли уже
    existing = await db.get_user_by_email(user_data.email)
    if existing:
        return {"error": "Email already registered"}, 400
    
    # Хешируем пароль (никогда не храним в открытом виде)
    hashed_password = pwd_context.hash(user_data.password)
    
    # Создаём пользователя
    user = await db.create_user(
        email=user_data.email,
        password_hash=hashed_password,
        name=user_data.name
    )
    
    return {"id": user.id, "email": user.email, "name": user.name}

2. Аутентификация (Логин)

Пользователь вводит учетные данные:

from datetime import datetime, timedelta
import jwt
from typing import Dict

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Login(BaseModel):
    email: str
    password: str

@app.post("/api/v1/auth/login")
async def login(credentials: Login):
    # Получаем пользователя из БД
    user = await db.get_user_by_email(credentials.email)
    
    # Проверяем существование и пароль
    if not user or not pwd_context.verify(credentials.password, user.password_hash):
        return {"error": "Invalid email or password"}, 401
    
    # Проверяем, активен ли пользователь
    if not user.is_active:
        return {"error": "Account disabled"}, 403
    
    # Создаём JWT токены
    access_token = create_access_token(user.id)
    refresh_token = create_refresh_token(user.id)
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "user": {"id": user.id, "email": user.email, "name": user.name}
    }

def create_access_token(user_id: str) -> str:
    """Создать JWT access token с коротким сроком действия."""
    payload = {
        "sub": user_id,
        "type": "access",
        "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: str) -> str:
    """Создать JWT refresh token с длительным сроком действия."""
    payload = {
        "sub": user_id,
        "type": "refresh",
        "exp": datetime.utcnow() + timedelta(days=7),
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

3. Использование Access Token

Клиент отправляет токен в заголовке Authorization:

// Frontend (JavaScript)
const response = await fetch('/api/v1/users/me', {
  headers: {
    'Authorization': `Bearer ${accessToken}`
  }
});
const user = await response.json();

Bэкенд проверяет токен:

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> Dict:
    """Проверить JWT токен."""
    token = credentials.credentials
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        token_type = payload.get("type")
        
        if not user_id or token_type != "access":
            raise HTTPException(status_code=401, detail="Invalid token")
        
        return {"user_id": user_id}
    
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/api/v1/users/me")
async def get_current_user(token: Dict = Depends(verify_token)):
    user = await db.get_user_by_id(token["user_id"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "email": user.email, "name": user.name}

4. Обновление токена (Refresh)

Когда access token истекает, используем refresh:

@app.post("/api/v1/auth/refresh")
async def refresh_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        token_type = payload.get("type")
        
        if not user_id or token_type != "refresh":
            raise HTTPException(status_code=401, detail="Invalid refresh token")
        
        # Проверяем в БД (если используем blacklist)
        if not await db.refresh_token_exists(user_id, refresh_token):
            raise HTTPException(status_code=401, detail="Refresh token revoked")
        
        # Создаём новый access token
        new_access = create_access_token(user_id)
        
        return {
            "access_token": new_access,
            "token_type": "bearer"
        }
    
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Refresh token expired")

5. Разделение прав доступа (Authorization)

Роли и права

from enum import Enum
from typing import List

class Role(str, Enum):
    ADMIN = "admin"
    MODERATOR = "moderator"
    USER = "user"

def check_role(required_role: Role):
    async def verify(token: Dict = Depends(verify_token)) -> Dict:
        user = await db.get_user_by_id(token["user_id"])
        if user.role != required_role and user.role != Role.ADMIN:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return {"user_id": token["user_id"], "role": user.role}
    return verify

# Использование
@app.delete("/api/v1/users/{user_id}")
async def delete_user(user_id: str, auth: Dict = Depends(check_role(Role.ADMIN))):
    # Только админ может удалить пользователя
    await db.delete_user(user_id)
    return {"ok": True}

6. OAuth 2.0 / Social Login

Делегированная авторизация через третьих сторон:

from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
google = oauth.register(
    name='google',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@app.get("/api/v1/auth/google/callback")
async def google_callback(request: Request):
    token = await google.authorize_access_token(request)
    user_info = token['userinfo']
    
    # Создаём или обновляем пользователя
    user = await db.get_or_create_user(
        email=user_info['email'],
        name=user_info['name'],
        google_id=user_info['sub']
    )
    
    # Создаём токены
    access = create_access_token(user.id)
    refresh = create_refresh_token(user.id)
    
    return {"access_token": access, "refresh_token": refresh}

Типичный workflow

1. Регистрация → Пользователь создает аккаунт
2. Логин → Проверяем пароль, выдаём access + refresh токены
3. Запросы → Клиент отправляет access в заголовке Authorization
4. Проверка → Backend проверяет токен и права доступа
5. Обновление → При истечении access используем refresh для новой пары
6. Логаут → Инвалидируем refresh token

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

• Никогда не храни пароли в открытом виде (используй bcrypt)
• Access токены — short-lived (15 мин)
• Refresh токены — long-lived (7 дней), хранятся в БД
• HTTPS — всегда, обязательно
• CSRF защита — для форм
• Rate limiting — на логин и регистрацию
• 2FA — для критичных операций