feat: Guests API keys. Closes #3
This commit is contained in:
@@ -1,27 +1,74 @@
|
||||
import os
|
||||
import logging
|
||||
from fastapi import Depends, HTTPException, Security
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
from fastapi import Depends, HTTPException
|
||||
from fastapi.security import APIKeyHeader
|
||||
from starlette.status import HTTP_403_FORBIDDEN
|
||||
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 = os.getenv("IGNIS_API_KEY")
|
||||
if not API_KEY:
|
||||
MASTER_KEY = os.getenv("IGNIS_API_KEY")
|
||||
if not MASTER_KEY:
|
||||
logger.warning("IGNIS_API_KEY не задан -- авторизация отключена!")
|
||||
|
||||
API_KEY_NAME = "X-API-Key"
|
||||
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
||||
|
||||
|
||||
async def verify_token(header_value: str = Depends(api_key_header)):
|
||||
if not API_KEY:
|
||||
return None
|
||||
if header_value == API_KEY:
|
||||
return header_value
|
||||
raise HTTPException(
|
||||
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
|
||||
)
|
||||
@dataclass
|
||||
class AuthContext:
|
||||
"""Результат авторизации -- передаётся в роуты через Depends."""
|
||||
is_master: bool # мастер-ключ из .env
|
||||
is_admin: bool # право на CRUD групп, расписания, ресканирование
|
||||
key_name: str # имя ключа (для логов)
|
||||
|
||||
|
||||
async def verify_token(header_value: str = Depends(api_key_header)) -> AuthContext:
|
||||
"""
|
||||
Проверка API-ключа:
|
||||
1. Если IGNIS_API_KEY не задан -- авторизация отключена, полный доступ
|
||||
2. Мастер-ключ из .env -- полный доступ
|
||||
3. Ключ из БД (api_keys) -- проверяем active и is_admin
|
||||
4. Иначе -- 403
|
||||
"""
|
||||
# Авторизация отключена
|
||||
if not MASTER_KEY:
|
||||
return AuthContext(is_master=True, is_admin=True, key_name="no-auth")
|
||||
|
||||
if not header_value:
|
||||
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="API-ключ не передан")
|
||||
|
||||
# Мастер-ключ
|
||||
if 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
|
||||
|
||||
Reference in New Issue
Block a user