feat(store): миграции схемы v1, репозиторий с аудитом, атомарный apply/rollback (#20)

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

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)}")