diff --git a/backend/awg_api/store/README.md b/backend/awg_api/store/README.md new file mode 100644 index 0000000..970014d --- /dev/null +++ b/backend/awg_api/store/README.md @@ -0,0 +1,47 @@ +# 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` в той же транзакции. +- **Атомарность** — применяемое состояние проходит `validate → stage → apply`; падение внешнего применителя откатывает транзакцию БД и вызывает `Applier.revert(previous)`, `ApplyError.restored=True`. +- **Откат** — `ApplyEngine.rollback(plan_id)` возвращает состояние из снимка. + +## Пример + +```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`). diff --git a/backend/awg_api/store/audit.py b/backend/awg_api/store/audit.py new file mode 100644 index 0000000..3d77bd3 --- /dev/null +++ b/backend/awg_api/store/audit.py @@ -0,0 +1,125 @@ +"""Журнал аудита «кто/когда/что/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 diff --git a/backend/awg_api/store/db.py b/backend/awg_api/store/db.py new file mode 100644 index 0000000..8d0743a --- /dev/null +++ b/backend/awg_api/store/db.py @@ -0,0 +1,139 @@ +"""Открытие и жизненный цикл SQLite-хранилища состояния (issue #20, ADR-006). + +Путь по умолчанию — ``/data/awg-dash.db`` (персистентный volume контейнера), +переопределяется аргументом или переменной окружения ``AWG_DASH_DB``. +Каталог создаётся автоматически, поэтому на пустом ``/data`` схема v1 +поднимается без ручных действий. + +Соединение открывается в режиме WAL: читатели не блокируют писателя, +``synchronous=NORMAL`` достаточен для WAL (риск потери только последних +транзакций при падении операционной системы, БД при этом не разрушается). +""" + +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: + """Открытое соединение (для сервисного слоя).""" + return self._conn + + @property + def journal_mode(self) -> str: + 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: + 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: + """Проверить соответствие фактической схемы ожидаемой версии.""" + 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() + + 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() diff --git a/backend/awg_api/store/service.py b/backend/awg_api/store/service.py new file mode 100644 index 0000000..dbe9169 --- /dev/null +++ b/backend/awg_api/store/service.py @@ -0,0 +1,47 @@ +"""Точка входа слоя хранения: открыть БД, применить миграции, собрать движок.""" + +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) diff --git a/backend/awg_api/store/state.py b/backend/awg_api/store/state.py new file mode 100644 index 0000000..5ccfcce --- /dev/null +++ b/backend/awg_api/store/state.py @@ -0,0 +1,174 @@ +"""Снимки состояния и восстановление после отката (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 diff --git a/backend/awg_api/store/validation.py b/backend/awg_api/store/validation.py new file mode 100644 index 0000000..30c3801 --- /dev/null +++ b/backend/awg_api/store/validation.py @@ -0,0 +1,131 @@ +"""Чистые функции валидации состояния (стадия ``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