diff --git a/backend/awg_api/store/README.md b/backend/awg_api/store/README.md new file mode 100644 index 0000000..631e124 --- /dev/null +++ b/backend/awg_api/store/README.md @@ -0,0 +1,66 @@ +# 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`). diff --git a/backend/awg_api/store/__init__.py b/backend/awg_api/store/__init__.py new file mode 100644 index 0000000..c13009f --- /dev/null +++ b/backend/awg_api/store/__init__.py @@ -0,0 +1,103 @@ +"""Слой хранения состояния 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", +] diff --git a/backend/awg_api/store/apply.py b/backend/awg_api/store/apply.py new file mode 100644 index 0000000..3ecc2fe --- /dev/null +++ b/backend/awg_api/store/apply.py @@ -0,0 +1,325 @@ +"""Атомарное применение состояния: ``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 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/clock.py b/backend/awg_api/store/clock.py new file mode 100644 index 0000000..0b3f33d --- /dev/null +++ b/backend/awg_api/store/clock.py @@ -0,0 +1,16 @@ +"""Источник времени для 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") diff --git a/backend/awg_api/store/db.py b/backend/awg_api/store/db.py new file mode 100644 index 0000000..bf03f2d --- /dev/null +++ b/backend/awg_api/store/db.py @@ -0,0 +1,175 @@ +"""Открытие и жизненный цикл 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() diff --git a/backend/awg_api/store/errors.py b/backend/awg_api/store/errors.py new file mode 100644 index 0000000..4e2c752 --- /dev/null +++ b/backend/awg_api/store/errors.py @@ -0,0 +1,33 @@ +"""Иерархия ошибок хранилища 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 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)}") diff --git a/backend/awg_api/store/models.py b/backend/awg_api/store/models.py new file mode 100644 index 0000000..f095955 --- /dev/null +++ b/backend/awg_api/store/models.py @@ -0,0 +1,186 @@ +"""Модель данных хранилища 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, + } diff --git a/backend/awg_api/store/repository.py b/backend/awg_api/store/repository.py new file mode 100644 index 0000000..a21dcb2 --- /dev/null +++ b/backend/awg_api/store/repository.py @@ -0,0 +1,599 @@ +"""Репозиторий сущностей состояния (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 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 diff --git a/backend/tests/test_store_apply.py b/backend/tests/test_store_apply.py new file mode 100644 index 0000000..75bdbb0 --- /dev/null +++ b/backend/tests/test_store_apply.py @@ -0,0 +1,311 @@ +"""Тесты атомарного применения: 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() diff --git a/backend/tests/test_store_audit.py b/backend/tests/test_store_audit.py new file mode 100644 index 0000000..4de1802 --- /dev/null +++ b/backend/tests/test_store_audit.py @@ -0,0 +1,145 @@ +"""Тесты журнала аудита «кто/когда/что/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) == [] diff --git a/backend/tests/test_store_concurrency.py b/backend/tests/test_store_concurrency.py new file mode 100644 index 0000000..1b879cf --- /dev/null +++ b/backend/tests/test_store_concurrency.py @@ -0,0 +1,308 @@ +"""Тесты потокобезопасности слоя хранения (#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) diff --git a/backend/tests/test_store_migrations.py b/backend/tests/test_store_migrations.py new file mode 100644 index 0000000..b00bbfb --- /dev/null +++ b/backend/tests/test_store_migrations.py @@ -0,0 +1,153 @@ +"""Тесты схемы, миграций и проверки версии при старте (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") diff --git a/backend/tests/test_store_persistence.py b/backend/tests/test_store_persistence.py new file mode 100644 index 0000000..46d0edb --- /dev/null +++ b/backend/tests/test_store_persistence.py @@ -0,0 +1,184 @@ +"""Персистентность: рестарт «контейнера» сохраняет правила, конфиг и аудит (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", " ") diff --git a/backend/tests/test_store_repository.py b/backend/tests/test_store_repository.py new file mode 100644 index 0000000..ff54771 --- /dev/null +++ b/backend/tests/test_store_repository.py @@ -0,0 +1,312 @@ +"""Тесты репозитория: 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"