smoke 1792234257

.gitea/workflows/contract-openapi.yml

@@ -1,55 +0,0 @@
-name: contract-openapi
-# Контракт OpenAPI 3.1 (issue #19): линтеры, контрактный тест «схема ↔ реализация»
-# и проверка, что сгенерированный TS-клиент актуален относительно docs/openapi.yaml.
-on:
-  push:
-    branches:
-      - "**"
-  pull_request:
-    branches:
-      - main
-  workflow_dispatch:
-
-jobs:
-  contract:
-    name: openapi-contract
-    runs-on: ubuntu-latest
-    steps:
-      - name: checkout
-        uses: actions/checkout@v4
-
-      - name: python 3.12
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.12"
-
-      - name: install dependencies
-        run: |
-          python -m pip install --upgrade pip
-          pip install -r requirements-dev.txt -r requirements-contract.txt
-
-      - name: lint (ruff + mypy)
-        run: |
-          ruff check backend
-          ruff format --check backend
-          mypy backend
-
-      - name: contract and unit tests (coverage >= 70%)
-        run: pytest
-
-  ts-client:
-    name: ts-client-is-current
-    runs-on: ubuntu-latest
-    steps:
-      - name: checkout
-        uses: actions/checkout@v4
-
-      - name: node 20
-        uses: actions/setup-node@v4
-        with:
-          node-version: "20"
-
-      - name: regenerate client and compare
-        run: |
-          bash ui/scripts/generate-api.sh
-          git diff --exit-code -- ui/src/api/generated/schema.d.ts

SMOKE_1792234257.md

@@ -0,0 +1 @@
+SMOKE 1792234257

backend/awg_api/__init__.py

@@ -1,5 +0,0 @@
-"""FastAPI-приложение awg-dash: REST API v1, WebSocket, статика UI."""
-
-__version__ = "0.0.1"
-
-__all__ = ["__version__"]

backend/awg_api/app.py

@@ -1,47 +0,0 @@
-"""Сборка 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

backend/awg_api/contract.py

@@ -1,242 +0,0 @@
-"""Загрузка 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)
-        )

backend/awg_api/deps.py

@@ -1,53 +0,0 @@
-"""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)]

backend/awg_api/errors.py

@@ -1,88 +0,0 @@
-"""Единый формат ошибок 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),
-        )

backend/awg_api/routers/__init__.py

@@ -1 +0,0 @@
-"""Роутеры API v1."""

backend/awg_api/routers/auth.py

@@ -1,34 +0,0 @@
-"""Аутентификация 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()

backend/awg_api/routers/config.py

@@ -1,27 +0,0 @@
-"""Конфигурация 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)

backend/awg_api/routers/devices.py

@@ -1,29 +0,0 @@
-"""Устройства и журнал: /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)

backend/awg_api/routers/ops.py

@@ -1,36 +0,0 @@
-"""Служебные эндпоинты: /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())

backend/awg_api/routers/routing.py

@@ -1,67 +0,0 @@
-"""Маршрутизация: правила, режим, 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)

backend/awg_api/routers/status.py

@@ -1,16 +0,0 @@
-"""Состояние туннеля: /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()

backend/awg_api/schemas.py

@@ -1,120 +0,0 @@
-"""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})

backend/awg_api/state.py

@@ -1,471 +0,0 @@
-"""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,
-        }

backend/awg_api/ws/__init__.py

@@ -1 +0,0 @@
-"""WebSocket-поток метрик."""

backend/awg_api/ws/metrics.py

@@ -1,47 +0,0 @@
-"""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

backend/tests/conftest.py

@@ -1,48 +0,0 @@
-"""Общие фикстуры контрактных тестов."""
-
-from __future__ import annotations
-
-from collections.abc import Iterator
-from typing import Any
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-from awg_api.app import create_app
-from awg_api.contract import load_spec
-from awg_api.state import MockState
-
-TEST_TOKEN = "test-token-contract"
-
-
-@pytest.fixture(scope="session")
-def spec() -> dict[str, Any]:
-    """Спецификация контракта (docs/openapi.yaml)."""
-    return load_spec()
-
-
-@pytest.fixture
-def state() -> MockState:
-    return MockState()
-
-
-@pytest.fixture
-def app(state: MockState) -> FastAPI:
-    return create_app(state=state, token=TEST_TOKEN)
-
-
-@pytest.fixture
-def client(app: FastAPI) -> Iterator[TestClient]:
-    with TestClient(app) as test_client:
-        yield test_client
-
-
-@pytest.fixture
-def auth_headers() -> dict[str, str]:
-    return {"Authorization": f"Bearer {TEST_TOKEN}"}
-
-
-@pytest.fixture
-def token() -> str:
-    return TEST_TOKEN

backend/tests/test_contract_negative.py

@@ -1,110 +0,0 @@
-"""Негативные проверки контрактного теста: отклонение реализации обязано его ронять.
-
-Эти тесты — доказательство AC «GIVEN реализованный эндпоинт с отклонением от схемы,
-WHEN запущен контрактный тест, THEN CI падает с указанием расхождения»: они подменяют
-реализацию на заведомо расходящуюся и проверяют, что тот же самый валидатор её ловит.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-from fastapi import FastAPI, Request
-from fastapi.testclient import TestClient
-
-from awg_api.app import create_app
-from awg_api.contract import ContractViolationError, assert_routes_match, validate_response
-from awg_api.state import PRIVATE_KEY_SENTINEL, MockState
-
-
-class DeviatingStatusState(MockState):
-    """Реализация, «забывшая» контракт: неверный тип и лишнее поле."""
-
-    def get_status(self) -> dict[str, Any]:
-        payload = super().get_status()
-        payload["tunnel"]["rx_bytes"] = "много"  # намеренно неверный тип вместо integer
-        payload["extra_field"] = True  # additionalProperties: false
-        return payload
-
-
-class LeakingConfigState(MockState):
-    """Реализация, возвращающая приватный ключ в ответе."""
-
-    def get_config(self) -> dict[str, Any]:
-        payload = super().get_config()
-        payload["tunnel"]["private_key"] = PRIVATE_KEY_SENTINEL
-        return payload
-
-
-class DroppingFieldState(MockState):
-    """Реализация, потерявшая обязательное поле."""
-
-    def get_config(self) -> dict[str, Any]:
-        payload = super().get_config()
-        del payload["routing"]["kill_switch"]
-        return payload
-
-
-def _client_for(state: MockState, token: str) -> TestClient:
-    return TestClient(create_app(state=state, token=token))
-
-
-def test_valid_implementation_passes(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
-    """Контроль ложных срабатываний: корректная реализация проходит без замечаний."""
-    response = client.get("/api/v1/status", headers=auth_headers)
-    validate_response(spec, "/api/v1/status", "get", 200, response.json())
-
-
-def test_wrong_type_and_extra_field_are_caught(spec: dict[str, Any], token: str) -> None:
-    with _client_for(DeviatingStatusState(), token) as client:
-        response = client.get("/api/v1/status", headers={"Authorization": f"Bearer {token}"})
-    assert response.status_code == 200
-    with pytest.raises(ContractViolationError) as excinfo:
-        validate_response(spec, "/api/v1/status", "get", 200, response.json())
-    message = str(excinfo.value)
-    assert "tunnel/rx_bytes" in message
-    assert "extra_field" in message
-
-
-def test_private_key_in_response_is_caught(spec: dict[str, Any], token: str) -> None:
-    with _client_for(LeakingConfigState(), token) as client:
-        response = client.get("/api/v1/config", headers={"Authorization": f"Bearer {token}"})
-    assert PRIVATE_KEY_SENTINEL in response.text, "подготовка теста: ключ действительно утёк"
-    with pytest.raises(ContractViolationError) as excinfo:
-        validate_response(spec, "/api/v1/config", "get", 200, response.json())
-    assert "private_key" in str(excinfo.value)
-
-
-def test_missing_required_field_is_caught(spec: dict[str, Any], token: str) -> None:
-    with _client_for(DroppingFieldState(), token) as client:
-        response = client.get("/api/v1/config", headers={"Authorization": f"Bearer {token}"})
-    with pytest.raises(ContractViolationError) as excinfo:
-        validate_response(spec, "/api/v1/config", "get", 200, response.json())
-    assert "kill_switch" in str(excinfo.value)
-
-
-def test_undocumented_route_is_caught(spec: dict[str, Any], token: str) -> None:
-    app = create_app(token=token)
-
-    @app.get("/api/v1/undocumented")
-    async def undocumented(request: Request) -> dict[str, bool]:
-        return {"ok": True}
-
-    with pytest.raises(ContractViolationError) as excinfo:
-        assert_routes_match(spec, app)
-    assert "нет в контракте" in str(excinfo.value)
-    assert "/api/v1/undocumented" in str(excinfo.value)
-
-
-def test_missing_route_is_caught(spec: dict[str, Any]) -> None:
-    with pytest.raises(ContractViolationError) as excinfo:
-        assert_routes_match(spec, FastAPI())
-    assert "нет в реализации" in str(excinfo.value)
-
-
-def test_undeclared_status_code_is_caught(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
-    response = client.get("/api/v1/status", headers=auth_headers)
-    with pytest.raises(ContractViolationError) as excinfo:
-        validate_response(spec, "/api/v1/status", "get", 202, response.json())
-    assert "не описан в спецификации" in str(excinfo.value)

backend/tests/test_contract_responses.py

@@ -1,321 +0,0 @@
-"""Контрактный тест «схема ↔ реализация»: каждый эндпоинт сверяется со схемой из docs/openapi.yaml.
-
-Любое расхождение (лишнее поле, неверный тип, отсутствующая обязательная ветка,
-утечка writeOnly-секрета) — падение с указанием пути до поля.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-from fastapi.testclient import TestClient
-from starlette.websockets import WebSocketDisconnect
-
-from awg_api.app import create_app
-from awg_api.contract import (
-    ContractViolationError,
-    assert_routes_match,
-    response_schema,
-    validator_for,
-    validate_response,
-)
-from awg_api.state import PRIVATE_KEY_SENTINEL, PRESHARED_KEY_SENTINEL, MockState
-
-RULE_ID_PATH = "/api/v1/routing/rules/{rule_id}"
-LEAKED_KEY = "NEW-PRIVATE-KEY-MUST-NOT-LEAK"
-SECRET_SENTINELS = (PRIVATE_KEY_SENTINEL, PRESHARED_KEY_SENTINEL, LEAKED_KEY)
-SECRET_FIELD_NAMES = ("private_key", "preshared_key", "password")
-
-TUNNEL_PAYLOAD: dict[str, Any] = {
-    "address": "10.8.0.9/32",
-    "listen_port": 51821,
-    "mtu": 1420,
-    "private_key": LEAKED_KEY,
-    "dns": ["10.8.0.1"],
-}
-ROUTING_PAYLOAD: dict[str, Any] = {
-    "mode": "all-vpn",
-    "kill_switch": True,
-    "mtu_clamping": True,
-    "ipv6": "block",
-    "geoip": {"source": "cache", "prefix_count": 11412, "updated_at": None},
-}
-RULE_PAYLOAD: dict[str, Any] = {
-    "name": "Правило контрактного теста",
-    "kind": "domain",
-    "value": "contract.example.net",
-    "action": "vpn",
-}
-IMPORT_PAYLOAD: dict[str, Any] = {
-    "format": "amneziawg-conf",
-    "content": "[Peer]\nPublicKey = abc\nPresharedKey = def\n",
-}
-
-# (метод, путь, тело) — успешный путь для каждой операции контракта.
-HAPPY_PATHS: tuple[tuple[str, str, dict[str, Any] | None], ...] = (
-    ("get", "/api/v1/health", None),
-    ("get", "/api/v1/ready", None),
-    ("get", "/metrics", None),
-    ("get", "/api/v1/status", None),
-    ("get", "/api/v1/config", None),
-    ("put", "/api/v1/config", {"tunnel": TUNNEL_PAYLOAD}),
-    ("post", "/api/v1/config/validate", {"tunnel": TUNNEL_PAYLOAD}),
-    ("get", "/api/v1/routing/rules", None),
-    ("post", "/api/v1/routing/rules", RULE_PAYLOAD),
-    ("put", "/api/v1/routing/mode", {"mode": "dns"}),
-    ("post", "/api/v1/routing/geoip/update", {"force": True}),
-    ("post", "/api/v1/routing/import", IMPORT_PAYLOAD),
-    ("get", "/api/v1/devices", None),
-    ("get", "/api/v1/logs", None),
-    ("get", "/api/v1/auth/session", None),
-    ("post", "/api/v1/auth/logout", None),
-    ("post", "/api/v1/auth/login", {"username": "admin", "password": "awg-dash-dev-password"}),
-)
-
-EXPECTED_STATUS: dict[tuple[str, str], int] = {("post", "/api/v1/routing/rules"): 201}
-
-
-def _check(spec: dict[str, Any], method: str, path: str, response: Any) -> None:
-    """Валидирует ответ против схемы контракта (JSON — как объект, иначе — как текст)."""
-    content_type = response.headers.get("content-type", "")
-    payload = response.json() if content_type.startswith("application/json") else response.text
-    validate_response(spec, path, method, response.status_code, payload)
-
-
-def test_every_spec_operation_is_implemented(spec: dict[str, Any], app: Any) -> None:
-    assert_routes_match(spec, app)
-
-
-def test_openapi_json_serves_the_contract(client: TestClient, spec: dict[str, Any]) -> None:
-    response = client.get("/openapi.json")
-    assert response.status_code == 200
-    assert response.json() == spec
-
-
-@pytest.mark.parametrize(("method", "path", "body"), HAPPY_PATHS)
-def test_success_response_matches_spec(
-    spec: dict[str, Any],
-    client: TestClient,
-    auth_headers: dict[str, str],
-    method: str,
-    path: str,
-    body: dict[str, Any] | None,
-) -> None:
-    response = client.request(method.upper(), path, json=body, headers=auth_headers)
-    expected = EXPECTED_STATUS.get((method, path), 200)
-    assert response.status_code == expected, f"{method.upper()} {path}: {response.status_code} {response.text}"
-    _check(spec, method, path, response)
-
-
-def test_rule_crud_cycle_matches_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
-    created = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
-    assert created.status_code == 201, created.text
-    _check(spec, "post", "/api/v1/routing/rules", created)
-    rule_id = created.json()["id"]
-
-    fetched = client.get(f"/api/v1/routing/rules/{rule_id}", headers=auth_headers)
-    assert fetched.status_code == 200
-    _check(spec, "get", RULE_ID_PATH, fetched)
-
-    updated = client.put(f"/api/v1/routing/rules/{rule_id}", json={"enabled": False}, headers=auth_headers)
-    assert updated.status_code == 200
-    assert updated.json()["enabled"] is False
-    _check(spec, "put", RULE_ID_PATH, updated)
-
-    deleted = client.delete(f"/api/v1/routing/rules/{rule_id}", headers=auth_headers)
-    assert deleted.status_code == 204
-    assert deleted.content == b""
-
-
-def test_rules_list_filters_match_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
-    response = client.get("/api/v1/routing/rules", params={"enabled": "true", "kind": "cidr"}, headers=auth_headers)
-    assert response.status_code == 200
-    payload = response.json()
-    assert all(rule["kind"] == "cidr" and rule["enabled"] for rule in payload["items"])
-    _check(spec, "get", "/api/v1/routing/rules", response)
-
-
-def test_logs_pagination_matches_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
-    first = client.get("/api/v1/logs", params={"limit": 2}, headers=auth_headers)
-    assert first.status_code == 200
-    payload = first.json()
-    assert len(payload["items"]) == 2
-    assert payload["next_cursor"] is not None
-    _check(spec, "get", "/api/v1/logs", first)
-
-    second = client.get(
-        "/api/v1/logs",
-        params={"limit": 2, "cursor": payload["next_cursor"], "level": "info"},
-        headers=auth_headers,
-    )
-    assert second.status_code == 200
-    assert all(entry["level"] == "info" for entry in second.json()["items"])
-    _check(spec, "get", "/api/v1/logs", second)
-
-
-@pytest.mark.parametrize(
-    ("method", "path", "body", "expected_status", "expected_code"),
-    (
-        ("get", "/api/v1/status", None, 401, "unauthorized"),
-        ("get", "/api/v1/config", None, 401, "unauthorized"),
-        ("get", "/api/v1/routing/rules", None, 401, "unauthorized"),
-        ("get", "/api/v1/devices", None, 401, "unauthorized"),
-        ("get", "/api/v1/logs", None, 401, "unauthorized"),
-        ("post", "/api/v1/routing/rules", {"name": "x", "kind": "domain", "value": "x", "action": "vpn"}, 401, "unauthorized"),
-        (
-            "put",
-            "/api/v1/config",
-            {"tunnel": {**TUNNEL_PAYLOAD, "unknown_field": 1}},
-            422,
-            "validation_failed",
-        ),
-        ("post", "/api/v1/config/validate", {"tunnel": {"address": "10.8.0.2/32", "listen_port": 99999}}, 422, "validation_failed"),
-        ("post", "/api/v1/routing/rules", {"name": "", "kind": "nope", "value": "x"}, 422, "validation_failed"),
-        ("put", "/api/v1/routing/mode", {"mode": "turbo"}, 422, "validation_failed"),
-        ("post", "/api/v1/auth/login", {"username": "admin", "password": "wrong"}, 401, "unauthorized"),
-        ("post", "/api/v1/auth/login", {"username": "", "password": ""}, 422, "validation_failed"),
-        ("get", "/api/v1/logs", None, 422, "validation_failed"),
-    ),
-)
-def test_error_response_matches_error_schema(
-    spec: dict[str, Any],
-    client: TestClient,
-    auth_headers: dict[str, str],
-    method: str,
-    path: str,
-    body: dict[str, Any] | None,
-    expected_status: int,
-    expected_code: str,
-) -> None:
-    headers = dict(auth_headers)
-    params: dict[str, Any] = {}
-    if expected_status == 401 and path != "/api/v1/auth/login":
-        headers = {}
-    if path == "/api/v1/logs" and expected_status == 422:
-        params = {"cursor": "not-a-number"}
-    response = client.request(method.upper(), path, json=body, headers=headers, params=params)
-    assert response.status_code == expected_status, response.text
-    payload = response.json()
-    assert payload["code"] == expected_code
-    assert set(payload) == {"code", "message", "details"}
-    _check(spec, method, path, response)
-
-
-def test_not_found_rule_matches_error_schema(
-    spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]
-) -> None:
-    missing = "00000000-0000-4000-8000-000000000000"
-    url = f"/api/v1/routing/rules/{missing}"
-    responses = [
-        ("get", client.get(url, headers=auth_headers)),
-        ("put", client.put(url, json={"enabled": True}, headers=auth_headers)),
-        ("delete", client.delete(url, headers=auth_headers)),
-    ]
-    for method, response in responses:
-        assert response.status_code == 404, response.text
-        assert response.json()["code"] == "not_found"
-        _check(spec, method, RULE_ID_PATH, response)
-
-
-def test_conflict_on_duplicate_rule_matches_error_schema(
-    spec: dict[str, Any],
-    client: TestClient,
-    auth_headers: dict[str, str],
-) -> None:
-    first = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
-    assert first.status_code == 201
-    duplicate = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
-    assert duplicate.status_code == 409
-    assert duplicate.json()["code"] == "conflict"
-    _check(spec, "post", "/api/v1/routing/rules", duplicate)
-
-
-def test_ready_503_matches_error_schema(spec: dict[str, Any], token: str) -> None:
-    app = create_app(state=MockState(ready=False), token=token)
-    with TestClient(app) as client:
-        response = client.get("/api/v1/ready")
-        assert response.status_code == 503
-        assert response.json()["code"] == "agent_unavailable"
-        _check(spec, "get", "/api/v1/ready", response)
-        health = client.get("/api/v1/health")
-        assert health.status_code == 200
-        assert health.json()["status"] == "degraded"
-        _check(spec, "get", "/api/v1/health", health)
-
-
-def test_validate_endpoint_reports_issues_without_applying(
-    spec: dict[str, Any],
-    client: TestClient,
-    auth_headers: dict[str, str],
-) -> None:
-    response = client.post(
-        "/api/v1/config/validate",
-        json={"tunnel": {"address": "10.8.0.2", "listen_port": 51820, "mtu": 1500, "private_key": "k"}},
-        headers=auth_headers,
-    )
-    assert response.status_code == 200, response.text
-    payload = response.json()
-    assert payload["valid"] is False
-    assert {issue["code"] for issue in payload["errors"]} == {"invalid_cidr"}
-    assert {issue["severity"] for issue in payload["warnings"]} == {"warning"}
-    _check(spec, "post", "/api/v1/config/validate", response)
-
-
-def test_metrics_endpoint_returns_prometheus_text(
-    spec: dict[str, Any],
-    client: TestClient,
-) -> None:
-    response = client.get("/metrics")
-    assert response.status_code == 200
-    assert response.headers["content-type"].startswith("text/plain")
-    assert "awg_dash_tunnel_up 1" in response.text
-    _check(spec, "get", "/metrics", response)
-
-
-def test_responses_contain_no_secret_material(
-    spec: dict[str, Any],
-    client: TestClient,
-    auth_headers: dict[str, str],
-) -> None:
-    """AC: в ответах нет приватных ключей — ни значений, ни имён полей."""
-    responses = [
-        client.get("/api/v1/config", headers=auth_headers),
-        client.put("/api/v1/config", json={"tunnel": TUNNEL_PAYLOAD}, headers=auth_headers),
-        client.get("/api/v1/status", headers=auth_headers),
-        client.get("/api/v1/devices", headers=auth_headers),
-        client.get("/api/v1/logs", headers=auth_headers),
-        client.get("/api/v1/auth/session", headers=auth_headers),
-        client.post("/api/v1/auth/logout", headers=auth_headers),
-    ]
-    for response in responses:
-        body = response.text
-        for sentinel in SECRET_SENTINELS:
-            assert sentinel not in body, f"утечка секрета {sentinel} в {response.request.url}"
-        for field in SECRET_FIELD_NAMES:
-            assert field not in body, f"поле {field} присутствует в ответе {response.request.url}"
-
-
-def test_websocket_frames_match_spec(spec: dict[str, Any], client: TestClient, token: str) -> None:
-    frame_schema = spec["components"]["schemas"]["WsFrame"]
-    metrics_schema = spec["components"]["schemas"]["MetricsSample"]
-    frame_validator = validator_for(frame_schema, spec)
-    data_validator = validator_for(metrics_schema, spec)
-    with client.websocket_connect(f"/api/v1/metrics/stream?token={token}&interval_ms=250") as websocket:
-        frame = websocket.receive_json()
-    assert list(frame_validator.iter_errors(frame)) == []
-    assert frame["type"] == "metrics"
-    assert list(data_validator.iter_errors(frame["data"])) == []
-
-
-def test_websocket_requires_token(client: TestClient) -> None:
-    with pytest.raises(WebSocketDisconnect), client.websocket_connect("/api/v1/metrics/stream") as websocket:
-        websocket.receive_json()
-
-
-def test_response_schema_lookup_fails_for_unknown_response(spec: dict[str, Any], client: TestClient) -> None:
-    """Неописанный в контракте ответ — ошибка контракта, а не молчаливый пропуск."""
-    response = client.get("/api/v1/health")
-    with pytest.raises(ContractViolationError):
-        validate_response(spec, "/api/v1/health", "get", 418, response.json())
-    assert response_schema(spec, "/api/v1/health", "get", 418) is None

backend/tests/test_openapi_spec.py

@@ -1,186 +0,0 @@
-"""Проверки самой спецификации: валидность 3.1, полнота по ТЗ, writeOnly-секреты, единый Error."""
-
-from __future__ import annotations
-
-from typing import Any, cast
-
-import pytest
-from openapi_spec_validator import validate as validate_openapi
-from pydantic import BaseModel
-
-from awg_api.contract import (
-    SECRET_FIELD_NAMES,
-    error_codes,
-    iter_operations,
-    response_schema,
-    validator_for,
-    write_only_fields,
-)
-from awg_api.schemas import (
-    AuthLoginRequest,
-    ConfigUpdateRequest,
-    DnsConfigIn,
-    GeoIpUpdateRequest,
-    ImportRequest,
-    ModeUpdateRequest,
-    PeerConfigIn,
-    RoutingConfigIn,
-    RoutingRuleCreate,
-    RoutingRuleUpdate,
-    TunnelConfigIn,
-)
-
-# Операции, требуемые ТЗ/issue #19.
-REQUIRED_OPERATIONS: tuple[tuple[str, str], ...] = (
-    ("/api/v1/status", "get"),
-    ("/api/v1/config", "get"),
-    ("/api/v1/config", "put"),
-    ("/api/v1/config/validate", "post"),
-    ("/api/v1/routing/rules", "get"),
-    ("/api/v1/routing/rules", "post"),
-    ("/api/v1/routing/rules/{rule_id}", "get"),
-    ("/api/v1/routing/rules/{rule_id}", "put"),
-    ("/api/v1/routing/rules/{rule_id}", "delete"),
-    ("/api/v1/routing/mode", "put"),
-    ("/api/v1/routing/geoip/update", "post"),
-    ("/api/v1/routing/import", "post"),
-    ("/api/v1/metrics/stream", "get"),
-    ("/api/v1/health", "get"),
-    ("/api/v1/ready", "get"),
-    ("/metrics", "get"),
-    ("/api/v1/devices", "get"),
-    ("/api/v1/logs", "get"),
-    ("/api/v1/auth/login", "post"),
-    ("/api/v1/auth/logout", "post"),
-    ("/api/v1/auth/session", "get"),
-)
-
-REQUEST_MODEL_CASES: tuple[tuple[str, type[BaseModel]], ...] = (
-    ("TunnelConfig", TunnelConfigIn),
-    ("PeerConfig", PeerConfigIn),
-    ("DnsConfig", DnsConfigIn),
-    ("RoutingConfig", RoutingConfigIn),
-    ("ConfigUpdateRequest", ConfigUpdateRequest),
-    ("RoutingRuleCreate", RoutingRuleCreate),
-    ("RoutingRuleUpdate", RoutingRuleUpdate),
-    ("ModeUpdateRequest", ModeUpdateRequest),
-    ("GeoIpUpdateRequest", GeoIpUpdateRequest),
-    ("ImportRequest", ImportRequest),
-    ("AuthLoginRequest", AuthLoginRequest),
-)
-
-STATUS_OK_RANGE = range(200, 300)
-# 101 Switching Protocols — WebSocket-апгрейд, не ошибка и не JSON-ответ.
-STATUS_PROTOCOL_SWITCH = 101
-
-
-def _spec_schema(spec: dict[str, Any], name: str) -> dict[str, Any]:
-    return cast(dict[str, Any], spec["components"]["schemas"][name])
-
-
-def test_spec_is_valid_openapi_31(spec: dict[str, Any]) -> None:
-    assert spec["openapi"].startswith("3.1")
-    validate_openapi(spec)
-
-
-@pytest.mark.parametrize(("path", "method"), REQUIRED_OPERATIONS)
-def test_required_operation_declared(spec: dict[str, Any], path: str, method: str) -> None:
-    assert method in spec["paths"].get(path, {}), f"{method.upper()} {path} отсутствует в контракте"
-
-
-def test_every_non_2xx_response_uses_error_schema(spec: dict[str, Any]) -> None:
-    """Единый формат ошибок: любой не-2xx ответ описан схемой Error."""
-    offenders: list[str] = []
-    for path, method, operation in iter_operations(spec):
-        for status in operation.get("responses", {}):
-            if int(status) in STATUS_OK_RANGE or int(status) == STATUS_PROTOCOL_SWITCH:
-                continue
-            schema = response_schema(spec, path, method, status)
-            if not isinstance(schema, dict) or not str(schema.get("$ref", "")).endswith("/Error"):
-                offenders.append(f"{method.upper()} {path} -> {status}")
-    assert offenders == [], f"ответы без схемы Error: {offenders}"
-
-
-def test_error_codes_are_closed_enum(spec: dict[str, Any]) -> None:
-    codes = error_codes(spec)
-    assert {"validation_failed", "unauthorized", "not_found", "internal_error"} <= codes
-
-
-def test_secret_fields_are_write_only(spec: dict[str, Any]) -> None:
-    declared = write_only_fields(spec)
-    assert SECRET_FIELD_NAMES <= declared, f"секреты без writeOnly: {SECRET_FIELD_NAMES - declared}"
-
-
-def test_write_only_secrets_removed_from_response_schema(spec: dict[str, Any]) -> None:
-    """Схема ответа (после вырезания writeOnly) не содержит приватных ключей."""
-    schema = response_schema(spec, "/api/v1/config", "get", 200)
-    assert schema is not None
-    response_document = validator_for(schema, spec).schema
-    tunnel = response_document["components"]["schemas"]["TunnelConfig"]
-    peer = response_document["components"]["schemas"]["PeerConfig"]
-    assert "private_key" not in tunnel["properties"]
-    assert "private_key" not in tunnel.get("required", [])
-    assert "preshared_key" not in peer["properties"]
-
-
-def test_private_key_in_response_fails_validation(spec: dict[str, Any]) -> None:
-    schema = response_schema(spec, "/api/v1/config", "get", 200)
-    assert schema is not None
-    validator = validator_for(schema, spec)
-    payload = {
-        "tunnel": {
-            "address": "10.8.0.2/32",
-            "listen_port": 51820,
-            "mtu": 1420,
-            "private_key": "must-not-be-here",
-        },
-        "peers": [],
-        "dns": {"mode": "internal", "servers": []},
-        "routing": {
-            "mode": "geoip",
-            "kill_switch": True,
-            "mtu_clamping": True,
-            "ipv6": "direct",
-            "geoip": {"source": "cache", "prefix_count": 1, "updated_at": None},
-        },
-        "updated_at": "2026-10-22T00:00:00Z",
-    }
-    assert list(validator.iter_errors(payload)), "приватный ключ в ответе обязан ломать валидацию"
-
-
-@pytest.mark.parametrize(("schema_name", "model"), REQUEST_MODEL_CASES)
-def test_request_models_match_spec(spec: dict[str, Any], schema_name: str, model: type[BaseModel]) -> None:
-    """Модель запроса и схема контракта совпадают по полям и обязательности."""
-    spec_properties = set(_spec_schema(spec, schema_name).get("properties", {}))
-    spec_required = set(_spec_schema(spec, schema_name).get("required", []))
-    model_schema = model.model_json_schema()
-    model_properties = set(model_schema.get("properties", {}))
-    model_required = set(model_schema.get("required", []))
-    assert model_properties == spec_properties, (
-        f"{schema_name}: поля расходятся (только в контракте: "
-        f"{spec_properties - model_properties}, только в модели: {model_properties - spec_properties})"
-    )
-    assert model_required == spec_required, (
-        f"{schema_name}: обязательность расходится (только в контракте: "
-        f"{spec_required - model_required}, только в модели: {model_required - spec_required})"
-    )
-
-
-def test_secret_fields_write_only_in_request_models() -> None:
-    tunnel_schema = TunnelConfigIn.model_json_schema()
-    peer_schema = PeerConfigIn.model_json_schema()
-    auth_schema = AuthLoginRequest.model_json_schema()
-    assert tunnel_schema["properties"]["private_key"].get("writeOnly") is True
-    assert peer_schema["properties"]["preshared_key"].get("writeOnly") is True
-    assert auth_schema["properties"]["password"].get("writeOnly") is True
-
-
-def test_spec_paths_are_versioned(spec: dict[str, Any]) -> None:
-    """Все прикладные эндпоинты живут под /api/v1 (кроме служебных /metrics и /ready-хелперов)."""
-    allowed_unversioned = {"/metrics"}
-    offenders = [
-        path
-        for path, _method, _operation in iter_operations(spec)
-        if not path.startswith("/api/v1/") and path not in allowed_unversioned
-    ]
-    assert offenders == [], f"эндпоинты вне /api/v1: {offenders}"

requirements-contract.txt

@@ -1,5 +0,0 @@
-# Зависимости контрактных тестов API (issue #19).
-# После мержа каркаса монорепо (PR #47) строки переезжают в requirements-dev.txt.
-jsonschema>=4.23
-openapi-spec-validator>=0.7
-PyYAML>=6

ui/scripts/generate-api.sh

@@ -1,20 +0,0 @@
-#!/usr/bin/env bash
-# Генерация TypeScript-типов клиента из контракта (docs/openapi.yaml, issue #19).
-#
-# Требуется Node.js >= 20 (генератор запускается через npx, версия запинена).
-# Результат коммитится в ui/src/api/generated/schema.d.ts и проверяется в CI.
-set -euo pipefail
-
-root_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
-spec="${root_dir}/docs/openapi.yaml"
-out_dir="${root_dir}/ui/src/api/generated"
-generator="openapi-typescript@7"
-
-if [[ ! -f "${spec}" ]]; then
-  echo "contract not found: ${spec}" >&2
-  exit 1
-fi
-
-mkdir -p "${out_dir}"
-npx --yes "${generator}" "${spec}" -o "${out_dir}/schema.d.ts"
-echo "OK: ${out_dir}/schema.d.ts <- ${spec}"

ui/src/api/generated/README.md

@@ -1,27 +0,0 @@
-# ui/src/api/generated — сгенерированный API-клиент
-
-**Файлы в этой папке не редактируются руками.**
-
-| Файл | Источник |
-|---|---|
-| `schema.d.ts` | `docs/openapi.yaml` (контракт API v1, issue #19) через `openapi-typescript@7` |
-
-## Перегенерация
-
-```bash
-bash ui/scripts/generate-api.sh
-```
-
-Требуется Node.js >= 20. CI-джоба `ts-client-is-current` повторяет генерацию и падает,
-если закоммиченный файл расходится со спецификацией.
-
-## Секреты
-
-Поля `private_key` (TunnelConfig), `preshared_key` (PeerConfig) и `password`
-(AuthLoginRequest) помечены в контракте `writeOnly`:
-
-* их **можно** передавать в запросах (`PUT /api/v1/config`, `POST /api/v1/auth/login`);
-* их **нельзя** ожидать в ответах — сервер обязан их вырезать (см. контрактный тест
-  `backend/tests/test_contract_responses.py::test_responses_contain_no_secret_material`).
-
-Поэтому UI читает секреты только из полей ввода и никогда не рендерит их из ответов API.