embeddevio
83d5a23406
- app.py: create_app() собирает мок-сервер и отдаёт /openapi.json из спецификации. - errors.py: ApiError + хендлеры, приводящие 4xx/5xx/422 к схеме Error. - deps.py: Bearer-токен и cookie-сессия (ADR-008). - schemas.py: pydantic-модели запросов, зеркалящие схемы контракта (extra=forbid, writeOnly для секретов).
54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
"""FastAPI-зависимости: доступ к состоянию и проверка токена доступа."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, Request
|
|
|
|
from awg_api.errors import ApiError
|
|
from awg_api.state import MockState
|
|
|
|
SESSION_COOKIE_NAME = "awg_session"
|
|
|
|
|
|
def get_state(request: Request) -> MockState:
|
|
"""In-memory состояние мок-сервера (до появления SQLite-хранилища, задача #20)."""
|
|
state: MockState = request.app.state.mock_state
|
|
return state
|
|
|
|
|
|
def _bearer_token(request: Request) -> str | None:
|
|
header = request.headers.get("authorization", "")
|
|
scheme, _, credentials = header.partition(" ")
|
|
if scheme.lower() == "bearer" and credentials.strip():
|
|
return credentials.strip()
|
|
return None
|
|
|
|
|
|
def token_matches(candidate: str | None, expected: str | None) -> bool:
|
|
if candidate is None or expected is None:
|
|
return False
|
|
return hmac.compare_digest(candidate, expected)
|
|
|
|
|
|
def authorize(request: Request, token: str | None = None) -> None:
|
|
"""Проверяет Bearer-токен или активную cookie-сессию; иначе — 401."""
|
|
expected: str = request.app.state.api_token
|
|
if token_matches(token, expected) or token_matches(_bearer_token(request), expected):
|
|
return
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if session_id and token_matches(session_id, request.app.state.mock_state.session_token):
|
|
return
|
|
raise ApiError(401, "unauthorized", "Требуется токен доступа")
|
|
|
|
|
|
async def require_token(request: Request) -> None:
|
|
"""Зависимость для защищённых роутеров."""
|
|
authorize(request)
|
|
|
|
|
|
StateDep = Annotated[MockState, Depends(get_state)]
|
|
TokenDep = Annotated[None, Depends(require_token)]
|