50 lines
2.0 KiB
Python
50 lines
2.0 KiB
Python
import uuid
|
|
from typing import Dict, Optional
|
|
from fastapi import Depends, HTTPException, status, Request
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
|
|
# Простое хранилище активных сессий (в реальном приложении лучше использовать Redis или другое хранилище)
|
|
ACTIVE_SESSIONS: Dict[str, str] = {}
|
|
|
|
# Схема безопасности для проверки токена
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
def create_session_token(username: str) -> str:
|
|
"""Создание простого сессионного токена"""
|
|
token = str(uuid.uuid4())
|
|
ACTIVE_SESSIONS[token] = username
|
|
return token
|
|
|
|
def remove_session_token(token: str) -> None:
|
|
"""Удаление сессионного токена"""
|
|
if token in ACTIVE_SESSIONS:
|
|
del ACTIVE_SESSIONS[token]
|
|
|
|
async def validate_session_token(
|
|
request: Request,
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
|
|
) -> Optional[str]:
|
|
"""Проверка сессионного токена"""
|
|
token_value = None
|
|
|
|
# Проверяем заголовок Authorization
|
|
if credentials:
|
|
token_value = credentials.credentials
|
|
# Или query-параметр token (для обратной совместимости)
|
|
elif 'token' in request.query_params:
|
|
token_value = request.query_params['token']
|
|
|
|
if not token_value or token_value not in ACTIVE_SESSIONS:
|
|
return None
|
|
|
|
return ACTIVE_SESSIONS[token_value]
|
|
|
|
async def require_auth(token: Optional[str] = Depends(validate_session_token)) -> str:
|
|
"""Требование обязательной аутентификации"""
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Не авторизован",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
return token |