1 Commits

Author SHA1 Message Date
Hermes TechLead Bot 99dd8b3c0c smoke 1792234409 2026-10-17 13:53:30 +03:00
75 changed files with 39 additions and 9335 deletions
-22
View File
@@ -1,22 +0,0 @@
root = true
[*]
charset = utf-8
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true
[*.py]
indent_style = space
indent_size = 4
max_line_length = 100
[*.{yml,yaml,json,ts,tsx,js,css,html}]
indent_style = space
indent_size = 2
[Makefile]
indent_style = tab
[*.md]
trim_trailing_whitespace = false
-325
View File
@@ -1,325 +0,0 @@
name: ci
# CI-пайплайн с гейтами (issue #17, Refs #15)
# lint -> unit (coverage >= 70%) -> security (bandit/pip-audit) -> build -> image-scan (trivy) -> push staging-<sha>
# Гейты (роняют pipeline):
# * unit — покрытие < 70 % (--cov-fail-under=70)
# * security — bandit HIGH, pip-audit: любая известная уязвимость
# * image-scan — trivy: HIGH/CRITICAL в образе
# * build стартует только после зелёных lint+unit+security; push — только после зелёного image-scan.
on:
push:
branches: ["feat/**", "fix/**", "chore/**", "ci/**", "main"]
pull_request:
branches: [main]
workflow_dispatch:
env:
PYTHON_VERSION: "3.12"
REGISTRY: 192.168.1.111:3000
IMAGE_NAME: ${{ github.repository }}
jobs:
# ─────────────────────────────── 1. LINT ────────────────────────────────
lint:
name: lint
runs-on: ubuntu-latest
outputs:
has_dockerfile: ${{ steps.dockerfile.outputs.has_dockerfile }}
steps:
- name: checkout
id: checkout
uses: actions/checkout@v4
continue-on-error: true
# Дефект Gitea git-http: upload-pack отдаёт пустой pack (Content-Length: 90),
# git clone/fetch падает "early EOF / invalid index-pack output".
# Обход: эндпоинт archive/<sha>.tar.gz работает — забираем исходники им.
# Когда git-http починят, шаг просто перестанет срабатывать (outcome != failure).
- name: "checkout fallback (archive tarball)"
if: steps.checkout.outcome == 'failure'
run: |
set -euo pipefail
curl -fsSL -o /tmp/src.tar.gz "http://192.168.1.111:3000/${IMAGE_NAME}/archive/${GITHUB_SHA}.tar.gz"
tar -xzf /tmp/src.tar.gz --strip-components=1 -C .
echo "sources: archive fallback, $(ls -1 | wc -l) entries"
- name: "python 3.12"
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
- name: install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-dev.txt -r requirements-contract.txt
- name: ruff check
run: ruff check backend
- name: ruff format --check
run: ruff format --check backend
- name: "mypy (strict)"
run: mypy backend
- name: detect deploy/Dockerfile
id: dockerfile
run: |
if [ -f deploy/Dockerfile ]; then
echo "has_dockerfile=true" >> "$GITHUB_OUTPUT"
echo "deploy/Dockerfile найден — jobs build/image-scan/push активны"
else
echo "has_dockerfile=false" >> "$GITHUB_OUTPUT"
echo "::warning::deploy/Dockerfile отсутствует — build/image-scan/push пропущены (ждут merge #18 / PR #49)"
fi
# ─────────────────── 2. UNIT + ГЕЙТ ПОКРЫТИЯ >= 70 % ───────────────────
unit:
name: unit
needs: lint
runs-on: ubuntu-latest
steps:
- name: checkout
id: checkout
uses: actions/checkout@v4
continue-on-error: true
- name: "checkout fallback (archive tarball)"
if: steps.checkout.outcome == 'failure'
run: |
set -euo pipefail
curl -fsSL -o /tmp/src.tar.gz "http://192.168.1.111:3000/${IMAGE_NAME}/archive/${GITHUB_SHA}.tar.gz"
tar -xzf /tmp/src.tar.gz --strip-components=1 -C .
- name: "python 3.12"
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
- name: install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements-dev.txt -r requirements-contract.txt
- name: "pytest + coverage gate (fail below 70%)"
run: |
python -m pytest \
--cov=awg_agent --cov=awg_api \
--cov-report=term-missing \
--cov-report=xml:coverage.xml \
--cov-report=html:htmlcov \
--cov-fail-under=70
- name: coverage summary
if: always()
run: python -m coverage report --sort=cover
- name: upload coverage artifacts
if: always()
uses: actions/upload-artifact@v3
with:
name: coverage
path: |
coverage.xml
htmlcov/
retention-days: 14
# ───────────── 3. SECURITY: bandit (HIGH) + pip-audit ──────────────────
security:
name: security
needs: lint
runs-on: ubuntu-latest
steps:
- name: checkout
id: checkout
uses: actions/checkout@v4
continue-on-error: true
- name: "checkout fallback (archive tarball)"
if: steps.checkout.outcome == 'failure'
run: |
set -euo pipefail
curl -fsSL -o /tmp/src.tar.gz "http://192.168.1.111:3000/${IMAGE_NAME}/archive/${GITHUB_SHA}.tar.gz"
tar -xzf /tmp/src.tar.gz --strip-components=1 -C .
- name: "python 3.12"
uses: actions/setup-python@v5
with:
python-version: ${{ env.PYTHON_VERSION }}
- name: install scanners
run: |
python -m pip install --upgrade pip
pip install "bandit[toml]>=1.7" "pip-audit>=2.7"
pip install -r requirements.txt -r requirements-contract.txt
- name: "bandit -lll (HIGH severity gate)"
run: |
set -o pipefail
bandit -lll -r backend 2>&1 | tee bandit-report.txt
bandit -lll -f json -o bandit-report.json -r backend || true
echo "bandit: HIGH/CRITICAL не найдены"
- name: "pip-audit (known vulnerabilities gate)"
run: |
pip-audit -r requirements.txt -r requirements-contract.txt --progress-spinner off
- name: pip-audit json report
if: always()
run: pip-audit -r requirements.txt -r requirements-contract.txt --progress-spinner off --format json -o pip-audit-report.json || true
- name: upload security artifacts
if: always()
uses: actions/upload-artifact@v3
with:
name: security-reports
path: |
bandit-report.txt
bandit-report.json
pip-audit-report.json
retention-days: 14
# ───────────────── 4. BUILD: docker build (нужен #18) ──────────────────
build:
name: build
needs: [lint, unit, security]
if: needs.lint.outputs.has_dockerfile == 'true'
runs-on: ubuntu-latest
steps:
- name: checkout
id: checkout
uses: actions/checkout@v4
continue-on-error: true
- name: "checkout fallback (archive tarball)"
if: steps.checkout.outcome == 'failure'
run: |
set -euo pipefail
curl -fsSL -o /tmp/src.tar.gz "http://192.168.1.111:3000/${IMAGE_NAME}/archive/${GITHUB_SHA}.tar.gz"
tar -xzf /tmp/src.tar.gz --strip-components=1 -C .
- name: "guard: deploy/Dockerfile"
run: |
if [ ! -f deploy/Dockerfile ]; then
echo "::error::deploy/Dockerfile отсутствует: сборка образа ждёт merge #18 (PR #49)."
echo "Пайплайн целиком проверен на ветке feat/17-verify (она же + deploy/Dockerfile)."
exit 1
fi
- name: image meta
id: meta
run: |
SHA_SHORT="${GITHUB_SHA::8}"
IMAGE="${REGISTRY}/${IMAGE_NAME}:staging-${SHA_SHORT}"
echo "sha_short=${SHA_SHORT}" >> "$GITHUB_OUTPUT"
echo "image=${IMAGE}" >> "$GITHUB_OUTPUT"
echo "IMAGE=${IMAGE}" >> "$GITHUB_ENV"
echo "image: ${IMAGE}"
- name: docker build
run: |
set -o pipefail
docker build --pull -f deploy/Dockerfile -t "$IMAGE" . 2>&1 | tee build.log
- name: image digest
run: |
docker image inspect "$IMAGE" --format 'id={{.Id}} size={{.Size}}'
docker image inspect "$IMAGE" --format '{{.Id}}' > image-id.txt
- name: upload build artifacts
if: always()
uses: actions/upload-artifact@v3
with:
name: "build-${{ github.sha }}"
path: |
build.log
image-id.txt
retention-days: 14
# ──────── 5. IMAGE-SCAN: trivy, гейт HIGH/CRITICAL ────────────────────
image-scan:
name: image-scan
needs: [lint, build]
if: needs.lint.outputs.has_dockerfile == 'true'
runs-on: ubuntu-latest
env:
TRIVY_VERSION: "0.76.1"
TRIVY_SEVERITY: "HIGH,CRITICAL"
steps:
# trivy запускается официальным образом через docker-сокет: не требует
# скачивания бинаря (github releases отдаёт медленно) + БД кэшируется в
# /opt/trivy-cache на build-хосте.
- name: "trivy image (HIGH/CRITICAL -> fail)"
run: |
set -o pipefail
IMAGE="${REGISTRY}/${IMAGE_NAME}:staging-${GITHUB_SHA::8}"
echo "scan: ${IMAGE}"
docker run --rm \
-v /var/run/docker.sock:/var/run/docker.sock \
-v /opt/trivy-cache:/root/.cache/trivy \
"aquasec/trivy:${TRIVY_VERSION}" image \
--image-src docker --scanners vuln \
--severity "$TRIVY_SEVERITY" --ignore-unfixed \
--exit-code 1 --format table "$IMAGE" 2>&1 | tee trivy-report.txt
- name: trivy json report
if: always()
run: |
set -o pipefail
IMAGE="${REGISTRY}/${IMAGE_NAME}:staging-${GITHUB_SHA::8}"
mkdir -p reports
docker run --rm \
-v /var/run/docker.sock:/var/run/docker.sock \
-v /opt/trivy-cache:/root/.cache/trivy \
-v "$PWD/reports:/reports" \
"aquasec/trivy:${TRIVY_VERSION}" image \
--image-src docker --scanners vuln \
--severity "$TRIVY_SEVERITY" --ignore-unfixed \
--format json -o /reports/trivy-report.json "$IMAGE" || true
cp reports/trivy-report.json ./trivy-report.json 2>/dev/null || true
- name: upload trivy report
if: always()
uses: actions/upload-artifact@v3
with:
name: trivy-report
path: |
trivy-report.txt
trivy-report.json
retention-days: 14
# ─────── 6. PUSH: staging-<sha> в registry Gitea ──────────────────────
push:
name: push
needs: [lint, build, image-scan]
if: needs.lint.outputs.has_dockerfile == 'true' && github.event_name == 'push'
runs-on: ubuntu-latest
steps:
- name: docker login
run: |
set -o pipefail
# Приоритет — секрет с правом write:package (REGISTRY_TOKEN), иначе встроенный токен Actions.
TOKEN="${{ secrets.REGISTRY_TOKEN }}"
if [ -z "$TOKEN" ]; then
TOKEN="${{ secrets.GITHUB_TOKEN }}"
fi
echo "$TOKEN" | docker login "$REGISTRY" -u "${{ github.actor }}" --password-stdin
- name: "push staging-<sha>"
run: |
set -o pipefail
IMAGE="${REGISTRY}/${IMAGE_NAME}:staging-${GITHUB_SHA::8}"
docker push "$IMAGE" 2>&1 | tee push.log
echo "${IMAGE}" | tee pushed-image.txt
- name: upload push record
if: always()
uses: actions/upload-artifact@v3
with:
name: pushed-image
path: |
push.log
pushed-image.txt
retention-days: 14
-55
View File
@@ -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
-36
View File
@@ -1,36 +0,0 @@
# Python
__pycache__/
*.py[cod]
.venv/
.venv*/
*.egg-info/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
coverage.xml
htmlcov/
# UI
node_modules/
ui/dist/
ui/.vite/
# Runtime / состояние шлюза (живёт в томах, не в репозитории)
data/
config/
*.db
*.db-shm
*.db-wal
# Секреты и конфиги клиентов
.env
.env.*
!.env.example
*.conf
!deploy/*.conf.example
# IDE / OS
.idea/
.vscode/
.DS_Store
-24
View File
@@ -1,24 +0,0 @@
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.9
hooks:
- id: ruff
args: [--fix]
- id: ruff-format
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.11.2
hooks:
- id: mypy
files: ^backend/
additional_dependencies: [pydantic>=2.9]
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.6.0
hooks:
- id: end-of-file-fixer
- id: trailing-whitespace
- id: check-yaml
- id: check-merge-conflict
- id: check-added-large-files
args: [--maxkb=512]
-44
View File
@@ -1,44 +0,0 @@
PY ?= python3
VENV ?= .venv
BIN := $(VENV)/bin
IMAGE ?= awg-dash:dev
COMPOSE_DEV := deploy/docker-compose.dev.yml
.PHONY: help venv lint fmt test build dev-up dev-down dev-logs test-netns
help:
@grep -E '^[a-zA-Z_-]+:' $(MAKEFILE_LIST) | sed 's/:.*//' | sort -u
venv:
$(PY) -m venv $(VENV)
$(BIN)/pip install -U pip
$(BIN)/pip install -r requirements-dev.txt
lint:
$(BIN)/ruff check backend
$(BIN)/mypy backend
fmt:
$(BIN)/ruff format backend
$(BIN)/ruff check --fix backend
test:
$(BIN)/pytest
build:
@test -f deploy/Dockerfile || { echo 'deploy/Dockerfile ещё нет — задача #18'; exit 1; }
docker build -f deploy/Dockerfile -t $(IMAGE) .
dev-up:
@test -f $(COMPOSE_DEV) || { echo '$(COMPOSE_DEV) ещё нет — задача #21'; exit 1; }
docker compose -f $(COMPOSE_DEV) up -d --build
dev-down:
docker compose -f $(COMPOSE_DEV) down
dev-logs:
docker compose -f $(COMPOSE_DEV) logs -f --tail=100
test-netns:
@test -d tests/netns || { echo 'tests/netns ещё нет — задача #21'; exit 1; }
bash tests/netns/run.sh
+29 -53
View File
@@ -1,64 +1,40 @@
# awg-dash
VPN-шлюз **AmneziaWG** с интеллектуальной маршрутизацией: один Docker-образ = туннель + policy routing + split-tunnel (GeoIP / DNS) + REST API v1 + Web UI (тёмная тема).
# awg-dash 🚀
- Эпик и декомпозиция: Gitea issue **#15** (`embeddevio/awg-dash`).
- Исполнение: Hermes Kanban, борда **`awg-dash`** (`hermes kanban --board awg-dash list`).
- PoC-референс: `embeddevio/amneziawg-docker-gateway` (bash + iptables, `docs/technical-specification.md`).
**awg-dash** — это легковесная и удобная веб-панель (Dashboard) для визуального управления сервером **Amnezia WireGuard (AWG)**. Она позволяет создавать, удалять и просматривать конфигурации пользователей в один клик через современный веб-интерфейс, защищенный от блокировок.
## Архитектура
[![License: MIT](https://shields.io)](https://opensource.org)
[![GitHub Stars](https://shields.io)](https://github.com)
Два процесса в одном образе:
---
| Процесс | Пользователь | Ответственность |
|---|---|---|
| `awg-agent` | root | awg/wg-интерфейс, iptables/ipset, dnsmasq, policy routing, метрики |
| `awg-api` | непривилегированный | FastAPI: REST API v1, WebSocket, статика UI |
## ✨ Особенности / Features
Обмен — unix-socket `/run/awg/agent.sock`; состояние — SQLite в `/data`, конфиги `.conf` в `/config`.
Принятые решения зафиксированы в `docs/adr/`.
- 🛠️ **Полная поддержка Amnezia WireGuard:** Тонкая настройка параметров обфускации (`Jc`, `Jmin`, `Jmax`, `S1`, `S2`, `H1`, `H2`, `H3`, `H4`) прямо из панели.
- 👥 **Управление пользователями:** Создание, удаление, временная заморозка и управление лимитами пиров (клиентов).
- 📊 **Мониторинг в реальном времени:** Графики потребления трафика, статус подключения клиентов и общая нагрузка на сервер.
- 📱 **QR-коды и конфиги:** Мгновенная генерация QR-кодов и скачивание `.conf` файлов для импорта в официальное приложение Amnezia.
- 🔒 **Безопасность:** Авторизация по паролю/токену, поддержка сессий и возможность работы за обратным прокси (Nginx, Caddy).
- 🐳 **Docker-first:** Быстрый запуск одной командой через Docker Compose.
## Быстрый старт (после релиза 1.0)
---
## 🛠️ Технологический стек / Tech Stack
- **Backend:** Node.js (Express) / Python (FastAPI) / Go *(выберите нужное)*
- **Frontend:** Vue.js / React / Простой HTML+JS *(выберите нужное)*
- **База данных:** SQLite / JSON-file (минималистично, без тяжелых БД)
- **Ядро:** Amnezia WireGuard (модифицированная утилита `wg`)
---
## 🚀 Быстрый старт / Quick Start
### Способ 1: Через Docker Compose (Рекомендуемый)
1. Клонируйте репозиторий:
```bash
docker run -dit --name=vpn-gateway \
--network=host \
--device=/dev/net/tun:/dev/net/tun \
--cap-add=NET_ADMIN --cap-add=SYS_MODULE \
--restart always \
-v /srv/awg/config:/config -v /srv/awg/data:/data \
registry.example.com/vpn-gateway:latest
git clone https://github.com
cd awg-dash
```
Web UI: `http://<host>:8080`. До релиза образ собирается локально: `make build`.
## Разработка
```bash
make venv # локальное окружение (.venv)
make lint # ruff + mypy
make test # pytest + coverage (порог 70%)
make build # docker build образа (задача #18)
make dev-up # dev-стенд (задача #21)
```
Сетевые тесты выполняются в netns-стенде (`tests/netns/`), без реального VPN-провайдера.
## Структура репозитория
| Путь | Назначение |
|---|---|
| `backend/awg_agent/` | привилегированный агент: туннель, маршрутизация, GeoIP, DNS, метрики |
| `backend/awg_api/` | FastAPI-приложение (REST + WS + статика UI) |
| `backend/tests/` | unit/integration-тесты (pytest) |
| `ui/` | React + TypeScript + Tailwind + shadcn-ui |
| `deploy/` | Dockerfile, compose-профили, entrypoint, Grafana-дашборд |
| `docs/adr/` | архитектурные решения |
| `tests/netns/` | сетевой тестовый стенд |
## Процесс разработки
1 задача = 1 issue = 1 ветка `feat/<issue>-slug` = 1 PR.
Гейты мержа: CI green, coverage нового кода ≥ 70 %, security scan low/none, ревью владельца роли.
Merge в `main` и деплой в prod — только TechLead после гейтов. Все операции с репозиторием — через Gitea MCP.
+1
View File
@@ -0,0 +1 @@
SMOKE 1792234409
-16
View File
@@ -1,16 +0,0 @@
# awg_agent — привилегированный агент
Запускается как root в контейнере: всё, что требует NET_ADMIN и доступа к ядру.
| Модуль | Ответственность | Задача |
|---|---|---|
| `vpn/` | парсинг/валидация `.conf`, up/down/restart туннеля, `awg show` → статус | #22 |
| `watchdog/` | контроль handshake, авто-восстановление, fail-closed | #23 |
| `netfilter/` | iptables/ipset, policy routing, atomic swap, MSS clamping, IPv6 | #24 |
| `geoip/` | RIPE Stat, кэш-снапшот, расписание, fallback | #25 |
| `dns/` | dnsmasq, домены → ipset, защита от DNS-утечки | #26 |
| `routing/` | режимы (GeoIP / DNS / All-VPN / Direct-only), reconcile, исключения | #27 |
| `metrics/` | RX/TX, direct vs VPN, история, агрегаты | #28 |
| `devices/` | активные устройства LAN и их политики | #29 |
Интерфейс для `awg_api` — unix-socket `/run/awg/agent.sock` (см. ADR-002).
-5
View File
@@ -1,5 +0,0 @@
"""Привилегированный агент awg-dash: туннель, маршрутизация, GeoIP, DNS, метрики."""
__version__ = "0.0.1"
__all__ = ["__version__"]
-13
View File
@@ -1,13 +0,0 @@
# awg_api — control plane
Непривилегированный процесс: FastAPI + REST API v1 + WebSocket + отдача статики UI.
| Модуль | Ответственность | Задача |
|---|---|---|
| `app.py` | lifespan, конфиг, логирование, error model, статика | #31 |
| `routers/` | эндпоинты по OpenAPI (status, config, routing, devices, logs, auth) | #32 |
| `ws/` | `/api/v1/metrics/stream` | #33 |
| `security/` | токен, CSRF, rate limit, маскировка секретов | #34 |
| `store/` | SQLite (WAL), миграции, audit log, apply/rollback | #20 |
Контракт — `docs/openapi.yaml` (задача #19), TS-клиент генерируется в `ui/src/api/generated/`.
-5
View File
@@ -1,5 +0,0 @@
"""FastAPI-приложение awg-dash: REST API v1, WebSocket, статика UI."""
__version__ = "0.0.1"
__all__ = ["__version__"]
-47
View File
@@ -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
-242
View File
@@ -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)
)
-53
View File
@@ -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)]
-88
View File
@@ -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),
)
-1
View File
@@ -1 +0,0 @@
"""Роутеры API v1."""
-34
View File
@@ -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()
-27
View File
@@ -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)
-29
View File
@@ -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)
-36
View File
@@ -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())
-67
View File
@@ -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)
-16
View File
@@ -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()
-120
View File
@@ -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})
-471
View File
@@ -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,
}
-66
View File
@@ -1,66 +0,0 @@
# store — хранилище состояния (issue #20)
SQLite (WAL) в `/data/awg-dash.db`: правила маршрутизации, исключения, режим,
расписание GeoIP, метаданные конфигов, устройства, журнал аудита и снимки
состояния. Путь переопределяется аргументом или переменной `AWG_DASH_DB`.
| Модуль | Назначение |
|---|---|
| `db.py` | открытие соединения, PRAGMA (WAL, `synchronous=NORMAL`, `foreign_keys`, `busy_timeout`), транзакции `BEGIN IMMEDIATE` |
| `migrations.py` | версионированные миграции, `schema_migrations`, проверка схемы при старте (`verify_schema`) |
| `models.py` | dataclass-модели сущностей и перечисления (`Mode`, `Entity`, `Rule`, …) |
| `validation.py` | чистые функции валидации (domain/CIDR/ipset, MAC, cron, режим, политика) |
| `repository.py` | `Store`: чтения + аудируемые мутации «кто/когда/что/old→new» |
| `audit.py` | запись и чтение `audit_log` (мутация и её аудит — одна транзакция) |
| `state.py` | снимки состояния (`state_snapshots`), `capture_state`, `restore_state`, `rollback_to_snapshot` |
| `apply.py` | `ApplyEngine`: `validate → stage → apply → rollback` и протокол `Applier` (awg/iptables) |
| `service.py` | `open_storage()` — сборка `Database + Store + ApplyEngine` для старта приложения |
| `errors.py` | `StorageError`, `SchemaError`, `ValidationError`, `ApplyError(restored=…)` |
## Гарантии
- **Пустой `/data`** — каталог создаётся, миграции применяются, схема v1 появляется без ручных действий.
- **Версия схемы** — при старте схема проверяется: новее ожидаемой или не мигрированная → `SchemaError` (без «автоотката»).
- **Аудит** — каждая мутация пишет `actor/ts/action/entity/old_value/new_value` в `audit_log` в той же транзакции; `old_value` читается внутри той же транзакции, поэтому конкурентные обновления не рвут цепочку `old→new`.
- **Атомарность** — применяемое состояние проходит `validate → stage → apply`; падение внешнего применителя откатывает транзакцию БД и вызывает `Applier.revert(previous)`, `ApplyError.restored=True`. Ловится `BaseException`, поэтому отмена (`asyncio.CancelledError`), `KeyboardInterrupt` и `SystemExit` тоже дают `revert` и запись `failed` в журнале.
- **Откат** — `ApplyEngine.rollback(plan_id)` возвращает состояние из снимка.
## Потокобезопасность (#54)
Контракт: **одно соединение на процесс, все операции под одним `RLock`**.
| Операция | API | Поведение |
|---|---|---|
| Мутация | `Database.transaction()` | `BEGIN IMMEDIATE` … `COMMIT`/`ROLLBACK` под локом |
| Чтение | `Database.read()` | берёт тот же лок; ждёт завершения чужой транзакции |
| Прямой доступ | `Database.connection` / `Store.raw()` | лок **не** берётся — только однопоточный код (миграции, тесты, отладка) |
Следствия, на которые можно опираться в API-слое (#19/#21):
- обычные чтения (`GET /rules`, `GET /mode`, …) не падают от конкурентных мутаций и не отдают перемешанный результат операторов;
- читатель **не видит** незафиксированных строк другой транзакции;
- во время `apply` наружу не видно состояние, которое затем будет откачено (AC «не частично применённая конфигурация»);
- конкурентные запросы ждут, а не получают ошибку; потокобезопасность подтверждена `backend/tests/test_store_concurrency.py`.
WAL остаётся включённым: он даёт устойчивость к падению процесса и позволяет читать БД снаружи (`sqlite3` CLI) во время работы шлюза. Внутрипроцессного параллелизма чтений контракт не обещает: если позже понадобится параллельная пропускная способность, меняем контракт на «соединение на поток» (`threading.local`), а не полагаемся на неявное поведение.
## Пример
```python
from awg_api.store import DesiredState, Rule, open_storage
with open_storage() as storage: # /data/awg-dash.db
storage.store.set_geoip_schedule(cron="0 3 * * *", actor="ui:admin")
plan = storage.engine.stage(
DesiredState(
mode=storage.store.get_mode(), rules=(Rule("yt", "domain", "youtube.com", "tun0"),)
),
reason="ui: split-tunnel для YouTube",
)
result = storage.engine.apply(plan)
```
## Тесты
`backend/tests/test_store_*.py` — миграции, аудит, репозиторий, apply/rollback,
персистентность после рестарта. Порог покрытия — 70 % (`pyproject.toml`).
-103
View File
@@ -1,103 +0,0 @@
"""Слой хранения состояния awg-dash (issue #20).
Состав: SQLite (WAL) в ``/data/awg-dash.db``, версионированные миграции,
проверка схемы при старте, журнал аудита «кто/когда/что/old→new» и движок
атомарного применения ``validate -> stage -> apply -> rollback``.
"""
from __future__ import annotations
from .apply import (
Applier,
ApplyEngine,
ApplyResult,
DesiredState,
NullApplier,
StagedPlan,
)
from .audit import count as audit_count
from .audit import last_entry as audit_last_entry
from .audit import list_entries as list_audit_entries
from .audit import record as record_audit
from .db import DB_PATH_ENV, DEFAULT_DB_PATH, Database, resolve_db_path
from .errors import ApplyError, SchemaError, StorageError, ValidationError
from .migrations import (
EXPECTED_TABLES,
LATEST_VERSION,
MIGRATIONS,
SCHEMA_MIGRATIONS,
Migration,
MigrationReport,
current_version,
migrate,
verify_schema,
)
from .models import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
DEVICE_POLICIES,
RULE_KINDS,
AuditEntry,
ConfigMeta,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
from .repository import Store
from .service import Storage, open_storage, rollback_to
from .state import StateSnapshot, capture_state, restore_state, rollback_to_snapshot
__all__ = [
"AUDIT_STATUS_FAILED",
"AUDIT_STATUS_OK",
"AUDIT_STATUS_STAGED",
"DB_PATH_ENV",
"DEFAULT_DB_PATH",
"DEVICE_POLICIES",
"EXPECTED_TABLES",
"LATEST_VERSION",
"MIGRATIONS",
"RULE_KINDS",
"SCHEMA_MIGRATIONS",
"Applier",
"ApplyEngine",
"ApplyError",
"ApplyResult",
"AuditEntry",
"ConfigMeta",
"Database",
"DesiredState",
"Device",
"Entity",
"GeoIpSchedule",
"Migration",
"MigrationReport",
"Mode",
"NullApplier",
"Rule",
"RuleException",
"SchemaError",
"StagedPlan",
"StateSnapshot",
"Storage",
"StorageError",
"Store",
"ValidationError",
"audit_count",
"audit_last_entry",
"capture_state",
"current_version",
"list_audit_entries",
"migrate",
"open_storage",
"record_audit",
"resolve_db_path",
"restore_state",
"rollback_to",
"rollback_to_snapshot",
"verify_schema",
]
-325
View File
@@ -1,325 +0,0 @@
"""Атомарное применение состояния: ``validate -> stage -> apply -> rollback`` (issue #20).
Гарантии:
* **validate** — несостоятельное желаемое состояние не доходит до БД;
* **stage** — перед мутацией сохраняется снимок текущего состояния (durable,
в отдельной таблице ``state_snapshots``), поэтому откат возможен в любой момент;
* **apply** — мутация БД и запись в ``audit_log`` выполняются в одной транзакции,
внешнее применение (iptables/awg) вызывается внутри неё: падение внешнего
применителя откатывает транзакцию, то есть БД возвращается к предыдущему
состоянию целиком, без частичных изменений;
* **rollback** — явный возврат к снимку по id.
Ошибка применения поднимает :class:`ApplyError` (поле ``restored=True``): API
должен ответить ошибкой, не «полуприменённой» конфигурацией. Ловится
``BaseException`` (#55 D-1), поэтому отмена запроса тоже вызывает
``Applier.revert(previous)`` и закрывает запись аудита исходом ``failed`` —
«висящих» записей ``staged`` без исхода не остаётся.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable
from . import audit, state, validation
from .errors import ApplyError, ValidationError
from .models import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
from .repository import Store
from .state import StateSnapshot
@dataclass(frozen=True, slots=True)
class DesiredState:
"""Полное желаемое состояние шлюза (то, что применяется как единое целое)."""
mode: Mode
rules: tuple[Rule, ...] = ()
geoip: GeoIpSchedule = field(default_factory=GeoIpSchedule)
exceptions: tuple[RuleException, ...] = ()
def to_payload(self) -> dict[str, Any]:
return {
"mode": self.mode.value,
"geoip": self.geoip.as_dict(),
"rules": [
{
"name": rule.name,
"kind": rule.kind,
"match": rule.match,
"target": rule.target,
"priority": rule.priority,
"enabled": rule.enabled,
}
for rule in self.rules
],
"exceptions": [
{
"match": exception.match,
"reason": exception.reason,
"enabled": exception.enabled,
}
for exception in self.exceptions
],
}
@classmethod
def from_payload(cls, payload: Mapping[str, Any]) -> DesiredState:
"""Разобрать желаемое состояние из JSON-совместимого словаря."""
raw_rules = payload.get("rules", [])
if not isinstance(raw_rules, (list, tuple)):
raise ValidationError("rules", "ожидается список правил")
rules = tuple(_rule_from_mapping(item) for item in raw_rules)
raw_exceptions = payload.get("exceptions", [])
if not isinstance(raw_exceptions, (list, tuple)):
raise ValidationError("exceptions", "ожидается список исключений")
exceptions = tuple(_exception_from_mapping(item) for item in raw_exceptions)
geoip_raw = payload.get("geoip") or {}
if not isinstance(geoip_raw, Mapping):
raise ValidationError("geoip", "ожидается объект расписания")
geoip = GeoIpSchedule(
enabled=bool(geoip_raw.get("enabled", True)),
cron=str(geoip_raw.get("cron", "0 4 * * *")),
provider=str(geoip_raw.get("provider", "ripe")),
)
return cls(
mode=validation.validate_mode(str(payload.get("mode", Mode.OFF.value))),
rules=rules,
geoip=geoip,
exceptions=exceptions,
)
def _rule_from_mapping(item: Any) -> Rule:
if not isinstance(item, Mapping):
raise ValidationError("rules", f"ожидается объект правила, получено {item!r}")
try:
return Rule(
name=str(item["name"]),
kind=str(item["kind"]),
match=str(item["match"]),
target=str(item["target"]),
priority=int(item.get("priority", 100)),
enabled=bool(item.get("enabled", True)),
)
except KeyError as exc:
raise ValidationError("rules", f"в правиле нет обязательного поля {exc.args[0]!r}") from exc
def _exception_from_mapping(item: Any) -> RuleException:
if not isinstance(item, Mapping):
raise ValidationError("exceptions", f"ожидается объект исключения, получено {item!r}")
try:
return RuleException(
match=str(item["match"]),
reason=str(item.get("reason", "")),
enabled=bool(item.get("enabled", True)),
)
except KeyError as exc:
raise ValidationError(
"exceptions", f"в исключении нет обязательного поля {exc.args[0]!r}"
) from exc
@runtime_checkable
class Applier(Protocol):
"""Внешний применитель состояния (awg/iptables/ipset)."""
def apply(self, payload: Mapping[str, Any]) -> None:
"""Применить состояние во внешней подсистеме. Падение -> ApplyError."""
def revert(self, payload: Mapping[str, Any]) -> None:
"""Вернуть внешнюю подсистему к состоянию ``payload`` при откате."""
@dataclass(frozen=True, slots=True)
class StagedPlan:
"""Промежуточный план: снимок предыдущего состояния + желаемое состояние."""
plan_id: int
desired: DesiredState
previous: dict[str, Any]
actor: str
reason: str
@dataclass(frozen=True, slots=True)
class ApplyResult:
"""Успешный результат применения."""
ok: bool
plan_id: int
audit_id: int
message: str
class NullApplier:
"""Заглушка внешнего применителя (для тестов и dry-run режима)."""
def apply(self, payload: Mapping[str, Any]) -> None:
return None
def revert(self, payload: Mapping[str, Any]) -> None:
return None
class ApplyEngine:
"""Оркестратор ``validate -> stage -> apply -> rollback``."""
def __init__(
self, store: Store, applier: Applier | None = None, *, actor: str = "system"
) -> None:
self._store = store
self._applier: Applier = applier if applier is not None else NullApplier()
self._actor = actor
# -- стадии ---------------------------------------------------------
def validate(self, desired: DesiredState) -> DesiredState:
"""Проверить желаемое состояние, не касаясь БД."""
validation.validate_mode(desired.mode)
validation.validate_cron(desired.geoip.cron)
validation.validate_provider(desired.geoip.provider)
seen: set[str] = set()
for rule in desired.rules:
validation.validate_name("name", rule.name)
kind = validation.validate_rule_kind(rule.kind)
validation.validate_match(kind, rule.match)
validation.validate_target(rule.target)
validation.validate_priority(rule.priority)
if rule.name in seen:
raise ValidationError("rules", f"дублирующееся имя правила: {rule.name!r}")
seen.add(rule.name)
for exception in desired.exceptions:
validation.validate_exception_match(exception.match)
return desired
def stage(self, desired: DesiredState, *, reason: str = "") -> StagedPlan:
"""Проверить состояние и сохранить снимок предыдущего состояния."""
self.validate(desired)
payload = desired.to_payload()
with self._store.database.transaction() as conn:
previous = state.capture_state(conn)
plan_id = state.save_snapshot(conn, previous, reason=reason or "pre-apply")
audit.record(
conn,
actor=self._actor,
action="stage",
entity=Entity.APPLY.value,
entity_id=plan_id,
old_value=previous,
new_value=payload,
status=AUDIT_STATUS_STAGED,
detail=reason,
)
return StagedPlan(
plan_id=plan_id, desired=desired, previous=previous, actor=self._actor, reason=reason
)
def apply(self, plan: StagedPlan) -> ApplyResult:
"""Применить план. При падении — откатить БД и внешнюю подсистему.
:raises ApplyError: если применитель упал; ``restored=True`` означает,
что предыдущее состояние подтверждённо восстановлено.
Ловится ``BaseException``, не только ``Exception`` (#55 D-1): отмена
запроса (``asyncio.CancelledError``), ``KeyboardInterrupt`` и
``SystemExit`` тоже обязаны вернуть внешнюю подсистему в предыдущее
состояние и закрыть запись аудита исходом ``failed``. Иначе iptables
останется в новом состоянии, тогда как ``audit_log`` сохранит
«висящую» запись ``staged`` без исхода. Исключения из ``Exception``
преобразуются в :class:`ApplyError`, остальные пробрасываются как есть
(семантика отмены сохраняется).
"""
payload = plan.desired.to_payload()
audit_id: int
try:
with self._store.database.transaction() as conn:
state.restore_state(conn, payload)
self._applier.apply(payload)
audit_id = audit.record(
conn,
actor=plan.actor,
action="apply",
entity=Entity.APPLY.value,
entity_id=plan.plan_id,
old_value=plan.previous,
new_value=payload,
status=AUDIT_STATUS_OK,
detail=plan.reason,
)
except BaseException as exc:
restored_ok = self._rollback_failed_apply(plan, exc)
if not isinstance(exc, Exception):
raise
raise ApplyError(
f"применение состояния не удалось: {exc}",
restored=restored_ok,
audit_id=self._last_failed_audit_id(plan),
) from exc
return ApplyResult(
ok=True, plan_id=plan.plan_id, audit_id=audit_id, message="состояние применено"
)
def apply_payload(self, payload: Mapping[str, Any], *, reason: str = "") -> ApplyResult:
"""validate + stage + apply одной командой (для сервисного слоя)."""
desired = self.validate(DesiredState.from_payload(payload))
return self.apply(self.stage(desired, reason=reason))
def rollback(self, plan_id: int, *, reason: str = "") -> StateSnapshot:
"""Явно вернуть состояние к снимку плана."""
with self._store.database.transaction() as conn:
return state.rollback_to_snapshot(conn, plan_id, actor=self._actor, reason=reason)
# -- внутреннее -----------------------------------------------------
def _rollback_failed_apply(self, plan: StagedPlan, exc: BaseException) -> bool:
"""Вернуть внешнюю подсистему и убедиться, что БД = предыдущее состояние.
Откат внешней подсистемы — best-effort: падение ``revert`` не мешает
восстановить БД и записать исход в журнал (причина падения revert
попадает в ``detail``), и наоборот — падение revert не должно скрывать
исходное исключение.
"""
revert_error: BaseException | None = None
try:
self._applier.revert(plan.previous)
except BaseException as revert_exc: # откат best-effort: исход важнее падения revert
revert_error = revert_exc
detail = f"{type(exc).__name__}: {exc}"
if revert_error is not None:
detail += f"; revert: {type(revert_error).__name__}: {revert_error}"
with self._store.database.transaction() as conn:
state.restore_state(conn, plan.previous)
audit.record(
conn,
actor=plan.actor,
action="apply",
entity=Entity.APPLY.value,
entity_id=plan.plan_id,
old_value=plan.desired.to_payload(),
new_value=plan.previous,
status=AUDIT_STATUS_FAILED,
detail=detail,
)
with self._store.database.read() as conn:
return state.capture_state(conn) == plan.previous
def _last_failed_audit_id(self, plan: StagedPlan) -> int | None:
with self._store.database.read() as conn:
entries = audit.list_entries(
conn, entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=1
)
if not entries:
return None
last = entries[0]
return last.id if last.status == AUDIT_STATUS_FAILED else None
-125
View File
@@ -1,125 +0,0 @@
"""Журнал аудита «кто/когда/что/old->new» (issue #20).
Запись всегда выполняется на переданном соединении, то есть внутри уже
открытой транзакции вызывающего кода: мутация и её аудит коммитятся вместе и
не могут разойтись.
"""
from __future__ import annotations
import json
import sqlite3
from collections.abc import Mapping
from typing import Any
from .clock import utcnow
from .models import AUDIT_STATUS_OK, AuditEntry
__all__ = ["list_entries", "record"]
def _dumps(value: Any) -> str | None:
if value is None:
return None
if isinstance(value, Mapping):
return json.dumps(dict(value), sort_keys=True, ensure_ascii=False, default=str)
return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)
def _loads(value: Any) -> Any:
if value is None:
return None
if isinstance(value, (bytes, bytearray)):
value = value.decode("utf-8")
return json.loads(str(value))
def record(
conn: sqlite3.Connection,
*,
actor: str,
action: str,
entity: str,
entity_id: str | int | None = None,
old_value: Any = None,
new_value: Any = None,
status: str = AUDIT_STATUS_OK,
detail: str | None = None,
ts: str | None = None,
) -> int:
"""Записать событие в ``audit_log`` и вернуть id записи."""
if not actor:
raise ValueError("actor обязателен: аудит должен знать, кто изменил состояние")
cursor = conn.execute(
"""
INSERT INTO audit_log (ts, actor, action, entity, entity_id,
old_value, new_value, status, detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ts or utcnow(),
actor,
action,
entity,
None if entity_id is None else str(entity_id),
_dumps(old_value),
_dumps(new_value),
status,
detail,
),
)
return int(cursor.lastrowid or 0)
def list_entries(
conn: sqlite3.Connection,
*,
entity: str | None = None,
entity_id: str | int | None = None,
limit: int = 100,
) -> list[AuditEntry]:
"""Прочитать журнал аудита (свежие первыми)."""
sql = "SELECT * FROM audit_log"
clauses: list[str] = []
params: list[Any] = []
if entity is not None:
clauses.append("entity = ?")
params.append(entity)
if entity_id is not None:
clauses.append("entity_id = ?")
params.append(str(entity_id))
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY id DESC LIMIT ?"
params.append(int(limit))
rows = conn.execute(sql, params).fetchall()
return [
AuditEntry(
id=int(row["id"]),
ts=str(row["ts"]),
actor=str(row["actor"]),
action=str(row["action"]),
entity=str(row["entity"]),
entity_id=None if row["entity_id"] is None else str(row["entity_id"]),
old_value=_loads(row["old_value"]),
new_value=_loads(row["new_value"]),
status=str(row["status"]),
detail=None if row["detail"] is None else str(row["detail"]),
)
for row in rows
]
def count(conn: sqlite3.Connection, *, entity: str | None = None) -> int:
"""Сколько записей в журнале (опционально — по сущности)."""
if entity is None:
row = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()
else:
row = conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity = ?", (entity,)).fetchone()
return int(row[0]) if row is not None else 0
def last_entry(conn: sqlite3.Connection, *, entity: str | None = None) -> AuditEntry | None:
"""Последняя запись журнала (опционально — по сущности)."""
entries = list_entries(conn, entity=entity, limit=1)
return entries[0] if entries else None
-16
View File
@@ -1,16 +0,0 @@
"""Источник времени для audit_log и снимков состояния.
Отдельный модуль — чтобы тесты могли подменять время детерминированно
(``monkeypatch.setattr(clock, "utcnow", ...)``).
"""
from __future__ import annotations
from datetime import UTC, datetime
__all__ = ["utcnow"]
def utcnow() -> str:
"""Текущее время в UTC: ISO-8601, миллисекунды, суффикс ``Z``."""
return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
-175
View File
@@ -1,175 +0,0 @@
"""Открытие и жизненный цикл SQLite-хранилища состояния (issue #20, ADR-006).
Путь по умолчанию — ``/data/awg-dash.db`` (персистентный volume контейнера),
переопределяется аргументом или переменной окружения ``AWG_DASH_DB``.
Каталог создаётся автоматически, поэтому на пустом ``/data`` схема v1
поднимается без ручных действий.
Соединение открывается в режиме WAL: ``synchronous=NORMAL`` достаточен для WAL
(риск потери только последних транзакций при падении операционной системы, БД
при этом не разрушается).
Контракт потокобезопасности (#54): соединение одно на процесс, все операции —
и записи (:meth:`Database.transaction`), и чтения (:meth:`Database.read`) —
выполняются под одним ``RLock``. Поэтому:
* результаты операторов из разных потоков не перемешиваются (нет
``TypeError``/``IndexError``);
* читатель не видит незафиксированных строк чужой транзакции и состояния,
которое затем будет откачено (падение внешнего применителя);
* конкурентные запросы ждут, не падают.
WAL при этом остаётся: он даёт устойчивость к падению процесса и позволяет
читать БД снаружи (``sqlite3`` CLI) во время работы шлюза. Если позже
понадобится параллельная пропускная способность чтений, контракт меняется на
«соединение на поток» (``threading.local``) — это отдельное решение, не
молчаливое поведение.
"""
from __future__ import annotations
import os
import sqlite3
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from .errors import StorageError
from .migrations import MigrationReport, current_version, migrate, verify_schema
DEFAULT_DB_PATH = "/data/awg-dash.db"
DB_PATH_ENV = "AWG_DASH_DB"
BUSY_TIMEOUT_MS = 5000
_WAL_PRAGMAS: tuple[str, ...] = (
"PRAGMA journal_mode = WAL",
"PRAGMA synchronous = NORMAL",
"PRAGMA foreign_keys = ON",
"PRAGMA busy_timeout = 5000",
)
def resolve_db_path(path: str | os.PathLike[str] | None = None) -> Path:
"""Определить путь к БД: аргумент -> ``AWG_DASH_DB`` -> ``/data/awg-dash.db``."""
if path is None:
path = os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH
return Path(path)
class Database:
"""Соединение SQLite: миграции, проверка схемы, транзакции."""
def __init__(
self,
path: str | os.PathLike[str] | None = None,
*,
auto_migrate: bool = True,
) -> None:
self.path = resolve_db_path(path)
self._lock = threading.RLock()
self._conn = self._open(self.path)
self.migration_report: MigrationReport | None = None
if auto_migrate:
self.migrate()
else:
verify_schema(self._conn)
# -- соединение -----------------------------------------------------
@staticmethod
def _open(path: Path) -> sqlite3.Connection:
if str(path) != ":memory:":
try:
path.parent.mkdir(parents=True, exist_ok=True)
except OSError as exc:
raise StorageError(f"не удалось создать каталог {path.parent}: {exc}") from exc
try:
conn = sqlite3.connect(
str(path),
isolation_level=None,
check_same_thread=False,
timeout=BUSY_TIMEOUT_MS / 1000,
)
conn.row_factory = sqlite3.Row
for pragma in _WAL_PRAGMAS:
conn.execute(pragma)
except sqlite3.Error as exc:
raise StorageError(f"не удалось открыть БД {path}: {exc}") from exc
return conn
@property
def connection(self) -> sqlite3.Connection:
"""Прямой доступ к соединению — только для однопоточного кода.
Рантайм API/сервисного слоя использует :meth:`read` (чтения) и
:meth:`transaction` (мутации): они берут общий лок и гарантируют
согласованность результата (#54).
"""
return self._conn
@property
def journal_mode(self) -> str:
with self._lock:
row = self._conn.execute("PRAGMA journal_mode").fetchone()
return str(row[0]) if row is not None else ""
@property
def is_wal(self) -> bool:
return self.journal_mode.lower() == "wal"
@property
def schema_version(self) -> int:
with self._lock:
return current_version(self._conn)
# -- схема ----------------------------------------------------------
def migrate(self) -> MigrationReport:
"""Применить неприменённые миграции и проверить схему."""
with self._lock:
report = migrate(self._conn)
self.migration_report = report
return report
def verify_schema(self) -> None:
"""Проверить соответствие фактической схемы ожидаемой версии."""
with self._lock:
verify_schema(self._conn)
# -- транзакции и чтения --------------------------------------------
@contextmanager
def transaction(self) -> Iterator[sqlite3.Connection]:
"""Транзакция ALL-OR-NOTHING (``BEGIN IMMEDIATE`` + rollback при ошибке)."""
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
yield self._conn
except BaseException:
self._conn.rollback()
raise
else:
self._conn.commit()
@contextmanager
def read(self) -> Iterator[sqlite3.Connection]:
"""Согласованное чтение под общим локом (#54).
Ждёт завершения чужой транзакции и потому не видит незафиксированных
строк и состояния, которое затем будет откачено. Внутри уже открытой
транзакции того же потока лок реентерабелен, поэтому читать можно и из
неё.
"""
with self._lock:
yield self._conn
def close(self) -> None:
with self._lock:
try:
self._conn.close()
except sqlite3.Error as exc: # pragma: no cover - защита от гонок при shutdown
raise StorageError(f"ошибка закрытия БД: {exc}") from exc
def __enter__(self) -> Database:
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
-33
View File
@@ -1,33 +0,0 @@
"""Иерархия ошибок хранилища awg-dash."""
from __future__ import annotations
class StorageError(RuntimeError):
"""Базовая ошибка слоя хранения состояния."""
class SchemaError(StorageError):
"""Схема БД не соответствует ожидаемой версии (или требует отката)."""
class ValidationError(StorageError):
"""Данные не прошли валидацию до применения (стадия validate)."""
def __init__(self, field: str, message: str) -> None:
super().__init__(f"{field}: {message}")
self.field = field
self.message = message
class ApplyError(StorageError):
"""Ошибка применения конфигурации.
``restored`` — предыдущее состояние подтверждённо восстановлено,
``audit_id`` — id записи в audit_log, статус ``failed``.
"""
def __init__(self, message: str, *, restored: bool, audit_id: int | None) -> None:
super().__init__(message)
self.restored = restored
self.audit_id = audit_id
-299
View File
@@ -1,299 +0,0 @@
"""Версионированные миграции схемы состояния (issue #20).
Схема БД управляется списком :class:`Migration`. Каждая миграция выполняется
в своей транзакции, запись версии в ``schema_migrations`` — там же: падение
миграции не оставляет «полуприменённой» схемы.
При старте приложения вызывается :func:`migrate` (создание/обновление схемы)
и :func:`verify_schema` (проверка, что фактическая схема соответствует
ожидаемой версии).
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from .clock import utcnow
from .errors import SchemaError, StorageError
SCHEMA_MIGRATIONS = "schema_migrations"
@dataclass(frozen=True, slots=True)
class Migration:
"""Одна версия схемы: имя и список DDL-выражений."""
version: int
name: str
statements: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class MigrationReport:
"""Результат :func:`migrate`."""
from_version: int
to_version: int
applied: tuple[int, ...]
@property
def changed(self) -> bool:
return bool(self.applied)
_V1_STATEMENTS: tuple[str, ...] = (
f"""
CREATE TABLE IF NOT EXISTS {SCHEMA_MIGRATIONS} (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS routing_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
kind TEXT NOT NULL CHECK (kind IN ('domain', 'cidr', 'ipset')),
match TEXT NOT NULL,
target TEXT NOT NULL,
priority INTEGER NOT NULL DEFAULT 100,
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS rule_exceptions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
match TEXT NOT NULL UNIQUE,
reason TEXT NOT NULL DEFAULT '',
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS routing_mode (
id INTEGER PRIMARY KEY CHECK (id = 1),
mode TEXT NOT NULL CHECK (mode IN ('off', 'split', 'full')),
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS geoip_schedule (
id INTEGER PRIMARY KEY CHECK (id = 1),
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
cron TEXT NOT NULL,
provider TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS configs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
version INTEGER NOT NULL DEFAULT 1,
path TEXT NOT NULL,
checksum TEXT NOT NULL DEFAULT '',
size_bytes INTEGER NOT NULL DEFAULT 0,
applied INTEGER NOT NULL DEFAULT 0 CHECK (applied IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mac TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
address TEXT NOT NULL DEFAULT '',
policy TEXT NOT NULL DEFAULT 'default',
last_seen TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
actor TEXT NOT NULL,
action TEXT NOT NULL,
entity TEXT NOT NULL,
entity_id TEXT,
old_value TEXT,
new_value TEXT,
status TEXT NOT NULL DEFAULT 'ok',
detail TEXT
)
""",
"CREATE INDEX IF NOT EXISTS idx_audit_log_entity ON audit_log (entity, entity_id)",
"CREATE INDEX IF NOT EXISTS idx_audit_log_ts ON audit_log (ts DESC)",
"""
CREATE TABLE IF NOT EXISTS state_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
entity TEXT NOT NULL,
entity_id TEXT NOT NULL,
created_at TEXT NOT NULL,
payload TEXT NOT NULL,
reason TEXT NOT NULL DEFAULT ''
)
""",
"CREATE INDEX IF NOT EXISTS idx_state_snapshots_key ON state_snapshots (entity, entity_id, id)",
)
V1_SEED_STATEMENTS: tuple[str, ...] = (
"INSERT OR IGNORE INTO routing_mode (id, mode, updated_at) VALUES (1, 'off',"
" strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
"INSERT OR IGNORE INTO geoip_schedule (id, enabled, cron, provider, updated_at) VALUES"
" (1, 1, '0 4 * * *', 'ripe', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
)
MIGRATIONS: tuple[Migration, ...] = (
Migration(
version=1,
name="initial-state-schema",
statements=(*_V1_STATEMENTS, *V1_SEED_STATEMENTS),
),
)
LATEST_VERSION: int = max(migration.version for migration in MIGRATIONS)
#: Ожидаемые таблицы схемы v1 и обязательные колонки каждой из них.
EXPECTED_TABLES: dict[str, frozenset[str]] = {
SCHEMA_MIGRATIONS: frozenset({"version", "name", "applied_at"}),
"routing_rules": frozenset(
{"id", "name", "kind", "match", "target", "priority", "enabled", "created_at", "updated_at"}
),
"rule_exceptions": frozenset({"id", "match", "reason", "enabled", "created_at", "updated_at"}),
"routing_mode": frozenset({"id", "mode", "updated_at"}),
"geoip_schedule": frozenset({"id", "enabled", "cron", "provider", "updated_at"}),
"configs": frozenset(
{
"id",
"name",
"version",
"path",
"checksum",
"size_bytes",
"applied",
"created_at",
"updated_at",
}
),
"devices": frozenset(
{"id", "mac", "name", "address", "policy", "last_seen", "created_at", "updated_at"}
),
"audit_log": frozenset(
{
"id",
"ts",
"actor",
"action",
"entity",
"entity_id",
"old_value",
"new_value",
"status",
"detail",
}
),
"state_snapshots": frozenset({"id", "entity", "entity_id", "created_at", "payload", "reason"}),
}
def table_names(conn: sqlite3.Connection) -> set[str]:
"""Имена таблиц, фактически присутствующих в БД."""
rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
return {str(row[0]) for row in rows}
def table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
"""Колонки таблицы (через параметризованное table-valued PRAGMA)."""
rows = conn.execute("SELECT name FROM pragma_table_info(?)", (table,)).fetchall()
return {str(row[0]) for row in rows}
def current_version(conn: sqlite3.Connection) -> int:
"""Текущая версия схемы; ``0`` — если таблицы версий ещё нет."""
if SCHEMA_MIGRATIONS not in table_names(conn):
return 0
row = conn.execute(f"SELECT MAX(version) FROM {SCHEMA_MIGRATIONS}").fetchone() # noqa: S608
if row is None or row[0] is None:
return 0
return int(row[0])
def applied_migrations(conn: sqlite3.Connection) -> dict[int, str]:
"""Карта «версия -> имя» для применённых миграций."""
if SCHEMA_MIGRATIONS not in table_names(conn):
return {}
rows = conn.execute(f"SELECT version, name FROM {SCHEMA_MIGRATIONS}").fetchall() # noqa: S608
return {int(row[0]): str(row[1]) for row in rows}
def apply_migration(conn: sqlite3.Connection, migration: Migration) -> None:
"""Применить одну миграцию в отдельной транзакции (all-or-nothing)."""
conn.execute("BEGIN IMMEDIATE")
try:
for statement in migration.statements:
conn.execute(statement)
conn.execute(
f"INSERT INTO {SCHEMA_MIGRATIONS} (version, name, applied_at) VALUES (?, ?, ?)", # noqa: S608
(migration.version, migration.name, utcnow()),
)
except sqlite3.Error as exc: # pragma: no cover - защита от порчи БД
conn.rollback()
raise StorageError(
f"миграция v{migration.version} ({migration.name}) не применена: {exc}"
) from exc
except Exception:
conn.rollback()
raise
conn.commit()
def migrate(
conn: sqlite3.Connection, *, migrations: tuple[Migration, ...] = MIGRATIONS
) -> MigrationReport:
"""Применить все неприменённые миграции по возрастанию версии."""
start = current_version(conn)
known = sorted(migrations, key=lambda migration: migration.version)
latest = max((migration.version for migration in known), default=0)
if start > latest:
raise SchemaError(
f"схема БД новее, чем известно приложению: v{start} > v{latest};"
" требуется обновление образа, не автооткат"
)
applied: list[int] = []
for migration in known:
if migration.version <= start:
continue
apply_migration(conn, migration)
applied.append(migration.version)
verify_schema(conn)
return MigrationReport(
from_version=start, to_version=current_version(conn), applied=tuple(applied)
)
def verify_schema(conn: sqlite3.Connection) -> None:
"""Проверить, что фактическая схема соответствует ожидаемой версии.
Вызывает :class:`SchemaError`, если версия схемы новее ожидаемой или если
отсутствует таблица/колонка, на которую рассчитывает приложение.
"""
version = current_version(conn)
if version > LATEST_VERSION:
raise SchemaError(f"схема БД v{version} новее ожидаемой v{LATEST_VERSION}")
if version < LATEST_VERSION:
raise SchemaError(f"схема БД v{version} не мигрирована до v{LATEST_VERSION}")
existing = table_names(conn)
missing_tables = sorted(set(EXPECTED_TABLES) - existing)
if missing_tables:
raise SchemaError(f"в схеме отсутствуют таблицы: {', '.join(missing_tables)}")
for table, required in EXPECTED_TABLES.items():
missing_columns = sorted(required - table_columns(conn, table))
if missing_columns:
raise SchemaError(f"таблица {table}: отсутствуют колонки {', '.join(missing_columns)}")
-186
View File
@@ -1,186 +0,0 @@
"""Модель данных хранилища awg-dash (issue #20).
Каждая сущность описана неизменяемым dataclass-ом; идентификатор ``None``
означает «ещё не сохранено». Строки времени — UTC ISO-8601, суффикс ``Z``.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
RULE_KINDS: tuple[str, ...] = ("domain", "cidr", "ipset")
DEVICE_POLICIES: tuple[str, ...] = ("default", "tunnel", "direct", "block")
AUDIT_STATUS_OK = "ok"
AUDIT_STATUS_FAILED = "failed"
AUDIT_STATUS_STAGED = "staged"
class Mode(StrEnum):
"""Режим маршрутизации шлюза."""
OFF = "off"
SPLIT = "split"
FULL = "full"
@classmethod
def values(cls) -> tuple[str, ...]:
return tuple(member.value for member in cls)
class Entity(StrEnum):
"""Имена сущностей для audit_log и снимков состояния."""
RULE = "rule"
EXCEPTION = "exception"
MODE = "mode"
GEOIP_SCHEDULE = "geoip_schedule"
CONFIG = "config"
DEVICE = "device"
APPLY = "apply"
@dataclass(frozen=True, slots=True)
class Rule:
"""Правило маршрутизации (domain / cidr / ipset -> target)."""
name: str
kind: str
match: str
target: str
priority: int = 100
enabled: bool = True
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"kind": self.kind,
"match": self.match,
"target": self.target,
"priority": self.priority,
"enabled": self.enabled,
}
@dataclass(frozen=True, slots=True)
class RuleException:
"""Исключение из правил (host/domain, который никогда не маршрутизируется в туннель)."""
match: str
reason: str = ""
enabled: bool = True
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"match": self.match,
"reason": self.reason,
"enabled": self.enabled,
}
@dataclass(frozen=True, slots=True)
class GeoIpSchedule:
"""Расписание обновления GeoIP-наборов."""
enabled: bool = True
cron: str = "0 4 * * *"
provider: str = "ripe"
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"enabled": self.enabled,
"cron": self.cron,
"provider": self.provider,
}
@dataclass(frozen=True, slots=True)
class ConfigMeta:
"""Метаданные сгенерированного конфига (сам файл живёт в /config)."""
name: str
path: str
version: int = 1
checksum: str = ""
size_bytes: int = 0
applied: bool = False
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"path": self.path,
"version": self.version,
"checksum": self.checksum,
"size_bytes": self.size_bytes,
"applied": self.applied,
}
@dataclass(frozen=True, slots=True)
class Device:
"""Известный клиент шлюза."""
mac: str
name: str
address: str = ""
policy: str = "default"
last_seen: str | None = None
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"mac": self.mac,
"name": self.name,
"address": self.address,
"policy": self.policy,
"last_seen": self.last_seen,
}
@dataclass(frozen=True, slots=True)
class AuditEntry:
"""Запись журнала «кто/когда/что/old->new»."""
id: int
ts: str
actor: str
action: str
entity: str
entity_id: str | None
old_value: Any
new_value: Any
status: str
detail: str | None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"ts": self.ts,
"actor": self.actor,
"action": self.action,
"entity": self.entity,
"entity_id": self.entity_id,
"old_value": self.old_value,
"new_value": self.new_value,
"status": self.status,
"detail": self.detail,
}
-599
View File
@@ -1,599 +0,0 @@
"""Репозиторий сущностей состояния (issue #20).
Каждая мутация выполняется в одной транзакции, запись в ``audit_log``
(«кто/когда/что/old→new») коммитится там же. Чтения идут через
:meth:`Database.read` — под тем же ``RLock``, что и транзакции (#54), поэтому
читатель не видит незафиксированных строк и состояния, которое затем будет
откачено, и не может получить перемешанный результат операторов другого потока.
"""
from __future__ import annotations
import sqlite3
from dataclasses import replace
from typing import cast
from . import audit, validation
from .clock import utcnow
from .db import Database
from .errors import StorageError
from .models import (
AuditEntry,
ConfigMeta,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
def _rule_from_row(row: sqlite3.Row) -> Rule:
return Rule(
id=int(row["id"]),
name=str(row["name"]),
kind=str(row["kind"]),
match=str(row["match"]),
target=str(row["target"]),
priority=int(row["priority"]),
enabled=bool(row["enabled"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _exception_from_row(row: sqlite3.Row) -> RuleException:
return RuleException(
id=int(row["id"]),
match=str(row["match"]),
reason=str(row["reason"]),
enabled=bool(row["enabled"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _config_from_row(row: sqlite3.Row) -> ConfigMeta:
return ConfigMeta(
id=int(row["id"]),
name=str(row["name"]),
path=str(row["path"]),
version=int(row["version"]),
checksum=str(row["checksum"]),
size_bytes=int(row["size_bytes"]),
applied=bool(row["applied"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _device_from_row(row: sqlite3.Row) -> Device:
return Device(
id=int(row["id"]),
mac=str(row["mac"]),
name=str(row["name"]),
address=str(row["address"]),
policy=str(row["policy"]),
last_seen=None if row["last_seen"] is None else str(row["last_seen"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _geoip_from_row(conn: sqlite3.Connection) -> GeoIpSchedule:
row = conn.execute(
"SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1"
).fetchone()
if row is None:
return GeoIpSchedule()
return GeoIpSchedule(
enabled=bool(row["enabled"]),
cron=str(row["cron"]),
provider=str(row["provider"]),
updated_at=str(row["updated_at"]),
)
class Store:
"""Доступ к состоянию: чтение + аудируемые мутации."""
def __init__(self, db: Database) -> None:
self._db = db
@property
def database(self) -> Database:
return self._db
# -- чтение ---------------------------------------------------------
# Чтения идут через Database.read() под тем же RLock, что транзакции
# (#54): это исключает перемешивание операторов разных потоков и грязные
# чтения незафиксированных строк.
def get_mode(self) -> Mode:
with self._db.read() as conn:
row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
if row is None:
return Mode.OFF
return Mode(str(row["mode"]))
def get_geoip_schedule(self) -> GeoIpSchedule:
with self._db.read() as conn:
return _geoip_from_row(conn)
def list_rules(self, *, enabled_only: bool = False) -> list[Rule]:
sql = "SELECT * FROM routing_rules"
if enabled_only:
sql += " WHERE enabled = 1"
sql += " ORDER BY priority, id"
with self._db.read() as conn:
rows = conn.execute(sql).fetchall()
return [_rule_from_row(row) for row in rows]
def get_rule(self, rule_id: int) -> Rule | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
return None if row is None else _rule_from_row(row)
def get_rule_by_name(self, name: str) -> Rule | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
return None if row is None else _rule_from_row(row)
def list_exceptions(self) -> list[RuleException]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall()
return [_exception_from_row(row) for row in rows]
def get_exception(self, match: str) -> RuleException | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
return None if row is None else _exception_from_row(row)
def list_configs(self) -> list[ConfigMeta]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM configs ORDER BY name").fetchall()
return [_config_from_row(row) for row in rows]
def get_config(self, name: str) -> ConfigMeta | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
return None if row is None else _config_from_row(row)
def list_devices(self) -> list[Device]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM devices ORDER BY name").fetchall()
return [_device_from_row(row) for row in rows]
def get_device(self, mac: str) -> Device | None:
normalized = validation.normalize_mac(mac)
with self._db.read() as conn:
row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
return None if row is None else _device_from_row(row)
def list_audit(
self, *, entity: str | None = None, entity_id: str | int | None = None, limit: int = 100
) -> list[AuditEntry]:
with self._db.read() as conn:
return audit.list_entries(conn, entity=entity, entity_id=entity_id, limit=limit)
# -- мутации: правила ------------------------------------------------
def upsert_rule(self, rule: Rule, *, actor: str) -> Rule:
"""Создать или обновить правило (по ``id`` либо по уникальному ``name``)."""
validation.validate_name("name", rule.name)
kind = validation.validate_rule_kind(rule.kind)
match = validation.validate_match(kind, rule.match)
target = validation.validate_target(rule.target)
priority = validation.validate_priority(rule.priority)
now = utcnow()
with self._db.transaction() as conn:
existing = self._find_rule_row(conn, rule.id, rule.name)
core: tuple[str, str, str, str, int, int] = (
rule.name,
kind,
match,
target,
priority,
1 if rule.enabled else 0,
)
if existing is None:
cursor = conn.execute(
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(*core, now, now),
)
new_id = int(cursor.lastrowid or 0)
saved = replace(
rule,
id=new_id,
kind=kind,
match=match,
target=target,
priority=priority,
created_at=now,
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.RULE.value,
entity_id=new_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
rule_id = int(existing["id"])
conn.execute(
"UPDATE routing_rules SET name = ?, kind = ?, match = ?, target = ?,"
" priority = ?, enabled = ?, updated_at = ? WHERE id = ?",
(*core, now, rule_id),
)
saved = replace(
rule,
id=rule_id,
kind=kind,
match=match,
target=target,
priority=priority,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.RULE.value,
entity_id=rule_id,
old_value=_rule_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_rule(self, rule_id: int, *, actor: str) -> bool:
"""Удалить правило; ``False`` — если правила не было."""
with self._db.transaction() as conn:
row = self._find_rule_row(conn, rule_id, None)
if row is None:
return False
removed = _rule_from_row(row)
conn.execute("DELETE FROM routing_rules WHERE id = ?", (int(row["id"]),))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.RULE.value,
entity_id=int(row["id"]),
old_value=removed.as_dict(),
new_value=None,
)
return True
@staticmethod
def _find_rule_row(
conn: sqlite3.Connection, rule_id: int | None, name: str | None
) -> sqlite3.Row | None:
if rule_id is not None:
row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
return cast("sqlite3.Row | None", row)
if name is not None:
row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
return cast("sqlite3.Row | None", row)
return None
# -- мутации: исключения ---------------------------------------------
def upsert_exception(self, exception: RuleException, *, actor: str) -> RuleException:
match = validation.validate_exception_match(exception.match)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute(
"SELECT * FROM rule_exceptions WHERE match = ?", (match,)
).fetchone()
if existing is None:
conn.execute(
"INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)"
" VALUES (?, ?, ?, ?, ?)",
(match, exception.reason, 1 if exception.enabled else 0, now, now),
)
else:
conn.execute(
"UPDATE rule_exceptions SET reason = ?, enabled = ?, updated_at = ?"
" WHERE match = ?",
(exception.reason, 1 if exception.enabled else 0, now, match),
)
row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
if row is None: # pragma: no cover - невозможен после успешной записи
raise StorageError("правило-исключение исчезло сразу после записи")
saved = _exception_from_row(row)
audit.record(
conn,
actor=actor,
action="create" if existing is None else "update",
entity=Entity.EXCEPTION.value,
entity_id=saved.id,
old_value=None if existing is None else _exception_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_exception(self, exception_id: int, *, actor: str) -> bool:
with self._db.transaction() as conn:
row = conn.execute(
"SELECT * FROM rule_exceptions WHERE id = ?", (exception_id,)
).fetchone()
if row is None:
return False
removed = _exception_from_row(row)
conn.execute("DELETE FROM rule_exceptions WHERE id = ?", (exception_id,))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.EXCEPTION.value,
entity_id=exception_id,
old_value=removed.as_dict(),
new_value=None,
)
return True
# -- мутации: режим и GeoIP ------------------------------------------
def set_mode(self, mode: str | Mode, *, actor: str) -> Mode:
"""Установить режим маршрутизации; в журнал пишется old→new."""
validated = validation.validate_mode(mode)
with self._db.transaction() as conn:
previous = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
old = "off" if previous is None else str(previous["mode"])
now = utcnow()
conn.execute(
"INSERT INTO routing_mode (id, mode, updated_at) VALUES (1, ?, ?)"
" ON CONFLICT(id) DO UPDATE SET mode = excluded.mode,"
" updated_at = excluded.updated_at",
(validated.value, now),
)
audit.record(
conn,
actor=actor,
action="set_mode",
entity=Entity.MODE.value,
entity_id=1,
old_value={"mode": old},
new_value={"mode": validated.value},
)
return validated
def set_geoip_schedule(
self,
*,
enabled: bool | None = None,
cron: str | None = None,
provider: str | None = None,
actor: str,
) -> GeoIpSchedule:
"""Обновить расписание GeoIP (частичное обновление) и записать аудит."""
now = utcnow()
with self._db.transaction() as conn:
# чтение old_value — внутри той же транзакции, что и запись (#55 D-2):
# иначе при конкурентном обновлении журнал заявит устаревшее «старое».
current = _geoip_from_row(conn)
stored = GeoIpSchedule(
enabled=current.enabled if enabled is None else enabled,
cron=validation.validate_cron(current.cron if cron is None else cron),
provider=validation.validate_provider(
current.provider if provider is None else provider
),
updated_at=now,
)
conn.execute(
"INSERT INTO geoip_schedule (id, enabled, cron, provider, updated_at)"
" VALUES (1, ?, ?, ?, ?)"
" ON CONFLICT(id) DO UPDATE SET enabled = excluded.enabled, cron = excluded.cron,"
" provider = excluded.provider, updated_at = excluded.updated_at",
(1 if stored.enabled else 0, stored.cron, stored.provider, now),
)
audit.record(
conn,
actor=actor,
action="set_geoip_schedule",
entity=Entity.GEOIP_SCHEDULE.value,
entity_id=1,
old_value=current.as_dict(),
new_value=stored.as_dict(),
)
return stored
# -- мутации: конфиги -------------------------------------------------
def upsert_config(self, config: ConfigMeta, *, actor: str) -> ConfigMeta:
name = validation.validate_name("name", config.name)
path = validation.validate_config_path(config.path)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
if existing is None:
cursor = conn.execute(
"INSERT INTO configs (name, version, path, checksum, size_bytes, applied,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
name,
config.version,
path,
config.checksum,
config.size_bytes,
1 if config.applied else 0,
now,
now,
),
)
config_id = int(cursor.lastrowid or 0)
saved = replace(
config, id=config_id, name=name, path=path, created_at=now, updated_at=now
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.CONFIG.value,
entity_id=config_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
config_id = int(existing["id"])
conn.execute(
"UPDATE configs SET version = ?, path = ?, checksum = ?, size_bytes = ?,"
" applied = ?, updated_at = ? WHERE id = ?",
(
config.version,
path,
config.checksum,
config.size_bytes,
1 if config.applied else 0,
now,
config_id,
),
)
saved = replace(
config,
id=config_id,
name=name,
path=path,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.CONFIG.value,
entity_id=config_id,
old_value=_config_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def mark_config_applied(self, name: str, *, applied: bool, actor: str) -> ConfigMeta:
"""Отметить конфиг применённым/снятым и записать аудит."""
with self._db.transaction() as conn:
row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
if row is None:
raise StorageError(f"конфиг {name!r} не найден")
now = utcnow()
conn.execute(
"UPDATE configs SET applied = ?, updated_at = ? WHERE id = ?",
(1 if applied else 0, now, int(row["id"])),
)
updated = replace(_config_from_row(row), applied=applied, updated_at=now)
audit.record(
conn,
actor=actor,
action="set_applied",
entity=Entity.CONFIG.value,
entity_id=int(row["id"]),
old_value=_config_from_row(row).as_dict(),
new_value=updated.as_dict(),
)
return updated
# -- мутации: устройства ----------------------------------------------
def register_device(self, device: Device, *, actor: str) -> Device:
mac = validation.normalize_mac(device.mac)
name = validation.validate_name("name", device.name)
policy = validation.validate_device_policy(device.policy)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute("SELECT * FROM devices WHERE mac = ?", (mac,)).fetchone()
if existing is None:
cursor = conn.execute(
"INSERT INTO devices (mac, name, address, policy, last_seen, created_at,"
" updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(mac, name, device.address, policy, device.last_seen, now, now),
)
device_id = int(cursor.lastrowid or 0)
saved = replace(
device,
id=device_id,
mac=mac,
name=name,
policy=policy,
created_at=now,
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.DEVICE.value,
entity_id=device_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
device_id = int(existing["id"])
conn.execute(
"UPDATE devices SET name = ?, address = ?, policy = ?, last_seen = ?,"
" updated_at = ? WHERE id = ?",
(name, device.address, policy, device.last_seen, now, device_id),
)
saved = replace(
device,
id=device_id,
mac=mac,
name=name,
policy=policy,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.DEVICE.value,
entity_id=device_id,
old_value=_device_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_device(self, mac: str, *, actor: str) -> bool:
normalized = validation.normalize_mac(mac)
with self._db.transaction() as conn:
row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
if row is None:
return False
removed = _device_from_row(row)
conn.execute("DELETE FROM devices WHERE mac = ?", (normalized,))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.DEVICE.value,
entity_id=removed.id,
old_value=removed.as_dict(),
new_value=None,
)
return True
# -- служебное --------------------------------------------------------
def counts(self) -> dict[str, int]:
"""Количество строк в изменяемых таблицах (для отчётов и тестов)."""
tables = {
"rules": "routing_rules",
"exceptions": "rule_exceptions",
"configs": "configs",
"devices": "devices",
"audit": "audit_log",
"snapshots": "state_snapshots",
}
result: dict[str, int] = {}
with self._db.read() as conn: # один снимок на все таблицы: счётчики согласованы
for key, table in tables.items():
row = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone() # noqa: S608
result[key] = int(row[0]) if row is not None else 0
return result
def raw(self) -> sqlite3.Connection:
"""Соединение для кода, которому нужен доступ ниже уровня репозитория.
Лок не берётся: в рантайме используйте :meth:`Database.read` /
:meth:`Database.transaction` (#54), иначе возможны гонки на общем
соединении. Метод оставлен для однопоточного кода и миграций.
"""
return self._db.connection
-47
View File
@@ -1,47 +0,0 @@
"""Точка входа слоя хранения: открыть БД, применить миграции, собрать движок."""
from __future__ import annotations
import os
from dataclasses import dataclass
from .apply import Applier, ApplyEngine
from .db import Database
from .repository import Store
from .state import StateSnapshot
@dataclass(slots=True)
class Storage:
"""Собранный слой хранения для приложения."""
db: Database
store: Store
engine: ApplyEngine
def close(self) -> None:
self.db.close()
def __enter__(self) -> Storage:
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
def open_storage(
path: str | os.PathLike[str] | None = None,
*,
applier: Applier | None = None,
actor: str = "system",
auto_migrate: bool = True,
) -> Storage:
"""Открыть хранилище: на пустом ``/data`` схема поднимается автоматически."""
db = Database(path, auto_migrate=auto_migrate)
store = Store(db)
return Storage(db=db, store=store, engine=ApplyEngine(store, applier, actor=actor))
def rollback_to(engine: ApplyEngine, plan_id: int, *, reason: str = "") -> StateSnapshot:
"""Явный откат состояния к снимку плана (тонкая обёртка для API)."""
return engine.rollback(plan_id, reason=reason)
-174
View File
@@ -1,174 +0,0 @@
"""Снимки состояния и восстановление после отката (rollback, issue #20).
Снимок — это JSON-копия состояния (режим, расписание GeoIP, правила,
исключения) на момент перед применением. Снимок сохраняется ДО мутации,
поэтому даже при неудачном apply (внешний iptables упал) есть источник для
восстановления.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass
from typing import Any
from . import audit
from .clock import utcnow
from .errors import StorageError
from .models import Entity
@dataclass(frozen=True, slots=True)
class StateSnapshot:
"""Сохранённый снимок состояния."""
id: int
entity: str
entity_id: str
created_at: str
payload: dict[str, Any]
reason: str
def capture_state(conn: sqlite3.Connection, *, include_exceptions: bool = True) -> dict[str, Any]:
"""Снять текущее состояние в виде JSON-совместимого словаря."""
mode_row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
geoip_row = conn.execute(
"SELECT enabled, cron, provider FROM geoip_schedule WHERE id = 1"
).fetchone()
rules = [
{
"name": str(row["name"]),
"kind": str(row["kind"]),
"match": str(row["match"]),
"target": str(row["target"]),
"priority": int(row["priority"]),
"enabled": bool(row["enabled"]),
}
for row in conn.execute(
"SELECT name, kind, match, target, priority, enabled FROM routing_rules"
" ORDER BY priority, id"
).fetchall()
]
payload: dict[str, Any] = {
"mode": "off" if mode_row is None else str(mode_row["mode"]),
"geoip": {
"enabled": True if geoip_row is None else bool(geoip_row["enabled"]),
"cron": "0 4 * * *" if geoip_row is None else str(geoip_row["cron"]),
"provider": "ripe" if geoip_row is None else str(geoip_row["provider"]),
},
"rules": rules,
}
if include_exceptions:
payload["exceptions"] = [
{
"match": str(row["match"]),
"reason": str(row["reason"]),
"enabled": bool(row["enabled"]),
}
for row in conn.execute(
"SELECT match, reason, enabled FROM rule_exceptions ORDER BY id"
).fetchall()
]
return payload
def save_snapshot(
conn: sqlite3.Connection,
payload: dict[str, Any],
*,
entity: str = Entity.APPLY.value,
entity_id: str = "state",
reason: str = "",
) -> int:
"""Сохранить снимок состояния и вернуть идентификатор."""
cursor = conn.execute(
"INSERT INTO state_snapshots (entity, entity_id, created_at, payload, reason)"
" VALUES (?, ?, ?, ?, ?)",
(entity, entity_id, utcnow(), json.dumps(payload, sort_keys=True), reason),
)
return int(cursor.lastrowid or 0)
def load_snapshot(conn: sqlite3.Connection, snapshot_id: int) -> StateSnapshot:
"""Прочитать снимок по id."""
row = conn.execute("SELECT * FROM state_snapshots WHERE id = ?", (snapshot_id,)).fetchone()
if row is None:
raise StorageError(f"снимок состояния {snapshot_id} не найден")
return StateSnapshot(
id=int(row["id"]),
entity=str(row["entity"]),
entity_id=str(row["entity_id"]),
created_at=str(row["created_at"]),
payload=json.loads(str(row["payload"])),
reason=str(row["reason"]),
)
def restore_state(conn: sqlite3.Connection, payload: dict[str, Any]) -> None:
"""Полностью заменить изменяемые таблицы состоянием из снимка."""
conn.execute("DELETE FROM routing_rules")
for rule in payload.get("rules", []):
conn.execute(
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
rule["name"],
rule["kind"],
rule["match"],
rule["target"],
int(rule["priority"]),
1 if rule["enabled"] else 0,
utcnow(),
utcnow(),
),
)
if "exceptions" in payload:
conn.execute("DELETE FROM rule_exceptions")
for exception in payload["exceptions"]:
conn.execute(
"INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)"
" VALUES (?, ?, ?, ?, ?)",
(
exception["match"],
exception.get("reason", ""),
1 if exception["enabled"] else 0,
utcnow(),
utcnow(),
),
)
geoip = payload.get("geoip", {})
conn.execute(
"UPDATE routing_mode SET mode = ?, updated_at = ? WHERE id = 1",
(payload.get("mode", "off"), utcnow()),
)
conn.execute(
"UPDATE geoip_schedule SET enabled = ?, cron = ?, provider = ?,"
" updated_at = ? WHERE id = 1",
(
1 if geoip.get("enabled", True) else 0,
geoip.get("cron", "0 4 * * *"),
geoip.get("provider", "ripe"),
utcnow(),
),
)
def rollback_to_snapshot(
conn: sqlite3.Connection, snapshot_id: int, *, actor: str, reason: str = ""
) -> StateSnapshot:
"""Восстановить состояние из снимка и записать это в журнал аудита."""
snapshot = load_snapshot(conn, snapshot_id)
audit.record(
conn,
actor=actor,
action="rollback",
entity=snapshot.entity,
entity_id=snapshot.entity_id,
old_value=capture_state(conn),
new_value=snapshot.payload,
detail=reason or f"rollback к снимку #{snapshot.id}",
)
restore_state(conn, snapshot.payload)
return snapshot
-131
View File
@@ -1,131 +0,0 @@
"""Чистые функции валидации состояния (стадия ``validate`` перед apply, issue #20).
Валидация отделена от БД: те же функции вызывают и репозиторий (точечные
мутации), и движок атомарного применения (``validate -> stage -> apply``).
"""
from __future__ import annotations
import ipaddress
import re
from .errors import ValidationError
from .models import DEVICE_POLICIES, RULE_KINDS, Mode
_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_.:-]{0,62}$")
_DOMAIN_RE = re.compile(r"^(\*\.)?([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$")
_IPSET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$")
_MAC_RE = re.compile(r"^(?:[0-9a-f]{2}:){5}[0-9a-f]{2}$")
_CRON_FIELD_RE = re.compile(r"^[0-9*,/-]{1,20}$")
_PROVIDER_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,31}$")
def normalize_mac(mac: str) -> str:
"""Привести MAC к канонической форме ``aa:bb:cc:dd:ee:ff``."""
cleaned = mac.strip().replace("-", ":").lower()
if len(cleaned) == 12 and ":" not in cleaned:
cleaned = ":".join(cleaned[i : i + 2] for i in range(0, 12, 2))
if not _MAC_RE.match(cleaned):
raise ValidationError("mac", f"некорректный MAC-адрес: {mac!r}")
return cleaned
def validate_name(field: str, value: str) -> str:
"""Проверить идентификатор (имя правила/конфига устройства)."""
if not _NAME_RE.match(value.strip()):
raise ValidationError(
field, f"допустимы строчные латинские буквы, цифры и «_.:-», до 63 символов: {value!r}"
)
return value.strip()
def validate_rule_kind(kind: str) -> str:
if kind not in RULE_KINDS:
raise ValidationError("kind", f"ожидается одно из {RULE_KINDS}, получено {kind!r}")
return kind
def validate_match(kind: str, match: str) -> str:
"""Проверить образец правила по типу записи."""
validate_rule_kind(kind)
value = match.strip()
if not value:
raise ValidationError("match", "образец правила не может быть пустым")
if kind == "domain":
if not _DOMAIN_RE.match(value):
raise ValidationError("match", f"некорректный домен: {value!r}")
elif kind == "cidr":
try:
ipaddress.ip_network(value, strict=False)
except ValueError as exc:
raise ValidationError("match", f"некорректная сеть CIDR: {value!r}") from exc
elif not _IPSET_RE.match(value):
raise ValidationError("match", f"некорректное имя ipset: {value!r}")
return value
def validate_exception_match(match: str) -> str:
"""Проверить образец исключения: домен, имя ipset или сеть CIDR."""
value = match.strip()
if not value:
raise ValidationError("match", "образец исключения не может быть пустым")
if _DOMAIN_RE.match(value) or _IPSET_RE.match(value):
return value
try:
ipaddress.ip_network(value, strict=False)
except ValueError as exc:
raise ValidationError(
"match", f"ожидается домен, сеть CIDR или имя ipset: {match!r}"
) from exc
return value
def validate_target(target: str) -> str:
value = target.strip()
if not _IPSET_RE.match(value):
raise ValidationError("target", f"некорректная цель маршрутизации: {target!r}")
return value
def validate_priority(priority: int) -> int:
if not isinstance(priority, int) or isinstance(priority, bool):
raise ValidationError("priority", f"ожидается целое число, получено {priority!r}")
if not 0 <= priority <= 65535:
raise ValidationError("priority", f"вне диапазона 0..65535: {priority}")
return priority
def validate_mode(mode: str | Mode) -> Mode:
try:
return Mode(mode)
except ValueError as exc:
raise ValidationError(
"mode", f"ожидается одно из {Mode.values()}, получено {mode!r}"
) from exc
def validate_cron(cron: str) -> str:
fields = cron.split()
if len(fields) != 5 or not all(_CRON_FIELD_RE.match(part) for part in fields):
raise ValidationError("cron", f"ожидается 5-полевое cron-выражение: {cron!r}")
return " ".join(fields)
def validate_provider(provider: str) -> str:
value = provider.strip().lower()
if not _PROVIDER_RE.match(value):
raise ValidationError("provider", f"некорректное имя провайдера GeoIP: {provider!r}")
return value
def validate_device_policy(policy: str) -> str:
if policy not in DEVICE_POLICIES:
raise ValidationError("policy", f"ожидается одно из {DEVICE_POLICIES}, получено {policy!r}")
return policy
def validate_config_path(path: str) -> str:
value = path.strip()
if not value.startswith("/") or ".." in value.split("/"):
raise ValidationError("path", f"ожидается абсолютный путь без «..»: {path!r}")
return value
-1
View File
@@ -1 +0,0 @@
"""WebSocket-поток метрик."""
-47
View File
@@ -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
-48
View File
@@ -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
-110
View File
@@ -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)
-321
View File
@@ -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
-186
View File
@@ -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}"
-13
View File
@@ -1,13 +0,0 @@
"""Скелетные тесты: пакеты импортируются, версия объявлена."""
import awg_agent
import awg_api
def test_versions_match() -> None:
assert awg_agent.__version__ == awg_api.__version__
def test_public_api_declared() -> None:
assert "__version__" in awg_agent.__all__
assert "__version__" in awg_api.__all__
-311
View File
@@ -1,311 +0,0 @@
"""Тесты атомарного применения: validate -> stage -> apply -> rollback (issue #20)."""
from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from typing import Any
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
ApplyEngine,
ApplyError,
Database,
DesiredState,
Entity,
GeoIpSchedule,
Mode,
NullApplier,
Rule,
RuleException,
Store,
ValidationError,
capture_state,
)
class MockIptables:
"""Мок внешнего применителя: умеет «падать» как реальный iptables."""
def __init__(self, *, fail_apply: bool = False, fail_revert: bool = False) -> None:
self.applied: list[dict[str, Any]] = []
self.reverted: list[dict[str, Any]] = []
self.fail_apply = fail_apply
self.fail_revert = fail_revert
def apply(self, payload: Mapping[str, Any]) -> None:
if self.fail_apply:
self.applied.append(dict(payload))
raise RuntimeError("mock iptables: exit 4")
self.applied.append(dict(payload))
def revert(self, payload: Mapping[str, Any]) -> None:
if self.fail_revert:
raise RuntimeError("mock iptables: revert failed")
self.reverted.append(dict(payload))
def make_engine(
tmp_path: Path, applier: MockIptables | None = None
) -> tuple[Database, Store, ApplyEngine, MockIptables]:
db = Database(tmp_path / "data" / "awg-dash.db")
store = Store(db)
mock = applier if applier is not None else MockIptables()
return db, store, ApplyEngine(store, mock, actor="ui:admin"), mock
def desired(*rules: Rule, mode: Mode = Mode.SPLIT) -> DesiredState:
return DesiredState(mode=mode, rules=rules, geoip=GeoIpSchedule())
# -- validate ------------------------------------------------------------------
def test_validate_accepts_good_state(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
state = desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"))
assert engine.validate(state) is state
def test_validate_rejects_bad_state(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
cases: list[DesiredState] = [
DesiredState(mode="turbo"), # type: ignore[arg-type]
desired(Rule(name="dup", kind="domain", match="a.example.com", target="tun0"),
Rule(name="dup", kind="domain", match="b.example.com", target="tun0")),
desired(Rule(name="x", kind="cidr", match="10.0.0.0/99", target="tun0")),
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="всякий раз")),
DesiredState(
mode=Mode.SPLIT,
rules=(),
geoip=GeoIpSchedule(),
exceptions=(RuleException(match="!!!"),),
),
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(provider="!")),
]
for case in cases:
with pytest.raises(ValidationError):
engine.validate(case)
def test_validate_does_not_touch_database(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
with pytest.raises(ValidationError):
engine.apply_payload({"mode": "split", "rules": [{"name": "x", "kind": "cidr",
"match": "bad", "target": "tun0"}]})
assert store.counts() == {
"rules": 0,
"exceptions": 0,
"configs": 0,
"devices": 0,
"audit": 0,
"snapshots": 0,
}
# -- stage ---------------------------------------------------------------------
def test_stage_saves_snapshot_and_staged_audit(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"),
actor="system")
store.set_mode(Mode.OFF, actor="system")
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0")), reason="тест")
assert plan.previous == before
assert plan.desired.mode == Mode.SPLIT
# состояние ещё не изменено
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["old"]
snapshots = store.counts()["snapshots"]
assert snapshots == 1
staged = store.list_audit(entity=Entity.APPLY.value)[0]
assert staged.status == AUDIT_STATUS_STAGED
assert staged.action == "stage"
assert staged.detail == "тест"
assert staged.old_value == before
def test_stage_rejects_invalid_state_without_snapshot(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
with pytest.raises(ValidationError):
engine.stage(DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="bad")))
assert store.counts()["snapshots"] == 0
# -- apply ---------------------------------------------------------------------
def test_apply_writes_state_and_audit(tmp_path: Path) -> None:
db, store, engine, mock = make_engine(tmp_path)
with db:
plan = engine.stage(desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
mode=Mode.FULL), reason="включение")
result = engine.apply(plan)
assert result.ok is True
assert result.plan_id == plan.plan_id
assert result.audit_id > 0
assert store.get_mode() == Mode.FULL
assert [rule.name for rule in store.list_rules()] == ["yt"]
assert mock.applied == [plan.desired.to_payload()]
entry = store.list_audit(entity=Entity.APPLY.value)[0]
assert entry.status == AUDIT_STATUS_OK
assert entry.action == "apply"
assert entry.old_value == plan.previous
assert entry.new_value == plan.desired.to_payload()
def test_failed_apply_restores_previous_state(tmp_path: Path) -> None:
"""GIVEN ошибочный apply (мок падения iptables), THEN состояние прежнее + ApplyError."""
db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True))
with db:
store.upsert_rule(Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system")
store.set_mode(Mode.SPLIT, actor="system")
store.set_geoip_schedule(cron="0 5 * * *", actor="system")
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0"), mode=Mode.FULL), reason="сломается")
with pytest.raises(ApplyError) as excinfo:
engine.apply(plan)
error = excinfo.value
assert error.restored is True
assert error.audit_id is not None and error.audit_id > 0
# состояние, а не «частично применённая конфигурация»
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["keep"]
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().cron == "0 5 * * *"
assert mock.reverted == [before]
failed = store.list_audit(entity=Entity.APPLY.value)[0]
assert failed.status == AUDIT_STATUS_FAILED
assert failed.id == error.audit_id
assert failed.new_value == before
assert "mock iptables" in (failed.detail or "")
def test_failed_apply_with_failing_revert_still_restores_db(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(
tmp_path, MockIptables(fail_apply=True, fail_revert=True)
)
with db:
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="x", kind="domain", match="x.example.com", target="tun0")))
with pytest.raises(ApplyError) as excinfo:
engine.apply(plan)
assert excinfo.value.restored is True
assert capture_state(db.connection) == before
def test_apply_can_succeed_after_failed_apply(tmp_path: Path) -> None:
db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True))
with db:
bad = engine.stage(desired(Rule(name="bad", kind="domain", match="bad.example.com",
target="tun0")))
with pytest.raises(ApplyError):
engine.apply(bad)
mock.fail_apply = False
good_rule = Rule(name="good", kind="domain", match="good.example.com", target="tun0")
result = engine.apply(engine.stage(desired(good_rule, mode=Mode.FULL)))
assert result.ok is True
assert [rule.name for rule in store.list_rules()] == ["good"]
assert store.get_mode() == Mode.FULL
# -- rollback ------------------------------------------------------------------
def test_rollback_returns_previous_state(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"),
actor="ui:admin")
before = capture_state(db.connection)
result = engine.apply(
engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0"), mode=Mode.FULL), reason="применение"),
)
assert result.ok is True
snapshot = engine.rollback(result.plan_id, reason="откат оператором")
assert snapshot.id == result.plan_id
assert snapshot.payload == before
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["old"]
assert store.get_mode() == Mode.OFF
rollback_entry = store.list_audit(entity=Entity.APPLY.value)[0]
assert rollback_entry.action == "rollback"
assert rollback_entry.detail == "откат оператором"
assert rollback_entry.new_value == before
def test_apply_payload_sugar_and_null_applier(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
result = engine.apply_payload(
{
"mode": "split",
"rules": [{"name": "yt", "kind": "domain", "match": "youtube.com", "target": "tun0"}],
"geoip": {"enabled": False, "cron": "0 3 * * *", "provider": "ripe"},
"exceptions": [{"match": "bank.example.com", "reason": "банк"}],
},
reason="api: применить",
)
assert result.ok is True
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().enabled is False
assert store.list_exceptions()[0].match == "bank.example.com"
engine_null = ApplyEngine(store, NullApplier(), actor="system")
assert engine_null.apply_payload({"mode": "off"}).ok is True
assert store.get_mode() == Mode.OFF
def test_desired_state_payload_roundtrip() -> None:
state = DesiredState(
mode=Mode.FULL,
rules=(Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=7),),
geoip=GeoIpSchedule(enabled=False, cron="0 1 * * *", provider="maxmind"),
exceptions=(RuleException(match="bank.example.com", reason="банк", enabled=False),),
)
restored = DesiredState.from_payload(state.to_payload())
assert restored.mode == Mode.FULL
assert restored.rules == state.rules
assert restored.geoip == state.geoip
assert restored.exceptions == state.exceptions
@pytest.mark.parametrize(
"payload",
[
{"rules": "не список"},
{"rules": [{"kind": "domain", "match": "a.example.com", "target": "tun0"}]},
{"rules": ["строка"]},
{"exceptions": "не список"},
{"exceptions": [{"reason": "нет match"}]},
{"geoip": "не объект"},
{"mode": "turbo"},
],
)
def test_desired_state_from_payload_rejects_bad_payload(payload: dict[str, Any]) -> None:
with pytest.raises(ValidationError):
DesiredState.from_payload(payload)
def test_apply_engine_requires_store(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
assert engine.validate(desired()) == desired()
-145
View File
@@ -1,145 +0,0 @@
"""Тесты журнала аудита «кто/когда/что/old→new» (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
Database,
Entity,
Mode,
Rule,
Store,
)
from awg_api.store import audit
def make_db(tmp_path: Path) -> Database:
return Database(tmp_path / "data" / "awg-dash.db")
def test_record_and_read_back(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
entry_id = audit.record(
conn,
actor="ui:admin",
action="update",
entity=Entity.RULE.value,
entity_id=7,
old_value={"match": "old.example.com"},
new_value={"match": "new.example.com"},
)
assert entry_id > 0
entries = audit.list_entries(db.connection)
assert len(entries) == 1
entry = entries[0]
assert entry.actor == "ui:admin"
assert entry.entity == "rule"
assert entry.entity_id == "7"
assert entry.old_value == {"match": "old.example.com"}
assert entry.new_value == {"match": "new.example.com"}
assert entry.status == AUDIT_STATUS_OK
assert entry.ts.endswith("Z")
assert entry.as_dict()["entity_id"] == "7"
def test_actor_is_required(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn, pytest.raises(ValueError, match="actor"):
audit.record(
conn,
actor="",
action="update",
entity=Entity.RULE.value,
new_value={"x": 1},
)
def test_old_and_new_can_be_absent(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(conn, actor="system", action="noop", entity=Entity.APPLY.value)
entry = audit.last_entry(db.connection)
assert entry is not None
assert entry.old_value is None
assert entry.new_value is None
assert entry.entity_id is None
assert entry.detail is None
def test_filters_limit_and_ordering(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
store = Store(db)
store.set_mode(Mode.SPLIT, actor="ui:admin")
store.set_mode(Mode.FULL, actor="ui:admin")
store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
mode_entries = audit.list_entries(db.connection, entity=Entity.MODE.value)
assert len(mode_entries) == 2
assert mode_entries[0].new_value == {"mode": "full"} # свежие первыми
assert mode_entries[0].old_value == {"mode": "split"}
by_id = audit.list_entries(
db.connection, entity=Entity.MODE.value, entity_id=1, limit=1
)
assert len(by_id) == 1
assert audit.count(db.connection) == 3
assert audit.count(db.connection, entity=Entity.RULE.value) == 1
assert audit.last_entry(db.connection, entity=Entity.RULE.value) is not None
def test_ts_can_be_overridden(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(
conn,
actor="system",
action="stage",
entity=Entity.APPLY.value,
ts="2026-01-02T03:04:05.678Z",
)
entry = audit.last_entry(db.connection)
assert entry is not None and entry.ts == "2026-01-02T03:04:05.678Z"
def test_scalar_and_list_payloads_roundtrip(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(
conn,
actor="system",
action="apply",
entity=Entity.APPLY.value,
old_value=["a", "b"],
new_value="plain",
status=AUDIT_STATUS_FAILED,
detail="iptables: exit 4",
)
entry = audit.last_entry(db.connection)
assert entry is not None
assert entry.old_value == ["a", "b"]
assert entry.new_value == "plain"
assert entry.status == AUDIT_STATUS_FAILED
assert entry.detail == "iptables: exit 4"
def test_no_audit_row_when_transaction_rolls_back(tmp_path: Path) -> None:
"""Мутация и её аудит коммитятся вместе: откат транзакции не оставляет следа."""
with make_db(tmp_path) as db:
with pytest.raises(RuntimeError), db.transaction() as conn:
audit.record(conn, actor="system", action="update", entity=Entity.RULE.value)
raise RuntimeError("мутация не удалась")
assert audit.count(db.connection) == 0
def test_last_entry_returns_none_on_empty_log(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
assert audit.last_entry(db.connection) is None
assert audit.list_entries(db.connection) == []
-308
View File
@@ -1,308 +0,0 @@
"""Тесты потокобезопасности слоя хранения (#54, D-1/D-2 из #55; issue #20).
Контракт (задокументирован в ``backend/awg_api/store/README.md``): одно
соединение SQLite, все операции — чтения и записи — сериализуются одним
``RLock``. Читатель никогда не видит незафиксированных строк и состояния,
которое затем будет откачено, а конкурентные мутации дают согласованный
журнал аудита.
"""
from __future__ import annotations
import asyncio
import threading
import time
from collections.abc import Mapping
from pathlib import Path
from typing import Any
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
ApplyEngine,
ApplyError,
Database,
DesiredState,
Entity,
Mode,
Rule,
Store,
capture_state,
)
THREADS = 4
PER_THREAD = 25
JOIN_TIMEOUT = 30.0
# окно наблюдения: за это время незаблокированное чтение гарантированно успело бы вернуться
OBSERVE_SECONDS = 0.2
RULE_INSERT = (
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
)
def open_store(tmp_path: Path) -> tuple[Database, Store]:
db = Database(tmp_path / "data" / "awg-dash.db")
return db, Store(db)
def rule(tid: int, idx: int) -> Rule:
return Rule(
name=f"r{tid}x{idx}",
kind="domain",
match=f"t{tid}r{idx}.example.com",
target="tun0",
)
def run_threads(workers: list[Any]) -> list[threading.Thread]:
threads = [threading.Thread(target=worker) for worker in workers]
for thread in threads:
thread.start()
for thread in threads:
thread.join(JOIN_TIMEOUT)
assert not thread.is_alive(), f"поток {thread.name} не завершился"
return threads
# -- #54: конкурентные мутации вместе с чтениями ---------------------------------
def test_concurrent_upserts_and_reads_are_consistent(tmp_path: Path) -> None:
"""N потоков × M upsert с чтениями: без ошибок, rules == audit == N*M."""
db, store = open_store(tmp_path)
with db:
errors: list[str] = []
barrier = threading.Barrier(THREADS)
def worker(tid: int) -> None:
try:
barrier.wait(timeout=JOIN_TIMEOUT)
for idx in range(PER_THREAD):
store.upsert_rule(rule(tid, idx), actor=f"thread{tid}")
store.list_rules()
store.get_mode()
store.get_rule_by_name(f"r{tid}x{idx}")
except BaseException as exc: # noqa: BLE001 - фиксируем любое падение потока
errors.append(f"thread{tid}: {type(exc).__name__}: {exc}")
run_threads([lambda tid=tid: worker(tid) for tid in range(THREADS)])
total = THREADS * PER_THREAD
assert errors == []
assert store.counts()["rules"] == total
assert store.counts()["audit"] == total
assert len(store.list_rules()) == total
assert len(store.list_audit(limit=total * 2)) == total
# -- #54: чтение не видит незафиксированных строк -------------------------------
def test_reads_never_see_uncommitted_rows(tmp_path: Path) -> None:
"""Пока транзакция другого потока открыта, читатель ждёт коммита, а не читает грязь."""
db, store = open_store(tmp_path)
with db:
store.upsert_rule(
Rule(name="baseline", kind="domain", match="baseline.example.com", target="tun0"),
actor="system",
)
transaction_open = threading.Event()
reader_started = threading.Event()
release = threading.Event()
observed: list[list[str]] = []
writer_errors: list[str] = []
def writer() -> None:
try:
with db.transaction() as conn:
conn.execute(
RULE_INSERT,
(
"uncommitted",
"domain",
"uncommitted.example.com",
"tun0",
100,
1,
"2026-01-01T00:00:00Z",
"2026-01-01T00:00:00Z",
),
)
transaction_open.set()
release.wait(JOIN_TIMEOUT)
except BaseException as exc: # noqa: BLE001 - фиксируем падение писателя
writer_errors.append(f"{type(exc).__name__}: {exc}")
def reader() -> None:
reader_started.set()
observed.append([item.name for item in store.list_rules()])
writer_thread = threading.Thread(target=writer)
writer_thread.start()
assert transaction_open.wait(JOIN_TIMEOUT)
reader_thread = threading.Thread(target=reader)
reader_thread.start()
assert reader_started.wait(JOIN_TIMEOUT)
time.sleep(OBSERVE_SECONDS)
# незафиксированная строка наружу не видна: чтение ждёт завершения транзакции
assert observed == []
release.set()
writer_thread.join(JOIN_TIMEOUT)
reader_thread.join(JOIN_TIMEOUT)
assert writer_errors == []
assert observed == [["baseline", "uncommitted"]]
# -- #54 п.4: во время падающего apply наружу не отдаётся откаченное состояние ----
class BlockingApplier:
"""Применитель, который «висит» внутри apply и затем падает как iptables."""
def __init__(self) -> None:
self.entered = threading.Event()
self.release = threading.Event()
self.reverted: list[dict[str, Any]] = []
def apply(self, payload: Mapping[str, Any]) -> None:
self.entered.set()
assert self.release.wait(JOIN_TIMEOUT)
raise RuntimeError("mock iptables: exit 4")
def revert(self, payload: Mapping[str, Any]) -> None:
self.reverted.append(dict(payload))
def test_read_during_failing_apply_never_sees_discarded_state(tmp_path: Path) -> None:
"""AC #20: снаружи не видно «полуприменённой» конфигурации, которую затем откатят."""
db, store = open_store(tmp_path)
applier = BlockingApplier()
engine = ApplyEngine(store, applier, actor="ui:admin")
with db:
store.upsert_rule(
Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system",
)
before = capture_state(db.connection)
plan = engine.stage(
DesiredState(
mode=Mode.FULL,
rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),),
),
reason="сломается",
)
errors: list[BaseException] = []
observed: list[list[str]] = []
reader_started = threading.Event()
def apply_worker() -> None:
try:
engine.apply(plan)
except BaseException as exc: # noqa: BLE001 - ожидаем ApplyError
errors.append(exc)
def reader() -> None:
reader_started.set()
observed.append([item.name for item in store.list_rules()])
apply_thread = threading.Thread(target=apply_worker)
apply_thread.start()
assert applier.entered.wait(JOIN_TIMEOUT)
reader_thread = threading.Thread(target=reader)
reader_thread.start()
assert reader_started.wait(JOIN_TIMEOUT)
time.sleep(OBSERVE_SECONDS)
assert observed == [], "чтение вернуло состояние, которого ещё нет в БД"
applier.release.set()
apply_thread.join(JOIN_TIMEOUT)
reader_thread.join(JOIN_TIMEOUT)
assert len(errors) == 1
error = errors[0]
assert isinstance(error, ApplyError)
assert error.restored is True
assert observed == [["keep"]]
assert [item.name for item in store.list_rules()] == ["keep"]
assert capture_state(db.connection) == before
assert applier.reverted == [before]
# -- #55 D-1: BaseException не должен оставлять staged без исхода ---------------
class InterruptingApplier:
"""Применитель, прерывающий apply исключением из ``BaseException``."""
def __init__(self, exc_type: type[BaseException]) -> None:
self.exc_type = exc_type
self.applied = 0
self.reverted: list[dict[str, Any]] = []
def apply(self, payload: Mapping[str, Any]) -> None:
self.applied += 1
raise self.exc_type("внешний применитель прерван")
def revert(self, payload: Mapping[str, Any]) -> None:
self.reverted.append(dict(payload))
@pytest.mark.parametrize("exc_type", [KeyboardInterrupt, SystemExit, asyncio.CancelledError])
def test_base_exception_from_applier_reverts_and_records_failure(
tmp_path: Path, exc_type: type[BaseException]
) -> None:
"""Отмена/прерывание apply → revert внешней подсистемы + failed в audit_log."""
db, store = open_store(tmp_path)
applier = InterruptingApplier(exc_type)
engine = ApplyEngine(store, applier, actor="ui:admin")
with db:
store.upsert_rule(
Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system",
)
before = capture_state(db.connection)
plan = engine.stage(
DesiredState(
mode=Mode.FULL,
rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),),
),
reason="отмена",
)
with pytest.raises(exc_type):
engine.apply(plan)
assert applier.applied == 1
assert applier.reverted == [before]
assert capture_state(db.connection) == before
assert [item.name for item in store.list_rules()] == ["keep"]
entries = store.list_audit(entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=10)
assert entries[0].status == AUDIT_STATUS_FAILED
assert exc_type.__name__ in (entries[0].detail or "")
assert entries[0].new_value == before
# -- #55 D-2: конкурентные set_geoip_schedule дают связный журнал ---------------
def test_concurrent_geoip_updates_keep_audit_chain_consistent(tmp_path: Path) -> None:
"""Каждое изменение пишет корректный old_value: цепочка old→new не рвётся."""
db, store = open_store(tmp_path)
with db:
initial = store.get_geoip_schedule()
crons = ("0 1 * * *", "0 2 * * *")
errors: list[str] = []
barrier = threading.Barrier(len(crons))
def worker(cron: str) -> None:
try:
barrier.wait(timeout=JOIN_TIMEOUT)
store.set_geoip_schedule(cron=cron, actor=f"ui:{cron}")
except BaseException as exc: # noqa: BLE001 - фиксируем падение потока
errors.append(f"{cron}: {type(exc).__name__}: {exc}")
run_threads([lambda cron=cron: worker(cron) for cron in crons])
assert errors == []
entries = list(reversed(store.list_audit(entity=Entity.GEOIP_SCHEDULE.value, limit=10)))
assert len(entries) == len(crons)
first, second = entries
assert first.old_value["cron"] == initial.cron
assert first.new_value["cron"] == second.old_value["cron"]
assert second.new_value["cron"] == store.get_geoip_schedule().cron
assert {first.new_value["cron"], second.new_value["cron"]} == set(crons)
-153
View File
@@ -1,153 +0,0 @@
"""Тесты схемы, миграций и проверки версии при старте (issue #20)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from awg_api.store import (
EXPECTED_TABLES,
LATEST_VERSION,
MIGRATIONS,
Database,
Migration,
SchemaError,
StorageError,
current_version,
migrate,
resolve_db_path,
verify_schema,
)
from awg_api.store.migrations import applied_migrations, table_columns, table_names
def test_empty_data_dir_creates_schema(tmp_path: Path) -> None:
"""GIVEN пустой /data, WHEN старт, THEN схема v1 создана автоматически."""
db_path = tmp_path / "data" / "awg-dash.db"
assert not db_path.parent.exists()
with Database(db_path) as db:
assert db.schema_version == LATEST_VERSION == 1
assert db.migration_report is not None
assert db.migration_report.applied == (1,)
names = table_names(db.connection)
assert set(EXPECTED_TABLES) <= names
for table, columns in EXPECTED_TABLES.items():
assert columns <= table_columns(db.connection, table), table
def test_wal_mode_is_enabled(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
assert db.journal_mode.lower() == "wal"
assert db.is_wal is True
def test_seed_rows_present(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
row = db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
assert row is not None and row["mode"] == "off"
geoip = db.connection.execute("SELECT cron, provider FROM geoip_schedule WHERE id = 1").fetchone()
assert geoip is not None
assert geoip["provider"] == "ripe"
def test_migrate_is_idempotent(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
report = db.migrate()
assert report.applied == ()
assert report.changed is False
assert report.from_version == report.to_version == LATEST_VERSION
def test_applied_migrations_map(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
assert applied_migrations(db.connection) == {1: "initial-state-schema"}
def test_auto_migrate_false_verifies_existing_schema(tmp_path: Path) -> None:
path = tmp_path / "awg-dash.db"
with Database(path):
pass
with Database(path, auto_migrate=False) as db:
assert db.migration_report is None
assert current_version(db.connection) == LATEST_VERSION
def test_auto_migrate_false_on_unmigrated_db_raises(tmp_path: Path) -> None:
conn = sqlite3.connect(str(tmp_path / "empty.db"))
try:
with pytest.raises(SchemaError, match="не мигрирована"):
verify_schema(conn)
finally:
conn.close()
def test_verify_schema_detects_missing_table(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute("DROP TABLE routing_rules")
with pytest.raises(SchemaError, match="routing_rules"):
db.verify_schema()
def test_verify_schema_detects_missing_column(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute("ALTER TABLE devices DROP COLUMN address")
with pytest.raises(SchemaError, match="address"):
db.verify_schema()
def test_verify_schema_rejects_newer_schema(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute(
"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
(99, "from-the-future", "2099-01-01T00:00:00.000Z"),
)
with pytest.raises(SchemaError, match="новее"):
db.verify_schema()
with pytest.raises(SchemaError, match="новее"):
db.migrate()
def test_failed_migration_leaves_schema_untouched(tmp_path: Path) -> None:
broken = MIGRATIONS + (
Migration(version=2, name="broken", statements=("CREATE TABLE t2 (id INTEGER)", "NOT SQL AT ALL")),
)
with Database(tmp_path / "awg-dash.db") as db:
with pytest.raises(StorageError, match="миграция v2"):
migrate(db.connection, migrations=broken)
assert current_version(db.connection) == 1
assert "t2" not in table_names(db.connection)
def test_migrate_rejects_downgrade(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute(
"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
(2, "newer", "2099-01-01T00:00:00.000Z"),
)
with pytest.raises(SchemaError, match="новее"):
migrate(db.connection, migrations=MIGRATIONS)
def test_resolve_db_path_defaults_and_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("AWG_DASH_DB", raising=False)
assert resolve_db_path() == Path("/data/awg-dash.db")
monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "custom.db"))
assert resolve_db_path() == tmp_path / "custom.db"
assert resolve_db_path(tmp_path / "explicit.db") == tmp_path / "explicit.db"
def test_database_uses_env_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "nested" / "env.db"))
with Database() as db:
assert db.path == tmp_path / "nested" / "env.db"
assert db.path.exists()
def test_unwritable_path_raises_storage_error(tmp_path: Path) -> None:
blocker = tmp_path / "blocked"
blocker.write_text("not a directory")
with pytest.raises(StorageError, match="каталог"):
Database(blocker / "awg-dash.db")
-184
View File
@@ -1,184 +0,0 @@
"""Персистентность: рестарт «контейнера» сохраняет правила, конфиг и аудит (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
ConfigMeta,
Database,
DesiredState,
Device,
GeoIpSchedule,
Mode,
Rule,
StorageError,
Store,
ValidationError,
open_storage,
rollback_to,
)
from awg_api.store import validation
def test_state_survives_container_restart(tmp_path: Path) -> None:
"""GIVEN рестарт контейнера, WHEN он завершён, THEN правила и конфиг на месте."""
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as first:
store = Store(first)
store.set_mode(Mode.SPLIT, actor="ui:admin")
store.set_geoip_schedule(cron="0 3 * * *", actor="ui:admin")
store.upsert_rule(
Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10),
actor="ui:admin",
)
store.upsert_rule(
Rule(name="lan", kind="cidr", match="10.0.0.0/8", target="direct", priority=200),
actor="ui:admin",
)
store.upsert_config(
ConfigMeta(name="peer1", path="/config/peer1.conf", checksum="deadbeef", size_bytes=1024),
actor="ui:admin",
)
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="notebook"), actor="ui:admin")
audit_before = store.counts()["audit"]
assert audit_before == 6
# «рестарт контейнера»: новое соединение к тому же /data
with Database(db_path) as second:
store = Store(second)
assert second.is_wal is True
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().cron == "0 3 * * *"
assert [(rule.name, rule.target) for rule in store.list_rules()] == [
("yt", "tun0"),
("lan", "direct"),
]
config = store.get_config("peer1")
assert config is not None
assert config.checksum == "deadbeef"
assert store.get_device("aa:bb:cc:dd:ee:ff") is not None
assert store.counts()["audit"] == audit_before
assert second.migration_report is not None # миграции уже применены, повтор не нужен
assert second.migration_report.applied == ()
def test_wal_file_next_to_database(tmp_path: Path) -> None:
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as db:
Store(db).set_mode(Mode.FULL, actor="system")
assert db_path.with_name("awg-dash.db-wal").exists()
assert db_path.with_name("awg-dash.db-shm").exists()
def test_open_storage_builds_full_stack(tmp_path: Path) -> None:
with open_storage(tmp_path / "data" / "awg-dash.db", actor="api:token") as storage:
storage.store.set_mode(Mode.SPLIT, actor="api:token")
plan = storage.engine.stage(
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule()),
reason="через storage",
)
result = storage.engine.apply(plan)
assert result.ok is True
assert storage.store.counts()["snapshots"] == 1
assert storage.db.schema_version == 1
def test_rollback_helper_and_missing_snapshot(tmp_path: Path) -> None:
with open_storage(tmp_path / "data" / "awg-dash.db") as storage:
storage.store.set_mode(Mode.OFF, actor="system")
plan = storage.engine.stage(DesiredState(mode=Mode.FULL))
storage.engine.apply(plan)
assert storage.store.get_mode() == Mode.FULL
snapshot = rollback_to(storage.engine, plan.plan_id, reason="ручной откат")
assert snapshot.payload["mode"] == "off"
assert storage.store.get_mode() == Mode.OFF
with pytest.raises(StorageError, match="не найден"):
storage.engine.rollback(999999)
def test_transactions_are_isolated_per_connection(tmp_path: Path) -> None:
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as writer, Database(db_path) as reader:
Store(writer).set_mode(Mode.SPLIT, actor="system")
assert Store(reader).get_mode() == Mode.SPLIT
with pytest.raises(RuntimeError, match="откат"), writer.transaction() as conn:
conn.execute("UPDATE routing_mode SET mode = 'full' WHERE id = 1")
raise RuntimeError("откат")
assert Store(reader).get_mode() == Mode.SPLIT
# -- валидация (чистые функции) -------------------------------------------------
@pytest.mark.parametrize(
("raw", "expected"),
[
("AA:BB:CC:DD:EE:FF", "aa:bb:cc:dd:ee:ff"),
("aa-bb-cc-dd-ee-ff", "aa:bb:cc:dd:ee:ff"),
("aabbccddeeff", "aa:bb:cc:dd:ee:ff"),
],
)
def test_normalize_mac(raw: str, expected: str) -> None:
assert validation.normalize_mac(raw) == expected
def test_normalize_mac_rejects_garbage() -> None:
with pytest.raises(ValidationError, match="MAC"):
validation.normalize_mac("not-a-mac")
@pytest.mark.parametrize("value", ["ok", "peer_1", "rule-2", "a.b:c"])
def test_validate_name_accepts(value: str) -> None:
assert validation.validate_name("name", f" {value} ") == value
@pytest.mark.parametrize("value", ["", "UPPER", "-lead", "with space", "юникод", "x" * 64])
def test_validate_name_rejects(value: str) -> None:
with pytest.raises(ValidationError):
validation.validate_name("name", value)
@pytest.mark.parametrize("value", ["/config/peer1.conf", "/etc/awg-dash/rules.conf"])
def test_validate_config_path_accepts(value: str) -> None:
assert validation.validate_config_path(value) == value
@pytest.mark.parametrize("value", ["relative.conf", "/config/../../etc/shadow", ""])
def test_validate_config_path_rejects(value: str) -> None:
with pytest.raises(ValidationError):
validation.validate_config_path(value)
@pytest.mark.parametrize(("value", "expected"), [("RIPE", "ripe"), ("maxmind", "maxmind")])
def test_validate_provider(value: str, expected: str) -> None:
assert validation.validate_provider(value) == expected
@pytest.mark.parametrize("value", ["tunnel", "direct", "block", "default"])
def test_validate_device_policy_accepts(value: str) -> None:
assert validation.validate_device_policy(value) == value
def test_validate_device_policy_rejects() -> None:
with pytest.raises(ValidationError, match="policy"):
validation.validate_device_policy("root")
@pytest.mark.parametrize("value", ["*.example.com", "youtube.com", "a.b.c.example.co.uk"])
def test_validate_match_domain(value: str) -> None:
assert validation.validate_match("domain", value) == value
@pytest.mark.parametrize("value", ["10.0.0.0/8", "2001:db8::/32", "192.168.1.7"])
def test_validate_match_cidr(value: str) -> None:
assert validation.validate_match("cidr", value) == value
def test_validate_match_ipset_and_empty() -> None:
assert validation.validate_match("ipset", "awg_ru") == "awg_ru"
with pytest.raises(ValidationError, match="пустым"):
validation.validate_match("domain", " ")
-312
View File
@@ -1,312 +0,0 @@
"""Тесты репозитория: CRUD-мутации с аудитом «old→new» (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
ConfigMeta,
Database,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
StorageError,
Store,
ValidationError,
)
from awg_api.store import audit
from awg_api.store.validation import validate_priority
def make_store(tmp_path: Path) -> tuple[Database, Store]:
db = Database(tmp_path / "data" / "awg-dash.db")
return db, Store(db)
# -- правила -----------------------------------------------------------------
def test_upsert_rule_creates_with_audit(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(
Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10),
actor="ui:admin",
)
assert saved.id is not None and saved.created_at == saved.updated_at
assert store.get_rule(saved.id) == saved
assert store.get_rule_by_name("yt") == saved
entries = store.list_audit(entity=Entity.RULE.value)
assert len(entries) == 1
assert entries[0].action == "create"
assert entries[0].old_value is None
assert entries[0].new_value is not None
assert entries[0].new_value["name"] == "yt"
def test_upsert_rule_updates_by_name_and_records_old_new(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
first = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
second = store.upsert_rule(Rule(name="yt", kind="domain", match="youtu.be", target="tun1"),
actor="ui:admin")
assert second.id == first.id
assert second.match == "youtu.be" and second.target == "tun1"
assert store.counts()["rules"] == 1
entries = store.list_audit(entity=Entity.RULE.value)
assert [entry.action for entry in entries] == ["update", "create"]
assert entries[0].old_value["match"] == "youtube.com"
assert entries[0].new_value["match"] == "youtu.be"
assert entries[0].old_value["target"] == "tun0"
assert entries[0].new_value["target"] == "tun1"
def test_upsert_rule_by_id(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(Rule(name="a", kind="cidr", match="10.0.0.0/8", target="tun0"),
actor="system")
updated = store.upsert_rule(
Rule(id=saved.id, name="a", kind="cidr", match="10.1.0.0/16", target="tun0", enabled=False),
actor="system",
)
assert updated.id == saved.id
assert updated.match == "10.1.0.0/16"
assert updated.enabled is False
assert store.list_rules(enabled_only=True) == []
def test_rule_validation_rejects_bad_input(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
cases = [
Rule(name="Bad Name", kind="domain", match="youtube.com", target="tun0"),
Rule(name="ok", kind="wrong", match="youtube.com", target="tun0"),
Rule(name="ok2", kind="domain", match="not a domain", target="tun0"),
Rule(name="ok3", kind="cidr", match="10.0.0.0/99", target="tun0"),
Rule(name="ok4", kind="domain", match="youtube.com", target="bad target!"),
Rule(name="ok5", kind="domain", match="youtube.com", target="tun0", priority=-1),
]
for rule in cases:
with pytest.raises(ValidationError):
store.upsert_rule(rule, actor="system")
assert store.counts()["rules"] == 0
assert audit.count(db.connection) == 0
def test_priority_must_be_int() -> None:
with pytest.raises(ValidationError, match="целое"):
validate_priority(True) # bool не является допустимым приоритетом
with pytest.raises(ValidationError, match="диапазона"):
validate_priority(70000)
def test_delete_rule(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
assert saved.id is not None
assert store.delete_rule(saved.id, actor="ui:admin") is True
assert store.delete_rule(saved.id, actor="ui:admin") is False
assert store.get_rule(saved.id) is None
entries = store.list_audit(entity=Entity.RULE.value)
assert [entry.action for entry in entries] == ["delete", "create"]
assert entries[0].new_value is None
assert entries[0].old_value["name"] == "yt"
def test_rules_sorted_by_priority(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
store.upsert_rule(Rule(name="b", kind="domain", match="b.example.com", target="tun0",
priority=200), actor="system")
store.upsert_rule(Rule(name="a", kind="domain", match="a.example.com", target="tun0",
priority=5), actor="system")
assert [rule.name for rule in store.list_rules()] == ["a", "b"]
# -- исключения ---------------------------------------------------------------
def test_exception_upsert_update_delete(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.upsert_exception(RuleException(match="bank.example.com", reason="банк"),
actor="ui:admin")
assert created.id is not None
assert store.get_exception("bank.example.com") == created
updated = store.upsert_exception(RuleException(match="bank.example.com", reason="банк2"),
actor="ui:admin")
assert updated.id == created.id and updated.reason == "банк2"
assert store.counts()["exceptions"] == 1
assert store.list_exceptions() == [updated]
assert store.delete_exception(updated.id or 0, actor="ui:admin") is True
assert store.delete_exception(updated.id or 0, actor="ui:admin") is False
actions = [entry.action for entry in store.list_audit(entity=Entity.EXCEPTION.value)]
assert actions == ["delete", "update", "create"]
update_entry = store.list_audit(entity=Entity.EXCEPTION.value)[1]
assert update_entry.old_value["reason"] == "банк"
assert update_entry.new_value["reason"] == "банк2"
def test_exception_accepts_cidr_and_rejects_garbage(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
assert store.upsert_exception(RuleException(match="192.168.1.0/24"), actor="system").match == (
"192.168.1.0/24"
)
assert store.upsert_exception(RuleException(match="10.0.0.1"), actor="system").match == "10.0.0.1"
with pytest.raises(ValidationError, match="домен"):
store.upsert_exception(RuleException(match="!!!"), actor="system")
with pytest.raises(ValidationError, match="пустым"):
store.upsert_exception(RuleException(match=" "), actor="system")
# -- режим и GeoIP -------------------------------------------------------------
def test_set_mode_defaults_off_and_audits(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
assert store.get_mode() == Mode.OFF
assert store.set_mode("split", actor="ui:admin") == Mode.SPLIT
assert store.get_mode() == Mode.SPLIT
assert store.set_mode(Mode("full"), actor="ui:admin") == Mode.FULL
entries = store.list_audit(entity=Entity.MODE.value)
assert len(entries) == 2
assert entries[0].old_value == {"mode": "split"}
assert entries[0].new_value == {"mode": "full"}
assert entries[0].actor == "ui:admin"
def test_set_mode_rejects_unknown(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="mode"):
store.set_mode("turbo", actor="ui:admin")
assert store.get_mode() == Mode.OFF
def test_geoip_schedule_defaults_and_partial_update(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
default = store.get_geoip_schedule()
assert default == GeoIpSchedule(enabled=True, cron="0 4 * * *", provider="ripe",
updated_at=default.updated_at)
updated = store.set_geoip_schedule(cron="*/30 * * * *", actor="ui:admin")
assert updated.cron == "*/30 * * * *"
assert updated.provider == "ripe" and updated.enabled is True
disabled = store.set_geoip_schedule(enabled=False, provider="MAXMIND", actor="ui:admin")
assert disabled.enabled is False and disabled.provider == "maxmind"
assert store.get_geoip_schedule() == disabled
with pytest.raises(ValidationError, match="cron"):
store.set_geoip_schedule(cron="ежечасно", actor="ui:admin")
with pytest.raises(ValidationError, match="provider"):
store.set_geoip_schedule(provider="!", actor="ui:admin")
entries = store.list_audit(entity=Entity.GEOIP_SCHEDULE.value)
assert [entry.action for entry in entries] == ["set_geoip_schedule"] * 2
assert entries[0].old_value["provider"] == "ripe"
assert entries[0].new_value["provider"] == "maxmind"
# -- конфиги ------------------------------------------------------------------
def test_config_upsert_and_mark_applied(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.upsert_config(
ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=1, checksum="abc",
size_bytes=512), actor="ui:admin",
)
assert created.id is not None
assert store.get_config("client-peer1") == created
updated = store.upsert_config(
ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=2, checksum="def",
size_bytes=640, applied=True), actor="ui:admin",
)
assert updated.version == 2 and updated.applied is True
assert store.counts()["configs"] == 1
marked = store.mark_config_applied("client-peer1", applied=False, actor="ui:admin")
assert marked.applied is False
entries = store.list_audit(entity=Entity.CONFIG.value)
assert [entry.action for entry in entries] == ["set_applied", "update", "create"]
assert entries[0].old_value["applied"] is True
assert entries[0].new_value["applied"] is False
assert entries[1].old_value["version"] == 1
assert entries[1].new_value["version"] == 2
def test_config_validation_and_missing(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="path"):
store.upsert_config(ConfigMeta(name="x", path="relative.conf"), actor="system")
with pytest.raises(ValidationError, match="path"):
store.upsert_config(ConfigMeta(name="x", path="/config/../etc/passwd"), actor="system")
with pytest.raises(ValidationError, match="name"):
store.upsert_config(ConfigMeta(name="Bad Name", path="/config/x.conf"), actor="system")
with pytest.raises(StorageError, match="не найден"):
store.mark_config_applied("ghost", applied=True, actor="system")
# -- устройства -----------------------------------------------------------------
def test_device_register_normalizes_mac_and_audits(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.register_device(
Device(mac="AA-BB-CC-DD-EE-FF", name="notebook", address="192.168.1.50"),
actor="ui:admin",
)
assert created.mac == "aa:bb:cc:dd:ee:ff"
assert store.get_device("AA:BB:CC:DD:EE:FF") == created
updated = store.register_device(
Device(mac="aabbccddeeff", name="notebook", policy="direct", last_seen="2026-01-01T00:00:00.000Z"),
actor="ui:admin",
)
assert updated.id == created.id and updated.policy == "direct"
assert store.counts()["devices"] == 1
assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is True
assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is False
assert store.list_devices() == []
actions = [entry.action for entry in store.list_audit(entity=Entity.DEVICE.value)]
assert actions == ["delete", "update", "create"]
update_entry = store.list_audit(entity=Entity.DEVICE.value)[1]
assert update_entry.old_value["policy"] == "default"
assert update_entry.new_value["policy"] == "direct"
def test_device_validation(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="MAC"):
store.register_device(Device(mac="zz:zz", name="x"), actor="system")
with pytest.raises(ValidationError, match="policy"):
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="x", policy="кибератака"),
actor="system")
with pytest.raises(ValidationError, match="name"):
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="Not A Name"), actor="system")
def test_audit_actor_recorded_per_mutation(tmp_path: Path) -> None:
"""GIVEN изменение через сервисный слой, THEN в audit_log есть actor и old/new."""
db, store = make_store(tmp_path)
with db:
store.set_mode(Mode.SPLIT, actor="api:token-42")
entry = store.list_audit(entity=Entity.MODE.value)[0]
assert entry.actor == "api:token-42"
assert entry.ts.endswith("Z")
assert entry.old_value == {"mode": "off"}
assert entry.new_value == {"mode": "split"}
assert entry.status == "ok"
-79
View File
@@ -1,79 +0,0 @@
# syntax=docker/dockerfile:1.7
#
# awg-dash — единый образ: AmneziaWG gateway + policy routing + REST API + Web UI.
# Задача: #18 (эпик #15). ADR: docs/adr/002-single-image-modular-monolith.md
#
# ВАЖНО: AMNEZIAWG_GO_REF / AMNEZIAWG_TOOLS_REF по умолчанию указывают на master
# и должны быть запинены на проверенный тег/коммит перед релизом (см. #45).
ARG GO_IMAGE=golang:1.25-alpine
# --- stage 1: сборка awg-утилит (Go) -----------------------------------------
FROM ${GO_IMAGE} AS go-build
ARG AMNEZIAWG_GO_REPO=https://github.com/amnezia-vpn/amneziawg-go.git
ARG AMNEZIAWG_GO_REF=master
ARG AMNEZIAWG_TOOLS_REPO=https://github.com/amnezia-vpn/amneziawg-tools.git
ARG AMNEZIAWG_TOOLS_REF=master
RUN apk add --no-cache git make gcc musl-dev linux-headers
WORKDIR /src
RUN git clone --depth 1 --branch "${AMNEZIAWG_GO_REF}" "${AMNEZIAWG_GO_REPO}" amneziawg-go \
&& cd amneziawg-go && make && cp amneziawg-go /usr/local/bin/amneziawg-go
RUN git clone --depth 1 --branch "${AMNEZIAWG_TOOLS_REF}" "${AMNEZIAWG_TOOLS_REPO}" amneziawg-tools \
&& cd amneziawg-tools/src && make \
&& cp wg /usr/local/bin/awg \
&& if [ -f wg-quick ]; then cp wg-quick /usr/local/bin/awg-quick; fi
# --- stage 2: runtime -------------------------------------------------------
FROM alpine:3.21 AS runtime
LABEL org.opencontainers.image.title="awg-dash" \
org.opencontainers.image.description="AmneziaWG gateway with intelligent routing: REST API + Web UI + split-tunnel" \
org.opencontainers.image.source="http://192.168.1.111:3000/embeddevio/awg-dash"
# Сетевые утилиты: iptables/ip6tables, ipset, iproute2 (policy routing), dnsmasq (ADR-005),
# curl (HEALTHCHECK), tini (PID 1), su-exec (запуск API под непривилегированным пользователем).
RUN apk add --no-cache \
bash curl iproute2 ipset iptables ip6tables dnsmasq \
python3 py3-pip tini su-exec \
&& addgroup -S awg \
&& adduser -S -G awg -h /app awg
ENV PATH=/opt/venv/bin:$PATH \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
PYTHONPATH=/app/backend \
AWG_CONFIG_DIR=/config \
AWG_DATA_DIR=/data \
AWG_RUN_DIR=/run/awg \
AWG_API_HOST=0.0.0.0 \
AWG_API_PORT=8080
WORKDIR /app
COPY requirements.txt /app/requirements.txt
RUN apk add --no-cache --virtual .build-deps gcc musl-dev python3-dev libffi-dev \
&& python3 -m venv /opt/venv \
&& /opt/venv/bin/pip install --no-cache-dir --upgrade pip \
&& /opt/venv/bin/pip install --no-cache-dir -r /app/requirements.txt \
&& apk del .build-deps
COPY backend/ /app/backend/
COPY --from=go-build /usr/local/bin/awg /usr/local/bin/awg
COPY --from=go-build /usr/local/bin/amneziawg-go /usr/local/bin/amneziawg-go
COPY deploy/entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh \
&& mkdir -p /config /data /run/awg \
&& chown -R awg:awg /app /config /data /run/awg
VOLUME ["/config", "/data"]
EXPOSE 8080
# healthcheck ходит в /api/v1/health (задача #31). До реализации API контейнер
# будет помечаться как unhealthy — это ожидаемое состояние до #31.
HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
CMD curl -fsS "http://127.0.0.1:${AWG_API_PORT}/api/v1/health" >/dev/null || exit 1
ENTRYPOINT ["/sbin/tini", "--", "/entrypoint.sh"]
-12
View File
@@ -1,12 +0,0 @@
# deploy — сборка и развёртывание
| Артефакт | Назначение | Задача |
|---|---|---|
| `Dockerfile` | multi-stage: amneziawg-go + awg-tools → runtime alpine | #18 |
| `entrypoint.sh` | супервизор: `awg-agent` (root) + `awg-api` | #18 |
| `docker-compose.dev.yml` | dev-стенд: порт 8080, тома `/config` `/data`, healthcheck | #21 |
| `docker-compose.test.yml` | стенд для тестов | #21 |
| `grafana/` | дашборд JSON (опциональный профиль monitoring) | #46 |
| `awg0.conf.example` | пример клиентского конфига | #22 |
Запуск по ТЗ (после релиза): `--network=host`, `--device=/dev/net/tun`, `--cap-add=NET_ADMIN --cap-add=SYS_MODULE`, `--restart always`.
-32
View File
@@ -1,32 +0,0 @@
# Пример клиентского конфига AmneziaWG для awg-dash.
#
# Скопируйте в том /config как awg0.conf (имя файла = имя интерфейса) или
# загрузите через Web UI (задача #38). Значения ниже — заглушки, замените своими.
# Файл в репозитории не является рабочим конфигом и не содержит реальных ключей.
[Interface]
# Адрес внутри туннеля (из панели провайдера AmneziaWG)
Address = 10.8.1.2/32
# DNS: адрес должен указывать на этот шлюз, иначе не сработает DNS-split (ADR-005)
DNS = 10.8.1.1
PrivateKey = <PRIVATE_KEY>
# Обфускация AmneziaWG (Amnezia VPN: экспорт -> AmneziaWG)
Jc = 4
Jmin = 40
Jmax = 70
S1 = 0
S2 = 0
H1 = 1
H2 = 2
H3 = 3
H4 = 4
[Peer]
PublicKey = <SERVER_PUBLIC_KEY>
PresharedKey = <PRESHARED_KEY>
AllowedIPs = 0.0.0.0/0, ::/0
Endpoint = <SERVER_HOST>:<SERVER_PORT>
PersistentKeepalive = 25
-77
View File
@@ -1,77 +0,0 @@
#!/bin/sh
# Супервизор контейнера awg-dash (задача #18, ADR-002).
# Последовательность: preflight -> подготовка -> запуск awg-api под awg.
# Привилегированный агент awg-agent стартует отсюда же после реализации (#22).
#
# Роль процесса: все сетевые операции выполняет агент под root; здесь только
# проверяем, что нужные возможности и устройства проброшены, и падаем явной
# ошибкой, если контейнер запущен без них (а не тихо деградируем).
set -eu
CONFIG_DIR="${AWG_CONFIG_DIR:-/config}"
DATA_DIR="${AWG_DATA_DIR:-/data}"
RUN_DIR="${AWG_RUN_DIR:-/run/awg}"
API_HOST="${AWG_API_HOST:-0.0.0.0}"
API_PORT="${AWG_API_PORT:-8080}"
ROUTING_TABLE_ID=100
ROUTING_TABLE_NAME=awg-vpn
log() { printf '[entrypoint] %s\n' "$*"; }
fail() { printf '[entrypoint] FATAL: %s\n' "$*" >&2; exit 1; }
log 'preflight'
[ -e /dev/net/tun ] || fail '/dev/net/tun не проброшен: добавьте --device=/dev/net/tun:/dev/net/tun'
iptables -t mangle -L -n >/dev/null 2>&1 || fail 'iptables недоступен: нужен --cap-add=NET_ADMIN (и/или --privileged)'
command -v awg >/dev/null 2>&1 || fail 'бинарь awg не найден в образе'
command -v amneziawg-go >/dev/null 2>&1 || log 'WARN: amneziawg-go не найден — остаётся только kernel-модуль'
command -v ipset >/dev/null 2>&1 || fail 'ipset не найден в образе'
log 'подготовка каталогов состояния'
mkdir -p "$CONFIG_DIR" "$DATA_DIR" "$RUN_DIR"
chown -R awg:awg "$CONFIG_DIR" "$DATA_DIR" "$RUN_DIR" 2>/dev/null || true
chmod 700 "$CONFIG_DIR" "$DATA_DIR" 2>/dev/null || true
mkdir -p /etc/iproute2
touch /etc/iproute2/rt_tables 2>/dev/null || true
if ! grep -q "^${ROUTING_TABLE_ID} ${ROUTING_TABLE_NAME}" /etc/iproute2/rt_tables 2>/dev/null; then
printf '%s %s\n' "$ROUTING_TABLE_ID" "$ROUTING_TABLE_NAME" >> /etc/iproute2/rt_tables \
|| log 'WARN: не удалось записать таблицу маршрутизации в /etc/iproute2/rt_tables'
fi
sysctl -w net.ipv4.ip_forward=1 >/dev/null 2>&1 \
|| log 'WARN: net.ipv4.ip_forward не включён — задайте --sysctl net.ipv4.ip_forward=1 или --privileged (см. runbook #45)'
log "awg: $(awg --version 2>/dev/null | head -n 1 || echo 'version unknown')"
api_pid=''
if [ -f /app/backend/awg_api/app.py ]; then
log "запуск awg-api на ${API_HOST}:${API_PORT}"
su-exec awg /opt/venv/bin/uvicorn awg_api.app:app --host "$API_HOST" --port "$API_PORT" &
api_pid=$!
else
log 'WARN: awg_api/app.py ещё не реализован (задача #31) — API не запущен, контейнер работает только как агент'
fi
stop() {
if [ -n "$api_pid" ] && kill -0 "$api_pid" 2>/dev/null; then
log 'остановка awg-api'
kill -TERM "$api_pid" 2>/dev/null || true
fi
}
trap stop INT TERM
if [ -n "$api_pid" ]; then
# ждём завершения API и прокидываем его код возврата в Docker
set +e
wait "$api_pid"
rc=$?
set -e
log "awg-api завершился с кодом $rc"
exit "$rc"
fi
log 'активных процессов нет (agent #22 и api #31 ещё не реализованы) — ожидание сигнала'
while :; do
sleep 3600 &
wait $!
done
-21
View File
@@ -1,21 +0,0 @@
# NNN. Название решения
- **Статус:** предложено | принято | устарело
- **Дата:** YYYY-MM-DD
- **Контекст/issue:** #N
## Контекст
Что вынуждает принять решение и какие есть ограничения (хостовое окружение, привилегии, один шлюз, целевой UX).
## Решение
Что именно решили, одним абзацем.
## Альтернативы
Что рассмотрели и почему отклонили.
## Последствия
Положительные и отрицательные следствия; что решение накладывает на задачи эпика.
-27
View File
@@ -1,27 +0,0 @@
# 001. Backend — Python 3.12 + FastAPI
- **Статус:** принято
- **Контекст/issue:** #15, #31
## Контекст
Нужен control plane: REST API, WebSocket, планировщик, парсинг конфигов, работа с SQLite.
Низкоуровневая работа сети и шифрования выполняется не нашим кодом, а бинарями `awg` и `amneziawg-go`,
трафик маршрутизирует ядро.
## Решение
Control plane — **Python 3.12 + FastAPI (async)**. Go остаётся только там, где он уже есть:
`amneziawg-go` и `amneziawg-tools` собираются в multi-stage Dockerfile (ADR-002).
## Альтернативы
- **Go целиком** — быстрее на сетевом I/O, но маршрутизацией занимается ядро, а не процесс;
выигрыш не оправдывает рост стоимости контракта с UI (схемы, генерация клиента, миграции, веб-фреймворк).
- **Node.js** — преимуществ перед Python на нагрузке в единичного пользователя нет.
## Последствия
- OpenAPI-first: FastAPI отдаёт схему, что прямо соответствует задаче #19.
- В образе появляется Python runtime (плюс 40–60 MB к размеру).
- Дисциплина async: внешние команды — только `asyncio.create_subprocess_exec`, блокирующего кода в event loop нет.
@@ -1,32 +0,0 @@
# 002. Модульный монолит в одном образе, два процесса
- **Статус:** принято
- **Контекст/issue:** #15, #18
## Контекст
ТЗ требует единый Docker-образ с минимальными параметрами запуска. Часть операций требует root
(интерфейс awg, iptables/ipset, dnsmasq на порту 53), а веб-часть должна быть скомпрометирована безопасно — без root.
## Решение
Один образ, два процесса под супервизором `deploy/entrypoint.sh`:
| Процесс | Права | Ответственность |
|---|---|---|
| `awg-agent` | root, NET_ADMIN | туннель, netfilter, routing, dnsmasq, метрики |
| `awg-api` | непривилегированный | FastAPI, REST/WS, статика UI |
Обмен — unix-socket `/run/awg/agent.sock`, общее состояние — SQLite в `/data` и конфиги в `/config`.
## Альтернативы
- **Микросервисы** (отдельные контейнеры agent/api/dns) — общая сеть, координация и дублирование
привилегий; выгода масштабирования на одном шлюзе отсутствует.
- **Один процесс под root** — проще, но даёт веб-слою полные права; неприемлемо.
## Последствия
- Супервизор обязан пробрасывать сигналы и коды выхода, иначе Docker не заметит падения процесса.
- Права на unix-socket входят в security-проверки (#44).
- Перезапуск API не разрывает туннель.
-28
View File
@@ -1,28 +0,0 @@
# 003. Декларативный reconcile вместо императивных скриптов
- **Статус:** принято
- **Контекст/issue:** #15, #24, #27
## Контекст
PoC (`entrypoint.sh`) применяет правила императивно и только при старте: изменения не откатываются,
GeoIP не обновляется без рестарта, частичная ошибка оставляет систему в неопределённом состоянии.
## Решение
Единый reconcile-цикл: desired state (SQLite) -> diff -> идемпотентное применение -> проверка ->
rollback при ошибке. Операции netfilter идемпотентны (проверка `-C` перед `-A`), замена больших
наборов атомарна (`ipset create tmp`, `swap`, `destroy`), применение конфига транзакционно
(validate -> stage -> apply -> rollback).
## Альтернативы
- **Скрипты по событию** (как в PoC) — быстро написать, но невозможно гарантировать сходимость после сбоя.
- **Готовый IaC-инструмент** — избыточен, требует внешнего рантайма в образе и не покрывает ipset/dnsmasq.
## Последствия
- Старт контейнера = reconcile с нуля: состояние всегда восстановимо.
- Нужны тесты на идемпотентность (повторный apply не создаёт дублей) — #41.
- Сложнее скрипта: оправдано требованиями FR-2 (смена режимов и правил без рестарта).
- Diff reconcile логируется и отдаётся в UI (`/logs`) и audit log.
-28
View File
@@ -1,28 +0,0 @@
# 004. GeoIP: RIPE Stat + локальный кэш + абстракция провайдера
- **Статус:** принято
- **Контекст/issue:** #15, #25
## Контекст
Для split-tunnel нужны списки префиксов страны (в PoC — около 11 400 записей RIPE).
Варианты: RIPE Stat (бесплатно), ip2location и MaxMind (точнее, но лицензия и платный доступ).
## Решение
Основной источник — **RIPE Stat** (`country-resource-list`, IPv4 и IPv6): бесплатно и уже проверено в PoC.
Обязательные элементы: локальный кэш-снапшот в `/data/geoip/`, расписание обновления, ручное
обновление, явный статус источника (API или кэш с датой). Внутри — интерфейс `GeoProvider`, чтобы
добавить ip2location/MaxMind без переделки маршрутизации.
## Альтернативы
- **MaxMind GeoLite2** — точнее по ASN, но требует регистрации и обновления лицензии; оставляем как опцию.
- **Собственный список в репозитории** — устаревает и не покрывает смену маршрутной политики.
## Последствия
- Старт не зависит от внешней сети: при недоступности API работаем на кэше (#25, тест #41).
- Fallback не должен молча менять политику split-tunnel: неявные переходы в режим «всё в VPN» или
«всё напрямую» запрещены, поведение задаётся явно (#23).
- Загрузка 11k+ префиксов — измеримая операция; время старта и swap — часть AC #24 и #25.
-27
View File
@@ -1,27 +0,0 @@
# 005. DNS: встроенный dnsmasq и перехват порта 53
- **Статус:** принято
- **Контекст/issue:** #15, #26
## Контекст
DNS-based split-tunnel требует знать, какие IP принадлежат доменам из списка. Если клиенты сами
резолвят домены (свой DNS, DoH, другой upstream), шлюз не узнаёт об адресах и правило не срабатывает.
## Решение
В образе поднимается **встроенный dnsmasq** (процесс агента):
- домены из списка → `ipset` (через `--ipset`), трафик на эти адреса идёт в туннель;
- перехват порта 53 от LAN-клиентов шлюзом — устраняет DNS-утечку и лечит клиентов с ручным DNS;
- upstream-форвардинг и кэш по TTL; опция «внешний DNS» остаётся для нестандартных схем.
## Альтернативы
- **Отдельный DNS-контейнер** — лишняя зависимость и точка отказа, ломает требование единого образа.
- **Только DNS клиентов** — не работает без ручной настройки каждого устройства (противоречит FR-2).
## Последствия
- Порт 53 занят контейнером: в README/runbook нужно предупреждение (конфликт с локальным резольвером хоста).
- Обязательно тестировать домены с CNAME и динамическими IP (#26, #41).
- Список доменов и его размеры ограничиваются явно (валидация в API, #30).
-29
View File
@@ -1,29 +0,0 @@
# 006. Состояние: SQLite (WAL) + /config + /data
- **Статус:** принято
- **Контекст/issue:** #15, #20
## Контекст
Нужно хранить правила маршрутизации, исключения, режим, расписание GeoIP, метаданные конфигов,
устройства, audit log и историю метрик. Профиль нагрузки: единичный пользователь, десятки записей в секунду,
важна надёжность и отсутствие внешних зависимостей.
## Решение
- **SQLite в режиме WAL** в `/data/awg-dash.db` — всё структурированное состояние и audit log;
- `/config` — файлы `.conf`, права 0600, секреты в подкаталоге с ограниченным доступом;
- `/data` — кэш GeoIP и история метрик с retention.
Оба каталога объявляются томами образа: без них конфигурация не переживёт пересоздание контейнера.
## Альтернативы
- **Postgres/Redis** — лишний сервис в образе для единичного пользователя.
- **Только файлы JSON** — нет транзакций, аудита и атомарного apply/rollback.
## Последствия
- Миграции схемы обязательны с первого дня (#20) и проверяются при старте.
- Backup/restore = копирование `/data` и `/config` — отражается в runbook (#45).
- История метрик растёт: нужен retention и периодическая очистка (#28).
-26
View File
@@ -1,26 +0,0 @@
# 007. Web-сервер: встроенный uvicorn, без nginx
- **Статус:** принято
- **Контекст/issue:** #15, #31
## Контекст
UI — SPA, API — REST и WebSocket, всё в одном контейнере. Нагрузка: LAN, единичный пользователь.
Установка должна быть в одну команду `docker run` без дополнительных контейнеров.
## Решение
Один uvicorn-процесс `awg-api` отдаёт и API, и собранную статику UI, и WebSocket. Сборка UI выполняется
на этапе билда образа и попадает в `ui/dist`. Закладывается совместимость с внешним reverse-proxy:
поддержка base-path и заголовков `X-Forwarded-*`, без обязательной настройки хоста.
## Альтернативы
- **nginx в образе** — плюс к размеру и к конфигу, преимуществ при такой нагрузке нет.
- **Отдельный контейнер для UI** — нарушает требование единого образа.
## Последствия
- Заголовки кэширования статики и `index.html` настраиваются в FastAPI (иначе UI кэшируется после обновления).
- WebSocket идёт через тот же порт: учитывается при настройке reverse-proxy в runbook (#45).
- Docker HEALTHCHECK опирается на `/api/v1/health` того же процесса (#31).
-31
View File
@@ -1,31 +0,0 @@
# 008. Безопасность API/UI: токен, CSRF, rate limit, маскировка секретов
- **Статус:** принято
- **Контекст/issue:** #15, #34
## Контекст
Приложение управляет VPN-туннелем и принимает приватные ключи AmneziaWG. Оно доступно из LAN,
а в перспективе может быть выставлено через reverse-proxy. В ТЗ этот слой не описан, но без него
загрузка `.conf` через веб-интерфейс превращается в дыру.
## Решение
- API — токен доступа в заголовке (Bearer), токен только в виде хеша в БД;
- UI — сессия + CSRF-токен на мутирующие запросы;
- приватные ключи (`PrivateKey`, `PresharedKey`) — `writeOnly` в OpenAPI, маска в UI и логах,
файлы 0600 в отдельном каталоге;
- rate limit и lockout на `/auth`, ограничение допустимых источников, bind по умолчанию на LAN;
- audit log всех изменений конфигурации.
## Альтернативы
- **Без аутентификации, доступ только по LAN** — не защищает от заражённого устройства в сети и от
ошибочной публикации порта.
- **OIDC-провайдер** — не для локального однопользовательского приложения.
## Последствия
- Начальный токен генерируется при первом старте и выводится в лог один раз — это шаг в runbook (#45).
- Маскировка секретов проверяется автотестами (#41) и security-прогоном (#44).
- UI обязан корректно вести себя при 401 (переход на форму входа) — часть AC #35.
-16
View File
@@ -1,16 +0,0 @@
# ADR — архитектурные решения awg-dash
Один файл = одно решение. Формат: контекст -> решение -> последствия.
| ADR | Решение |
|---|---|
| [001](001-backend-python-fastapi.md) | Backend — Python 3.12 + FastAPI |
| [002](002-single-image-modular-monolith.md) | Модульный монолит в одном образе, 2 процесса |
| [003](003-reconcile-desired-state.md) | Декларативный reconcile вместо императивных скриптов |
| [004](004-geoip-ripe-provider.md) | GeoIP: RIPE Stat + локальный кэш + абстракция провайдера |
| [005](005-dns-dnsmasq.md) | DNS: встроенный dnsmasq + перехват порта 53 |
| [006](006-state-sqlite.md) | Состояние: SQLite (WAL) + /config + /data |
| [007](007-embedded-web-server.md) | Web-сервер: встроенный uvicorn, без nginx |
| [008](008-api-ui-security.md) | Безопасность API/UI: токен, CSRF, rate limit, маскировка секретов |
Новые решения добавляются со следующим номером и ссылкой на issue, для которого решение принято.
-1254
View File
File diff suppressed because it is too large Load Diff
-32
View File
@@ -1,32 +0,0 @@
[project]
name = "awg-dash"
version = "0.0.1"
description = "AmneziaWG gateway with intelligent routing: REST API + Web UI + split-tunnel"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []
[tool.ruff]
line-length = 100
target-version = "py312"
extend-exclude = ["ui", ".venv", "backend/tests"]
[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP", "B", "ASYNC", "S", "RUF"]
ignore = ["S101"]
[tool.ruff.lint.per-file-ignores]
"backend/tests/*" = ["S", "D"]
[tool.mypy]
python_version = "3.12"
mypy_path = "backend"
files = ["backend"]
strict = true
ignore_missing_imports = true
[tool.pytest.ini_options]
testpaths = ["backend/tests"]
pythonpath = ["backend"]
addopts = "-q --cov=awg_agent --cov=awg_api --cov-report=term-missing --cov-fail-under=70"
asyncio_mode = "auto"
-5
View File
@@ -1,5 +0,0 @@
# Зависимости контрактных тестов API (issue #19).
# После мержа каркаса монорепо (PR #47) строки переезжают в requirements-dev.txt.
jsonschema>=4.23
openapi-spec-validator>=0.7
PyYAML>=6
-8
View File
@@ -1,8 +0,0 @@
-r requirements.txt
ruff>=0.6.9
mypy>=1.11
pytest>=8.3
pytest-cov>=5.0
pytest-asyncio>=0.24
pre-commit>=3.8
-12
View File
@@ -1,12 +0,0 @@
# Runtime (контроль-плоскость)
fastapi>=0.115
uvicorn[standard]>=0.32
pydantic>=2.9
pydantic-settings>=2.5
aiosqlite>=0.20
httpx>=0.27
structlog>=24.4
prometheus-client>=0.21
# Сетевой слой вызывается через бинари (awg, iptables, ipset, dnsmasq),
# поэтому python-зависимостей для сети не требуется.
-13
View File
@@ -1,13 +0,0 @@
# tests/netns — сетевой тестовый стенд
Задача **#21**, используется тестами #41 и #43.
Цель: проверять split-tunnel без реального VPN-провайдера и без доступа к продакшн-сети.
План харнесса:
1. Создать netns: `client`, `gw` (шлюз с awg-интерфейсом), `uplink` (имитация ISP).
2. Поднять мок WireGuard-endpoint (userspace) вместо реального сервера.
3. Запустить шлюз в netns с теми же правилами, что в контейнере.
4. Проверить, куда ушёл пакет: интерфейс/метка/счётчики ipset и iptables.
Точки входа: `tests/netns/run.sh` (вызывается через `make test-netns`).
-14
View File
@@ -1,14 +0,0 @@
# ui — Web UI (React + TypeScript + Tailwind + shadcn-ui)
Тёмная тема, 6 страниц из ТЗ. Каркас — задача **#35**, страницы — #36–#39, E2E — #42.
| Маршрут | Страница | Задача |
|---|---|---|
| `/` | Dashboard: статус, трафик, handshake, аптайм | #36 |
| `/routing` | редактор правил, ipset, домены, режимы, DnD+diff | #37 |
| `/config` | загрузка `.conf`, валидация, маскировка ключей | #38 |
| `/devices` | устройства и политики | #39 |
| `/logs` | журнал, счётчики RETURN vs MARK | #39 |
| `/settings` | GeoIP-расписание, обновления, интеграции | #39 |
API-клиент генерируется из OpenAPI (#19) в `src/api/generated/`. Взаимодействие: REST + WebSocket.
-20
View File
@@ -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}"
-27
View File
@@ -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.
-1229
View File
File diff suppressed because it is too large Load Diff