95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
import os
|
|
import hmac
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import APIKeyHeader
|
|
from starlette.status import HTTP_403_FORBIDDEN, HTTP_503_SERVICE_UNAVAILABLE
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import select
|
|
|
|
from app.core.database import async_session
|
|
from app.models.api_key import ApiKeyModel
|
|
|
|
load_dotenv()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
API_KEY_NAME = "X-API-Key"
|
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
|
|
|
|
|
@dataclass
|
|
class AuthContext:
|
|
"""Результат авторизации -- передаётся в роуты через Depends."""
|
|
|
|
is_master: bool # мастер-ключ из .env
|
|
is_admin: bool # право на CRUD групп, расписания, ресканирование
|
|
key_name: str # имя ключа (для логов)
|
|
|
|
|
|
def get_master_key() -> str | None:
|
|
value = os.getenv("IGNIS_API_KEY", "").strip()
|
|
return value or None
|
|
|
|
|
|
async def verify_token(header_value: str = Depends(api_key_header)) -> AuthContext:
|
|
"""
|
|
Проверка API-ключа:
|
|
1. IGNIS_API_KEY должен быть задан, иначе сервер закрыт (fail-closed)
|
|
2. Мастер-ключ из .env -- полный доступ
|
|
3. Ключ из БД (api_keys) -- проверяем active и is_admin
|
|
4. Иначе -- 403
|
|
"""
|
|
master_key = get_master_key()
|
|
if not master_key:
|
|
logger.error("IGNIS_API_KEY не задан: защищённые API закрыты до настройки")
|
|
raise HTTPException(
|
|
status_code=HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Сервер не настроен: задайте IGNIS_API_KEY",
|
|
)
|
|
|
|
if not header_value:
|
|
raise HTTPException(
|
|
status_code=HTTP_403_FORBIDDEN, detail="API-ключ не передан"
|
|
)
|
|
|
|
# Мастер-ключ (timing-safe сравнение)
|
|
if hmac.compare_digest(header_value, master_key):
|
|
return AuthContext(is_master=True, is_admin=True, key_name="master")
|
|
|
|
# Ищем в БД
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(ApiKeyModel).where(ApiKeyModel.key == header_value)
|
|
)
|
|
api_key = result.scalar_one_or_none()
|
|
|
|
if api_key and api_key.active:
|
|
return AuthContext(
|
|
is_master=False,
|
|
is_admin=api_key.is_admin,
|
|
key_name=api_key.name,
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=HTTP_403_FORBIDDEN, detail="Неверный или деактивированный ключ"
|
|
)
|
|
|
|
|
|
def require_admin(auth: AuthContext = Depends(verify_token)) -> AuthContext:
|
|
"""Dependency для роутов, требующих админских прав."""
|
|
if not auth.is_admin:
|
|
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Недостаточно прав")
|
|
return auth
|
|
|
|
|
|
def require_master(auth: AuthContext = Depends(verify_token)) -> AuthContext:
|
|
"""Dependency для роутов, доступных только мастер-ключу."""
|
|
if not auth.is_master:
|
|
raise HTTPException(
|
|
status_code=HTTP_403_FORBIDDEN,
|
|
detail="Требуется мастер-ключ",
|
|
)
|
|
return auth
|