From 34610b069c54b0893f04b574764d73da84d942b7 Mon Sep 17 00:00:00 2001 From: embeddevio Date: Wed, 21 Oct 2026 23:43:45 +0300 Subject: [PATCH] =?UTF-8?q?feat(store):=20=D0=BC=D0=B8=D0=B3=D1=80=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D0=B8=20=D1=81=D1=85=D0=B5=D0=BC=D1=8B=20v1,=20?= =?UTF-8?q?=D1=80=D0=B5=D0=BF=D0=BE=D0=B7=D0=B8=D1=82=D0=BE=D1=80=D0=B8?= =?UTF-8?q?=D0=B9=20=D1=81=20=D0=B0=D1=83=D0=B4=D0=B8=D1=82=D0=BE=D0=BC,?= =?UTF-8?q?=20=D0=B0=D1=82=D0=BE=D0=BC=D0=B0=D1=80=D0=BD=D1=8B=D0=B9=20app?= =?UTF-8?q?ly/rollback=20(#20)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/awg_api/store/apply.py | 298 +++++++++++++++++++++++++++ backend/awg_api/store/migrations.py | 299 ++++++++++++++++++++++++++++ 2 files changed, 597 insertions(+) create mode 100644 backend/awg_api/store/apply.py create mode 100644 backend/awg_api/store/migrations.py diff --git a/backend/awg_api/store/apply.py b/backend/awg_api/store/apply.py new file mode 100644 index 0000000..df239c8 --- /dev/null +++ b/backend/awg_api/store/apply.py @@ -0,0 +1,298 @@ +"""Атомарное применение состояния: ``validate -> stage -> apply -> rollback`` (issue #20). + +Гарантии: + +* **validate** — несостоятельное желаемое состояние не доходит до БД; +* **stage** — перед мутацией сохраняется снимок текущего состояния (durable, + в отдельной таблице ``state_snapshots``), поэтому откат возможен в любой момент; +* **apply** — мутация БД и запись в ``audit_log`` выполняются в одной транзакции, + внешнее применение (iptables/awg) вызывается внутри неё: падение внешнего + применителя откатывает транзакцию, то есть БД возвращается к предыдущему + состоянию целиком, без частичных изменений; +* **rollback** — явный возврат к снимку по id. + +Ошибка применения поднимает :class:`ApplyError` (поле ``restored=True``): API +должен ответить ошибкой, не «полуприменённой» конфигурацией. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from contextlib import suppress +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`` означает, + что предыдущее состояние подтверждённо восстановлено. + """ + 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 Exception as exc: + restored_ok = self._rollback_failed_apply(plan, exc) + 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: Exception) -> bool: + """Вернуть внешнюю подсистему и убедиться, что БД = предыдущее состояние.""" + with suppress(Exception): # откат внешней системы — best-effort + self._applier.revert(plan.previous) + 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=f"{type(exc).__name__}: {exc}", + ) + return state.capture_state(self._store.raw()) == plan.previous + + def _last_failed_audit_id(self, plan: StagedPlan) -> int | None: + entries = audit.list_entries( + self._store.raw(), 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 diff --git a/backend/awg_api/store/migrations.py b/backend/awg_api/store/migrations.py new file mode 100644 index 0000000..7fd594e --- /dev/null +++ b/backend/awg_api/store/migrations.py @@ -0,0 +1,299 @@ +"""Версионированные миграции схемы состояния (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)}")