# 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": "Нет активной сессии"}