92 lines
4.3 KiB
Python
92 lines
4.3 KiB
Python
# app/api/auth.py
|
|
from fastapi import APIRouter, Header, Query, HTTPException, Depends
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
from pydantic import BaseModel
|
|
from app.core.config import get_settings
|
|
from app.core.security import create_session_token, validate_session_token, remove_session_token, ACTIVE_SESSIONS
|
|
from typing import Optional
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
security = HTTPBasic()
|
|
|
|
class LoginResponse(BaseModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
|
|
class UserResponse(BaseModel):
|
|
username: str
|
|
is_authenticated: bool
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(login_data: LoginRequest):
|
|
"""Базовая авторизация по логину и паролю"""
|
|
settings = get_settings()
|
|
|
|
if login_data.username != settings.auth.username or login_data.password != settings.auth.password:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Неверные учетные данные",
|
|
headers={"WWW-Authenticate": "Basic"},
|
|
)
|
|
|
|
# Создаем сессионный токен (без JWT, просто уникальный идентификатор)
|
|
token = create_session_token(login_data.username)
|
|
|
|
return LoginResponse(access_token=token)
|
|
@router.get("/me", response_model=UserResponse)
|
|
async def get_current_user(
|
|
authorization: Optional[str] = Header(None), # ← Указываем что это Header
|
|
token_param: Optional[str] = Query(None, alias="token") # ← Указываем что это Query
|
|
):
|
|
"""Получение информации о текущем пользователе"""
|
|
token_value = None
|
|
|
|
# Проверяем заголовок Authorization
|
|
if authorization and authorization.startswith("Bearer "):
|
|
token_value = authorization.replace("Bearer ", "")
|
|
print(f"✓ Токен из заголовка: {token_value}")
|
|
# Если токен не найден в заголовке, проверяем параметр запроса
|
|
elif token_param:
|
|
token_value = token_param
|
|
print(f"✓ Токен из query параметра: {token_value}")
|
|
# Для обратной совместимости проверяем, если authorization передан как параметр запроса
|
|
elif authorization and not authorization.startswith("Bearer "):
|
|
token_value = authorization
|
|
print(f"✓ Токен из authorization (без Bearer): {token_value}")
|
|
|
|
print(f"Проверка токена: {token_value}")
|
|
print(f"Активные сессии: {list(ACTIVE_SESSIONS.keys())}")
|
|
|
|
# Проверяем токен в активных сессиях
|
|
if token_value and token_value in ACTIVE_SESSIONS:
|
|
username = ACTIVE_SESSIONS[token_value]
|
|
print(f"✓ Токен найден! Пользователь: {username}")
|
|
return UserResponse(username=username, is_authenticated=True)
|
|
|
|
print(f"✗ Токен не найден в активных сессиях")
|
|
return UserResponse(username="", is_authenticated=False)
|
|
|
|
@router.post("/logout")
|
|
async def logout(authorization: Optional[str] = None, token_param: Optional[str] = None):
|
|
"""Выход из системы"""
|
|
token_value = None
|
|
|
|
# Проверяем заголовок Authorization
|
|
if authorization and authorization.startswith("Bearer "):
|
|
token_value = authorization.replace("Bearer ", "")
|
|
# Если токен не найден в заголовке, проверяем параметр запроса
|
|
elif token_param:
|
|
token_value = token_param
|
|
# Для обратной совместимости проверяем, если authorization передан как параметр запроса
|
|
elif authorization and not authorization.startswith("Bearer "):
|
|
token_value = authorization
|
|
|
|
if token_value and token_value in ACTIVE_SESSIONS:
|
|
remove_session_token(token_value)
|
|
return {"status": "success", "message": "Выход выполнен успешно"}
|
|
|
|
return {"status": "success", "message": "Нет активной сессии"} |