Merge pull request 'Sprint 0 / #19: OpenAPI 3.1 контракт API v1 + TS-клиент и контрактные тесты' (#57) from feat/19-openapi into main
contract-openapi / openapi-contract (push) Failing after 0s
contract-openapi / ts-client-is-current (push) Failing after 0s

Sprint 0 / #19: OpenAPI 3.1 контракт API v1 + TS-клиент и контрактные тесты

AC проверены независимо (ревью TechLead): генератор TS-клиента (blob сверен),
отклонение схемы роняет контрактный тест (мутации), секретов в ответах нет.
Гейты: ruff/mypy чисто, pytest 93 passed / coverage 95.48% (порог 70%).
CI green недостижим — runner отсутствует (#52); отклонение зафиксировано в PR.
Refs #19, #15, #52.
This commit was merged in pull request #57.
This commit is contained in:
2026-10-28 21:21:13 +03:00
25 changed files with 4534 additions and 0 deletions
+47
View File
@@ -0,0 +1,47 @@
"""Сборка FastAPI-приложения мок-контрол-плейна (контракт ``docs/openapi.yaml``)."""
from __future__ import annotations
import os
from typing import Any
from fastapi import FastAPI
from awg_api import __version__
from awg_api.contract import load_spec
from awg_api.errors import install_error_handlers
from awg_api.routers import auth, config, devices, ops, routing, status
from awg_api.state import DEV_TOKEN, MockState
from awg_api.ws import metrics
TOKEN_ENV_VAR = "AWG_API_TOKEN" # noqa: S105 — имя переменной окружения, не секрет
class ContractFastAPI(FastAPI):
"""FastAPI, отдающий ``/openapi.json`` из контрактной спецификации, не из кода."""
def openapi(self) -> dict[str, Any]:
return load_spec()
def create_app(state: MockState | None = None, token: str | None = None) -> FastAPI:
"""Создаёт приложение мок-сервера; ``state``/``token`` подменяются в тестах."""
resolved_token = token or os.environ.get(TOKEN_ENV_VAR) or DEV_TOKEN
mock_state = state if state is not None else MockState()
mock_state.token = resolved_token
app = ContractFastAPI(
title="awg-dash API",
version=__version__,
description="Мок-контрол-плейн по контракту docs/openapi.yaml (задача #19)",
)
app.state.mock_state = mock_state
app.state.api_token = resolved_token
install_error_handlers(app)
app.include_router(ops.router)
app.include_router(auth.router)
app.include_router(status.router)
app.include_router(config.router)
app.include_router(routing.router)
app.include_router(devices.router)
app.include_router(metrics.router)
return app
+242
View File
@@ -0,0 +1,242 @@
"""Загрузка OpenAPI-контракта и валидация ответов реализации против схем.
Контракт (``docs/openapi.yaml``) — источник истины: тесты
``backend/tests/test_contract_responses.py`` прогоняют каждый эндпоинт реализации
и сверяют ответ против схемы спецификации. Любое расхождение — ``ContractViolationError``,
то есть красный CI.
"""
from __future__ import annotations
import copy
import os
from functools import lru_cache
from pathlib import Path
from typing import Any
import yaml
from jsonschema import Draft202012Validator, FormatChecker
SPEC_RELATIVE_PATH = Path("docs") / "openapi.yaml"
SPEC_ENV_VAR = "AWG_OPENAPI_SPEC"
JSON_MEDIA_TYPE = "application/json"
HTTP_METHODS: tuple[str, ...] = ("get", "put", "post", "delete", "patch", "head", "options")
DOC_ROUTE_PATHS: frozenset[str] = frozenset(
{"/openapi.json", "/docs", "/redoc", "/docs/oauth2-redirect"}
)
SECRET_FIELD_NAMES: frozenset[str] = frozenset({"private_key", "preshared_key", "password"})
class ContractViolationError(AssertionError):
"""Ответ реализации не соответствует схеме контракта."""
def spec_path() -> Path:
"""Путь до спецификации (``AWG_OPENAPI_SPEC`` переопределяет поиск от корня репозитория)."""
override = os.environ.get(SPEC_ENV_VAR)
if override:
return Path(override)
for parent in Path(__file__).resolve().parents:
candidate = parent / SPEC_RELATIVE_PATH
if candidate.is_file():
return candidate
raise FileNotFoundError(f"спецификация {SPEC_RELATIVE_PATH} не найдена")
@lru_cache(maxsize=1)
def load_spec() -> dict[str, Any]:
"""Разобранная спецификация OpenAPI (кэшируется на процесс)."""
raw = yaml.safe_load(spec_path().read_text(encoding="utf-8"))
if not isinstance(raw, dict):
raise ContractViolationError("спецификация должна быть YAML-объектом на верхнем уровне")
return raw
def iter_operations(spec: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]:
"""Перечень описанных в контракте операций как (путь, метод, операция)."""
operations: list[tuple[str, str, dict[str, Any]]] = []
for path, item in spec.get("paths", {}).items():
if not isinstance(item, dict):
continue
for method in HTTP_METHODS:
operation = item.get(method)
if isinstance(operation, dict):
operations.append((path, method, operation))
return operations
def _resolve_ref(spec: dict[str, Any], ref: str) -> Any:
node: Any = spec
for part in ref.lstrip("#/").split("/"):
node = node[part]
return node
def _strip_write_only(node: Any) -> None:
"""Удаляет ``writeOnly``-свойства (секреты) из схемы ответа.
``writeOnly`` в JSON Schema 2020-12 означает «только в запросе», поэтому в ответе
такие поля запрещены: они вырезаются, включая упоминание в ``required``; при этом
``additionalProperties: false`` делает их присутствие ошибкой валидации.
"""
if isinstance(node, dict):
properties = node.get("properties")
if isinstance(properties, dict):
dropped = [
name
for name, sub in properties.items()
if isinstance(sub, dict) and sub.get("writeOnly") is True
]
for name in dropped:
del properties[name]
required = node.get("required")
if dropped and isinstance(required, list):
node["required"] = [name for name in required if name not in dropped]
for value in node.values():
_strip_write_only(value)
elif isinstance(node, list):
for item in node:
_strip_write_only(item)
def _document_for(schema: dict[str, Any], spec: dict[str, Any]) -> dict[str, Any]:
document = copy.deepcopy(schema)
document["components"] = copy.deepcopy(spec.get("components", {}))
_strip_write_only(document)
return document
def validator_for(schema: dict[str, Any], spec: dict[str, Any]) -> Draft202012Validator:
"""Валидатор фрагмента спецификации: внутренние ссылки ``#/components`` разрешаются."""
return Draft202012Validator(_document_for(schema, spec), format_checker=FormatChecker())
def response_schema(
spec: dict[str, Any], path: str, method: str, status: int | str
) -> dict[str, Any] | None:
"""Схема тела ответа для (path, method, status) или ``None``, если ответ не описан."""
operation = spec.get("paths", {}).get(path, {}).get(method)
if not isinstance(operation, dict):
return None
response = operation.get("responses", {}).get(str(status))
if isinstance(response, dict) and "$ref" in response:
response = _resolve_ref(spec, str(response["$ref"]))
if not isinstance(response, dict):
return None
media = response.get("content", {}).get(JSON_MEDIA_TYPE)
if not isinstance(media, dict):
# Non-JSON ответы (например, /metrics → text/plain): берём единственный медиатип.
content = response.get("content", {})
media = next((item for item in content.values() if isinstance(item, dict)), None)
if not isinstance(media, dict):
return None
schema = media.get("schema")
return schema if isinstance(schema, dict) else None
def validate_response(
spec: dict[str, Any],
path: str,
method: str,
status: int,
payload: Any,
) -> None:
"""Проверяет тело ответа против схемы контракта; поднимает ``ContractViolationError``."""
schema = response_schema(spec, path, method, status)
if schema is None:
raise ContractViolationError(
f"{method.upper()} {path} -> {status}: ответ не описан в спецификации"
)
errors = sorted(
validator_for(schema, spec).iter_errors(payload), key=lambda err: list(err.absolute_path)
)
if errors:
lines = [f"{method.upper()} {path} -> {status}: несоответствий схеме: {len(errors)}"]
for err in errors[:10]:
location = "/".join(str(part) for part in err.absolute_path) or "<root>"
lines.append(f" - {location}: {err.message}")
raise ContractViolationError("\n".join(lines))
def error_codes(spec: dict[str, Any]) -> set[str]:
"""Множество допустимых кодов ошибок из схемы ``ErrorCode``."""
enum = spec.get("components", {}).get("schemas", {}).get("ErrorCode", {}).get("enum", [])
return {str(item) for item in enum}
def write_only_fields(node: Any) -> set[str]:
"""Имена всех свойств, помеченных ``writeOnly`` (в любом месте документа)."""
found: set[str] = set()
if isinstance(node, dict):
for name, sub in node.get("properties", {}).items():
if isinstance(sub, dict) and sub.get("writeOnly") is True:
found.add(str(name))
found |= write_only_fields(sub)
for key, value in node.items():
if key != "properties":
found |= write_only_fields(value)
elif isinstance(node, list):
for item in node:
found |= write_only_fields(item)
return found
def iter_route_operations(app: Any) -> set[tuple[str, str]]:
"""Множество ``(путь, МЕТОД)`` для всех HTTP/WebSocket-маршрутов приложения.
Обходит вложенные роутеры: FastAPI 0.159 не разворачивает ``include_router``
в ``app.routes``: хранит обёртки, содержащие ``original_router``.
"""
found: set[tuple[str, str]] = set()
def walk(router: Any, seen: set[int]) -> None:
if id(router) in seen:
return
seen.add(id(router))
for route in getattr(router, "routes", []):
path = getattr(route, "path", None)
if isinstance(path, str):
if path in DOC_ROUTE_PATHS:
continue
methods = getattr(route, "methods", None)
if methods is None:
found.add((path, "GET")) # WebSocketRoute
continue
for method in methods:
if method.upper() not in {"HEAD", "OPTIONS"}:
found.add((path, method.upper()))
continue
inner = (
getattr(route, "original_router", None)
or getattr(route, "router", None)
or getattr(route, "app", None)
)
if inner is not None:
walk(inner, seen)
walk(app, set())
return found
def assert_routes_match(spec: dict[str, Any], app: Any) -> None:
"""Двусторонняя сверка маршрутов приложения и контракта.
* каждая операция контракта обязана существовать в приложении;
* каждый HTTP/WebSocket-маршрут приложения обязан быть описан в контракте.
"""
declared: set[tuple[str, str]] = {
(path, method.upper()) for path, method, _ in iter_operations(spec)
}
implemented = iter_route_operations(app)
missing = sorted(declared - implemented)
undocumented = sorted(implemented - declared)
problems: list[str] = []
if missing:
problems.append("нет в реализации: " + ", ".join(f"{m} {p}" for p, m in missing))
if undocumented:
problems.append("нет в контракте: " + ", ".join(f"{m} {p}" for p, m in undocumented))
if problems:
raise ContractViolationError(
"маршруты приложения и docs/openapi.yaml расходятся:\n " + "\n ".join(problems)
)
+53
View File
@@ -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)]
+88
View File
@@ -0,0 +1,88 @@
"""Единый формат ошибок API: ``{code, message, details}`` (схема Error из контракта)."""
from __future__ import annotations
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
STATUS_TO_CODE: dict[int, str] = {
400: "bad_request",
401: "unauthorized",
403: "forbidden",
404: "not_found",
409: "conflict",
422: "validation_failed",
429: "rate_limited",
500: "internal_error",
503: "agent_unavailable",
}
class ApiError(Exception):
"""Ошибка API, сериализуемая в схему Error."""
def __init__(
self,
status_code: int,
code: str,
message: str,
details: dict[str, Any] | None = None,
) -> None:
super().__init__(message)
self.status_code = status_code
self.code = code
self.message = message
self.details = details
def error_payload(code: str, message: str, details: dict[str, Any] | None = None) -> dict[str, Any]:
"""Тело ошибки в контрактном виде (``details`` всегда присутствует, может быть null)."""
return {"code": code, "message": message, "details": details}
def install_error_handlers(app: FastAPI) -> None:
"""Приводит все ошибки (включая 422 и 500) к схеме Error."""
@app.exception_handler(ApiError)
async def _api_error(_: Request, exc: ApiError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content=error_payload(exc.code, exc.message, exc.details),
)
@app.exception_handler(RequestValidationError)
async def _validation_error(_: Request, exc: RequestValidationError) -> JSONResponse:
issues = [
{
"path": "/" + "/".join(str(part) for part in error.get("loc", ())),
"code": str(error.get("type", "invalid")),
"message": str(error.get("msg", "invalid value")),
}
for error in exc.errors()
]
return JSONResponse(
status_code=422,
content=error_payload(
"validation_failed",
"Тело запроса не соответствует схеме",
{"issues": issues},
),
)
@app.exception_handler(StarletteHTTPException)
async def _http_error(_: Request, exc: StarletteHTTPException) -> JSONResponse:
code = STATUS_TO_CODE.get(exc.status_code, "internal_error")
return JSONResponse(
status_code=exc.status_code, content=error_payload(code, str(exc.detail), None)
)
@app.exception_handler(Exception)
async def _unhandled(_: Request, __: Exception) -> JSONResponse:
return JSONResponse(
status_code=500,
content=error_payload("internal_error", "Внутренняя ошибка", None),
)
+1
View File
@@ -0,0 +1 @@
"""Роутеры API v1."""
+34
View File
@@ -0,0 +1,34 @@
"""Аутентификация UI и API: /api/v1/auth/*."""
from __future__ import annotations
from typing import Annotated, Any
from fastapi import APIRouter, Depends, Response
from awg_api.deps import SESSION_COOKIE_NAME, StateDep, require_token
from awg_api.schemas import AuthLoginRequest
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
TokenDep = Annotated[None, Depends(require_token)]
@router.post("/login")
async def login(payload: AuthLoginRequest, state: StateDep, response: Response) -> dict[str, Any]:
session = state.login(payload.username, payload.password)
csrf_token = session["csrf_token"]
if isinstance(csrf_token, str):
response.set_cookie(
SESSION_COOKIE_NAME, csrf_token, httponly=True, samesite="strict", path="/"
)
return session
@router.post("/logout")
async def logout(state: StateDep, _: TokenDep = None) -> dict[str, Any]:
return state.logout()
@router.get("/session")
async def get_session(state: StateDep, _: TokenDep = None) -> dict[str, Any]:
return state.get_session()
+27
View File
@@ -0,0 +1,27 @@
"""Конфигурация AmneziaWG: чтение, атомарное применение, сухая валидация."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends
from awg_api.deps import StateDep, require_token
from awg_api.schemas import ConfigUpdateRequest
router = APIRouter(prefix="/api/v1/config", tags=["config"], dependencies=[Depends(require_token)])
@router.get("")
async def get_config(state: StateDep) -> dict[str, Any]:
return state.get_config()
@router.put("")
async def put_config(payload: ConfigUpdateRequest, state: StateDep) -> dict[str, Any]:
return state.apply_config(payload)
@router.post("/validate")
async def validate_config(payload: ConfigUpdateRequest, state: StateDep) -> dict[str, Any]:
return state.validate_config(payload)
+29
View File
@@ -0,0 +1,29 @@
"""Устройства и журнал: /api/v1/devices, /api/v1/logs."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, Query
from awg_api.deps import StateDep, require_token
router = APIRouter(
prefix="/api/v1", tags=["devices", "logs"], dependencies=[Depends(require_token)]
)
@router.get("/devices")
async def list_devices(state: StateDep) -> dict[str, Any]:
return state.list_devices()
@router.get("/logs")
async def list_logs(
state: StateDep,
limit: int = Query(default=100, ge=1, le=500),
cursor: str | None = None,
level: str | None = None,
q: str | None = None,
) -> dict[str, Any]:
return state.list_logs(limit=limit, cursor=cursor, level=level, query=q)
+36
View File
@@ -0,0 +1,36 @@
"""Служебные эндпоинты: /api/v1/health, /api/v1/ready, /metrics (без аутентификации)."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter
from fastapi.responses import JSONResponse, PlainTextResponse
from awg_api.deps import StateDep
from awg_api.errors import error_payload
router = APIRouter(tags=["ops"])
@router.get("/api/v1/health")
async def get_health(state: StateDep) -> dict[str, Any]:
return state.get_health()
@router.get("/api/v1/ready")
async def get_ready(state: StateDep) -> JSONResponse:
if state.ready:
return JSONResponse(status_code=200, content=state.get_ready())
checks = state.get_ready()["checks"]
return JSONResponse(
status_code=503,
content=error_payload(
"agent_unavailable", "Шлюз не готов обслуживать запросы", {"checks": checks}
),
)
@router.get("/metrics", response_class=PlainTextResponse)
async def get_prometheus_metrics(state: StateDep) -> PlainTextResponse:
return PlainTextResponse(state.prometheus_text())
+67
View File
@@ -0,0 +1,67 @@
"""Маршрутизация: правила, режим, GeoIP, импорт."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends
from awg_api.deps import StateDep, require_token
from awg_api.schemas import (
GeoIpUpdateRequest,
ImportRequest,
ModeUpdateRequest,
RoutingRuleCreate,
RoutingRuleUpdate,
)
router = APIRouter(
prefix="/api/v1/routing", tags=["routing"], dependencies=[Depends(require_token)]
)
@router.get("/rules")
async def list_rules(
state: StateDep,
enabled: bool | None = None,
kind: str | None = None,
) -> dict[str, Any]:
return state.list_rules(enabled=enabled, kind=kind)
@router.post("/rules", status_code=201)
async def create_rule(payload: RoutingRuleCreate, state: StateDep) -> dict[str, Any]:
return state.create_rule(payload)
@router.get("/rules/{rule_id}")
async def get_rule(rule_id: str, state: StateDep) -> dict[str, Any]:
return state.get_rule(rule_id)
@router.put("/rules/{rule_id}")
async def update_rule(rule_id: str, payload: RoutingRuleUpdate, state: StateDep) -> dict[str, Any]:
return state.update_rule(rule_id, payload)
@router.delete("/rules/{rule_id}", status_code=204)
async def delete_rule(rule_id: str, state: StateDep) -> None:
state.delete_rule(rule_id)
@router.put("/mode")
async def put_mode(payload: ModeUpdateRequest, state: StateDep) -> dict[str, Any]:
return state.set_mode(payload.mode)
@router.post("/geoip/update")
async def update_geoip(
state: StateDep, payload: GeoIpUpdateRequest | None = None
) -> dict[str, Any]:
force = bool(payload.force) if payload is not None else False
return state.update_geoip(force=force)
@router.post("/import")
async def import_config(payload: ImportRequest, state: StateDep) -> dict[str, Any]:
return state.import_config(payload)
+16
View File
@@ -0,0 +1,16 @@
"""Состояние туннеля: /api/v1/status."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends
from awg_api.deps import StateDep, require_token
router = APIRouter(prefix="/api/v1", tags=["status"], dependencies=[Depends(require_token)])
@router.get("/status")
async def get_status(state: StateDep) -> dict[str, Any]:
return state.get_status()
+120
View File
@@ -0,0 +1,120 @@
"""Pydantic-модели запросов, зеркалящие схемы ``docs/openapi.yaml``.
Соответствие проверяется тестом ``test_request_models_match_spec``: наборы полей,
обязательность и базовые типы моделей обязаны соответствовать схемам контракта.
"""
from __future__ import annotations
from typing import Literal, Self
from pydantic import BaseModel, ConfigDict, Field, model_validator
RoutingMode = Literal["geoip", "dns", "all-vpn", "direct-only"]
RuleKind = Literal["domain", "cidr", "geoip", "asn"]
RuleAction = Literal["vpn", "direct"]
Ipv6Policy = Literal["direct", "tunnel", "block"]
DnsMode = Literal["internal", "external"]
GeoIpSource = Literal["ripe-stat", "cache", "none"]
ImportFormat = Literal["amneziawg-conf", "wireguard-conf", "json"]
class StrictModel(BaseModel):
"""Модель запроса: лишние поля запрещены (``additionalProperties: false`` в контракте)."""
model_config = ConfigDict(extra="forbid")
class GeoIpStatusIn(StrictModel):
source: GeoIpSource
prefix_count: int = Field(ge=0)
updated_at: str | None
class TunnelConfigIn(StrictModel):
address: str
listen_port: int = Field(ge=1, le=65535)
mtu: int = Field(ge=576, le=1500)
private_key: str = Field(json_schema_extra={"writeOnly": True})
dns: list[str] | None = None
class PeerConfigIn(StrictModel):
public_key: str
preshared_key: str | None = Field(default=None, json_schema_extra={"writeOnly": True})
endpoint: str
allowed_ips: list[str]
persistent_keepalive: int | None = Field(default=None, ge=0, le=65535)
class DnsConfigIn(StrictModel):
mode: DnsMode
servers: list[str]
class RoutingConfigIn(StrictModel):
mode: RoutingMode
kill_switch: bool
mtu_clamping: bool
ipv6: Ipv6Policy
geoip: GeoIpStatusIn
class ConfigUpdateRequest(StrictModel):
"""Частичное обновление конфигурации (merge в текущее состояние)."""
tunnel: TunnelConfigIn | None = None
peers: list[PeerConfigIn] | None = None
dns: DnsConfigIn | None = None
routing: RoutingConfigIn | None = None
@model_validator(mode="after")
def _require_any_field(self) -> Self:
if not self.model_fields_set:
raise ValueError("нужно передать хотя бы одно поле конфигурации")
return self
class RoutingRuleCreate(StrictModel):
name: str = Field(min_length=1, max_length=128)
kind: RuleKind
value: str = Field(min_length=1)
action: RuleAction
enabled: bool = True
priority: int = Field(default=100, ge=0, le=10000)
comment: str | None = None
class RoutingRuleUpdate(StrictModel):
name: str | None = Field(default=None, min_length=1, max_length=128)
kind: RuleKind | None = None
value: str | None = Field(default=None, min_length=1)
action: RuleAction | None = None
enabled: bool | None = None
priority: int | None = Field(default=None, ge=0, le=10000)
comment: str | None = None
@model_validator(mode="after")
def _require_any_field(self) -> Self:
if not self.model_fields_set:
raise ValueError("нужно передать хотя бы одно поле правила")
return self
class ModeUpdateRequest(StrictModel):
mode: RoutingMode
class GeoIpUpdateRequest(StrictModel):
force: bool = False
class ImportRequest(StrictModel):
format: ImportFormat
content: str = Field(min_length=1)
apply: bool = False
class AuthLoginRequest(StrictModel):
username: str = Field(min_length=1)
password: str = Field(min_length=1, json_schema_extra={"writeOnly": True})
+471
View File
@@ -0,0 +1,471 @@
"""In-memory состояние мок-сервера (замена агента и SQLite до Sprint 1).
Секреты (private key, preshared key, пароль UI) хранятся отдельно от словарей-ответов
и никогда в них не попадают — это проверяется тестом «в ответах нет приватных ключей».
"""
from __future__ import annotations
import time
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any
from awg_api.errors import ApiError
from awg_api.schemas import (
ConfigUpdateRequest,
ImportRequest,
RoutingRuleCreate,
RoutingRuleUpdate,
)
VERSION = "0.1.0-mock"
GEOIP_PREFIX_COUNT = 11412
DEV_TOKEN = "awg-dash-dev-token" # noqa: S105 — документированный dev-дефолт мок-сервера
DEV_UI_PASSWORD = "awg-dash-dev-password" # noqa: S105 — документированный dev-дефолт
PRIVATE_KEY_SENTINEL = "PRIVATE-KEY-MUST-NOT-LEAK"
PRESHARED_KEY_SENTINEL = "PRESHARED-KEY-MUST-NOT-LEAK"
LOG_LEVELS = ("debug", "info", "warning", "error")
def _now() -> str:
"""RFC 3339 (UTC), суффикс Z — формат всех таймстемпов контракта."""
return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
class MockState:
"""Состояние мок-контрол-плейна: туннель, правила, устройства, журнал, сессия."""
def __init__(self, *, ready: bool = True) -> None:
self._started = time.monotonic()
self.ready = ready
self.token = DEV_TOKEN
self.ui_password = DEV_UI_PASSWORD
self.session_token: str | None = None
self.username: str | None = None
self.expires_at: str | None = None
self.mode: str = "geoip"
self.kill_switch = True
self.dns_leak_protection = True
self.mtu_clamping = True
self.ipv6: str = "direct"
self.geoip: dict[str, Any] = {
"source": "ripe-stat",
"prefix_count": GEOIP_PREFIX_COUNT,
"updated_at": _now(),
}
self.tunnel_state: str = "up"
self.handshake_age_s: int | None = 27
self.rx_bytes = 5_812_664
self.tx_bytes = 1_204_331
self.peer_endpoint = "vpn.example.net:51820"
self.secrets: dict[str, str] = {
"private_key": PRIVATE_KEY_SENTINEL,
"preshared_key": PRESHARED_KEY_SENTINEL,
}
self.tunnel_config: dict[str, Any] = {
"address": "10.8.0.2/32",
"listen_port": 51820,
"mtu": 1420,
"dns": ["10.8.0.1"],
}
self.peers: list[dict[str, Any]] = [
{
"public_key": "kZ9vT4publicKeyExampleOnly0000000000000000000=",
"endpoint": "vpn.example.net:51820",
"allowed_ips": ["0.0.0.0/0", "::/0"],
"persistent_keepalive": 25,
}
]
self.dns: dict[str, Any] = {"mode": "internal", "servers": ["10.8.0.1", "1.1.1.1"]}
self.rules: dict[str, dict[str, Any]] = {}
self.devices: list[dict[str, Any]] = [
{
"name": "laptop-anna",
"ip": "10.8.0.11",
"mac": "3c:22:fb:11:22:33",
"vpn": True,
"ipsets": ["awg_geoip", "awg_domains"],
"last_seen_at": _now(),
},
{
"name": "phone-max",
"ip": "10.8.0.12",
"mac": None,
"vpn": False,
"ipsets": [],
"last_seen_at": _now(),
},
]
self.logs: list[dict[str, Any]] = []
self.updated_at = _now()
self._seed_rules()
self._seed_logs()
# ------------------------------------------------------------------ служебное
def uptime_s(self) -> int:
return max(0, int(time.monotonic() - self._started))
def _seed_rules(self) -> None:
self.rules["0192a1b2-0000-7000-8000-000000000001"] = {
"id": "0192a1b2-0000-7000-8000-000000000001",
"name": "Маршрутизировать домены сервисов",
"kind": "domain",
"value": "*.example.net",
"action": "vpn",
"enabled": True,
"priority": 100,
"comment": None,
"created_at": _now(),
"updated_at": _now(),
}
self.rules["0192a1b2-0000-7000-8000-000000000002"] = {
"id": "0192a1b2-0000-7000-8000-000000000002",
"name": "Локальная подсеть напрямую",
"kind": "cidr",
"value": "192.168.1.0/24",
"action": "direct",
"enabled": True,
"priority": 10,
"comment": "LAN исключаем из туннеля",
"created_at": _now(),
"updated_at": _now(),
}
def _seed_logs(self) -> None:
seeds = [
(
"info",
"awg_agent.reconcile",
"Применён desired state: 2 правила, 1 пир",
{"mode": "geoip"},
),
(
"info",
"awg_api.routing",
"Режим split-tunnel: geoip",
{"prefix_count": GEOIP_PREFIX_COUNT},
),
("warning", "awg_agent.geoip", "Снапшот GeoIP старше 24 часов", {"age_h": 26}),
("error", "awg_agent.tunnel", "Краткий разрыв handshake", {"duration_s": 4}),
("debug", "awg_api.auth", "Создана сессия UI", {"username": "admin"}),
]
for level, logger, message, fields in seeds:
self.logs.append(
{
"ts": _now(),
"level": level,
"logger": logger,
"message": message,
"fields": fields,
}
)
# ------------------------------------------------------------------ ops
def get_health(self) -> dict[str, Any]:
return {
"status": "ok" if self.ready else "degraded",
"version": VERSION,
"uptime_s": self.uptime_s(),
"checks": [
{"name": "api", "ok": True, "detail": None},
{
"name": "agent_socket",
"ok": self.ready,
"detail": None if self.ready else "unix-socket не отвечает",
},
],
}
def get_ready(self) -> dict[str, Any]:
return {
"ready": self.ready,
"checks": [
{"name": "tunnel", "ok": self.ready, "detail": None},
{"name": "geoip_cache", "ok": True, "detail": None},
],
}
def prometheus_text(self) -> str:
tunnel_up = 1 if self.tunnel_state == "up" else 0
lines = [
"# HELP awg_dash_tunnel_up Tunnel state (1 = up)",
"# TYPE awg_dash_tunnel_up gauge",
f"awg_dash_tunnel_up {tunnel_up}",
"# HELP awg_dash_rx_bytes_total Received bytes on the tunnel",
"# TYPE awg_dash_rx_bytes_total counter",
f"awg_dash_rx_bytes_total {self.rx_bytes}",
"# HELP awg_dash_tx_bytes_total Transmitted bytes on the tunnel",
"# TYPE awg_dash_tx_bytes_total counter",
f"awg_dash_tx_bytes_total {self.tx_bytes}",
"# HELP awg_dash_geoip_prefixes GeoIP prefix count in the active snapshot",
"# TYPE awg_dash_geoip_prefixes gauge",
f"awg_dash_geoip_prefixes {self.geoip['prefix_count']}",
]
return "\n".join(lines) + "\n"
# ------------------------------------------------------------------ status
def get_status(self) -> dict[str, Any]:
return {
"version": VERSION,
"uptime_s": self.uptime_s(),
"mode": self.mode,
"tunnel": {
"interface": "awg0",
"state": self.tunnel_state,
"handshake_age_s": self.handshake_age_s,
"rx_bytes": self.rx_bytes,
"tx_bytes": self.tx_bytes,
"endpoint": self.peer_endpoint,
"last_handshake_at": _now(),
},
"geoip": dict(self.geoip),
"kill_switch": self.kill_switch,
"dns_leak_protection": self.dns_leak_protection,
}
def metrics_sample(self) -> dict[str, Any]:
return {
"rx_bytes": self.rx_bytes,
"tx_bytes": self.tx_bytes,
"rx_rate_bps": 1_048_576,
"tx_rate_bps": 262_144,
"handshake_age_s": self.handshake_age_s,
"active_peers": len(self.peers),
}
def metrics_frame(self) -> dict[str, Any]:
return {"type": "metrics", "ts": _now(), "data": self.metrics_sample()}
# ------------------------------------------------------------------ config
def get_config(self) -> dict[str, Any]:
"""Конфигурация без секретов: private_key/preshared_key не сериализуются."""
return {
"tunnel": dict(self.tunnel_config),
"peers": [dict(peer) for peer in self.peers],
"dns": dict(self.dns),
"routing": self._routing_config(),
"updated_at": self.updated_at,
}
def _routing_config(self) -> dict[str, Any]:
return {
"mode": self.mode,
"kill_switch": self.kill_switch,
"mtu_clamping": self.mtu_clamping,
"ipv6": self.ipv6,
"geoip": dict(self.geoip),
}
def apply_config(self, update: ConfigUpdateRequest) -> dict[str, Any]:
if update.tunnel is not None:
payload = update.tunnel.model_dump(exclude={"private_key"})
self.tunnel_config = {**self.tunnel_config, **payload}
self.secrets["private_key"] = update.tunnel.private_key
if update.peers is not None:
self.peers = [peer.model_dump(exclude={"preshared_key"}) for peer in update.peers]
sentinel = next(
(peer.preshared_key for peer in update.peers if peer.preshared_key is not None),
None,
)
if sentinel is not None:
self.secrets["preshared_key"] = sentinel
if update.dns is not None:
self.dns = update.dns.model_dump()
if update.routing is not None:
routing = update.routing.model_dump()
self.mode = routing["mode"]
self.kill_switch = routing["kill_switch"]
self.mtu_clamping = routing["mtu_clamping"]
self.ipv6 = routing["ipv6"]
self.geoip = routing["geoip"]
self.updated_at = _now()
return self.get_config()
def validate_config(self, update: ConfigUpdateRequest) -> dict[str, Any]:
errors: list[dict[str, Any]] = []
warnings: list[dict[str, Any]] = []
if update.tunnel is not None:
address = update.tunnel.address
if "/" not in address:
errors.append(
{
"path": "/tunnel/address",
"code": "invalid_cidr",
"message": "Адрес интерфейса должен быть в CIDR-нотации",
"severity": "error",
}
)
if update.tunnel.mtu > 1420:
warnings.append(
{
"path": "/tunnel/mtu",
"code": "mtu_above_recommended",
"message": "MTU выше 1420 может требовать MSS clamping",
"severity": "warning",
}
)
for index, peer in enumerate(update.peers or []):
if ":" not in peer.endpoint:
errors.append(
{
"path": f"/peers/{index}/endpoint",
"code": "invalid_endpoint",
"message": "Endpoint должен содержать host:port",
"severity": "error",
}
)
return {"valid": not errors, "errors": errors, "warnings": warnings}
# ------------------------------------------------------------------ routing
def list_rules(self, *, enabled: bool | None = None, kind: str | None = None) -> dict[str, Any]:
items = [
dict(rule)
for rule in sorted(
self.rules.values(), key=lambda rule: (rule["priority"], rule["created_at"])
)
if (enabled is None or rule["enabled"] is enabled)
and (kind is None or rule["kind"] == kind)
]
return {"items": items, "total": len(items)}
def create_rule(self, payload: RoutingRuleCreate) -> dict[str, Any]:
if any(
rule["kind"] == payload.kind and rule["value"] == payload.value
for rule in self.rules.values()
):
raise ApiError(409, "conflict", "Правило для такого kind/value уже существует")
timestamp = _now()
rule_id = str(uuid.uuid4())
rule = {
"id": rule_id,
"name": payload.name,
"kind": payload.kind,
"value": payload.value,
"action": payload.action,
"enabled": payload.enabled,
"priority": payload.priority,
"comment": payload.comment,
"created_at": timestamp,
"updated_at": timestamp,
}
self.rules[rule_id] = rule
return dict(rule)
def get_rule(self, rule_id: str) -> dict[str, Any]:
rule = self.rules.get(rule_id)
if rule is None:
raise ApiError(404, "not_found", f"Правило {rule_id} не найдено")
return dict(rule)
def update_rule(self, rule_id: str, payload: RoutingRuleUpdate) -> dict[str, Any]:
rule = self.get_rule(rule_id)
rule.update(payload.model_dump(exclude_unset=True))
rule["updated_at"] = _now()
self.rules[rule_id] = rule
return dict(rule)
def delete_rule(self, rule_id: str) -> None:
if rule_id not in self.rules:
raise ApiError(404, "not_found", f"Правило {rule_id} не найдено")
del self.rules[rule_id]
def set_mode(self, mode: str) -> dict[str, Any]:
self.mode = mode
self.updated_at = _now()
return {"mode": mode, "applied_at": _now(), "reconcile_id": str(uuid.uuid4())}
def update_geoip(self, *, force: bool = False) -> dict[str, Any]:
status = "updated" if force or self.geoip["source"] == "none" else "cached"
if status == "updated":
self.geoip = {
"source": "ripe-stat",
"prefix_count": GEOIP_PREFIX_COUNT,
"updated_at": _now(),
}
return {
"status": status,
"source": self.geoip["source"],
"prefix_count": self.geoip["prefix_count"],
"updated_at": self.geoip["updated_at"] or _now(),
"duration_ms": 1870 if status == "updated" else 3,
}
def import_config(self, payload: ImportRequest) -> dict[str, Any]:
warnings: list[str] = []
peers = 0
for line in payload.content.splitlines():
stripped = line.strip()
if stripped.startswith("[Peer]"):
peers += 1
elif stripped.startswith("PresharedKey") and payload.format != "json":
warnings.append("PresharedKey импортирован в секреты и не будет возвращён в API")
return {
"imported_peers": peers,
"imported_rules": 0,
"warnings": warnings,
}
# ------------------------------------------------------------------ devices / logs
def list_devices(self) -> dict[str, Any]:
return {"items": [dict(device) for device in self.devices], "total": len(self.devices)}
def list_logs(
self,
*,
limit: int = 100,
cursor: str | None = None,
level: str | None = None,
query: str | None = None,
) -> dict[str, Any]:
offset = 0
if cursor:
try:
offset = int(cursor)
except ValueError as exc:
raise ApiError(422, "validation_failed", "Некорректный курсор пагинации") from exc
entries = [entry for entry in self.logs if level is None or entry["level"] == level]
if query:
entries = [entry for entry in entries if query.lower() in entry["message"].lower()]
page = entries[offset : offset + limit]
next_offset = offset + len(page)
return {
"items": [dict(entry) for entry in page],
"total": len(entries),
"next_cursor": str(next_offset) if next_offset < len(entries) else None,
}
# ------------------------------------------------------------------ auth
def login(self, username: str, password: str) -> dict[str, Any]:
if password != self.ui_password:
raise ApiError(401, "unauthorized", "Неверные учётные данные")
self.username = username
self.session_token = str(uuid.uuid4())
self.expires_at = (
(datetime.now(UTC) + timedelta(hours=12))
.isoformat(timespec="seconds")
.replace("+00:00", "Z")
)
return self.get_session()
def logout(self) -> dict[str, Any]:
self.session_token = None
self.username = None
self.expires_at = None
return self.get_session()
def get_session(self) -> dict[str, Any]:
return {
"authenticated": self.session_token is not None,
"username": self.username,
"expires_at": self.expires_at,
"csrf_token": self.session_token,
}
+1
View File
@@ -0,0 +1 @@
"""WebSocket-поток метрик."""
+47
View File
@@ -0,0 +1,47 @@
"""WebSocket-поток метрик: GET /api/v1/metrics/stream."""
from __future__ import annotations
import asyncio
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from awg_api.deps import SESSION_COOKIE_NAME, token_matches
from awg_api.state import MockState
router = APIRouter(tags=["metrics"])
WS_ACCESS_DENIED = 4401
MIN_INTERVAL_MS = 250
def _authorized(websocket: WebSocket, token: str | None) -> bool:
expected: str = websocket.app.state.api_token
header = websocket.headers.get("authorization", "")
scheme, _, credentials = header.partition(" ")
if scheme.lower() == "bearer" and token_matches(credentials.strip(), expected):
return True
if token_matches(token, expected):
return True
session_id = websocket.cookies.get(SESSION_COOKIE_NAME)
state: MockState = websocket.app.state.mock_state
return token_matches(session_id, state.session_token)
@router.websocket("/api/v1/metrics/stream")
async def stream_metrics(
websocket: WebSocket, token: str | None = None, interval_ms: int = 1000
) -> None:
"""Отправляет кадры ``WsFrame``: метрики до отключения клиента."""
if not _authorized(websocket, token):
await websocket.close(code=WS_ACCESS_DENIED)
return
state: MockState = websocket.app.state.mock_state
await websocket.accept()
interval_s = max(MIN_INTERVAL_MS, interval_ms) / 1000
try:
while True:
await websocket.send_json(state.metrics_frame())
await asyncio.sleep(interval_s)
except (WebSocketDisconnect, RuntimeError):
return