feat(api): мок-контрол-плейн по контракту + единый формат ошибок (Refs #19)
- 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 для секретов).
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
"""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)]
|
||||
Reference in New Issue
Block a user