1 Commits

Author SHA1 Message Date
Hermes TechLead Bot 9687f440ff Add TechLead agent payload and templates 2026-10-16 23:04:14 +03:00
27 changed files with 183 additions and 3692 deletions
+30
View File
@@ -0,0 +1,30 @@
---
name: Epic
about: Большая задача/эпик — используйте для крупных фич и релизов
labels: epic
---
# Заголовок epica
Опишите цель эпика, scope и критерии приёма.
## Описание
(Кратко опишите проблему/цель)
## Критерии приёма (GIVEN / WHEN / THEN)
- GIVEN ...
- WHEN ...
- THEN ...
## Артефакты
- OpenAPI spec (если есть)
- Proposed image tag
- Доп. доки
## Дополнительно
Предполагаемые owners: @backend, @devops, @qa
Оценка: (story points)
+16
View File
@@ -0,0 +1,16 @@
---
name: Release Request
about: Запрос на релиз
labels: release
---
# Release Request
Версия:
Image tag:
Список PRs/links:
CI status link:
Test report link:
Релиз в staging: [ ]
Релиз в prod: [ ] (требует TechLead approval)
+18
View File
@@ -0,0 +1,18 @@
---
name: Task
about: Конкретная задача
labels: task
---
# Заголовок задачи
Описание:
Acceptance criteria:
- GIVEN
- WHEN
- THEN
Owner: @
Estimate (SP):
Priority: low/medium/high
+28
View File
@@ -0,0 +1,28 @@
# Labels for project
- name: epic
color: ffbf00
description: large feature or project
- name: backend
color: 0e8a16
description: backend related work
- name: frontend
color: 1d76db
description: frontend related work
- name: devops
color: e99695
description: infra / deployment
- name: qa
color: cfd3d7
description: testing
- name: task
color: ededed
description: generic task
- name: blocker
color: b60205
description: blocking issue
- name: release
color: 5319e7
description: release request
- name: hotfix
color: fbca04
description: urgent hotfix
+13
View File
@@ -0,0 +1,13 @@
# PR Title
Описание изменений:
Checklist:
- [ ] CI green
- [ ] Unit tests passed
- [ ] Integration tests (mock) passed
- [ ] Coverage >= 70%
- [ ] Security scan: no high/critical findings
- [ ] Assigned reviewer(s)
Ссылки на связанные issue:
+7
View File
@@ -0,0 +1,7 @@
# CODEOWNERS file
# Format: path @owner
/backend/ @backend
/frontend/ @frontend
/devops/ @devops
/tests/ @qa
-66
View File
@@ -1,66 +0,0 @@
# store — хранилище состояния (issue #20)
SQLite (WAL) в `/data/awg-dash.db`: правила маршрутизации, исключения, режим,
расписание GeoIP, метаданные конфигов, устройства, журнал аудита и снимки
состояния. Путь переопределяется аргументом или переменной `AWG_DASH_DB`.
| Модуль | Назначение |
|---|---|
| `db.py` | открытие соединения, PRAGMA (WAL, `synchronous=NORMAL`, `foreign_keys`, `busy_timeout`), транзакции `BEGIN IMMEDIATE` |
| `migrations.py` | версионированные миграции, `schema_migrations`, проверка схемы при старте (`verify_schema`) |
| `models.py` | dataclass-модели сущностей и перечисления (`Mode`, `Entity`, `Rule`, …) |
| `validation.py` | чистые функции валидации (domain/CIDR/ipset, MAC, cron, режим, политика) |
| `repository.py` | `Store`: чтения + аудируемые мутации «кто/когда/что/old→new» |
| `audit.py` | запись и чтение `audit_log` (мутация и её аудит — одна транзакция) |
| `state.py` | снимки состояния (`state_snapshots`), `capture_state`, `restore_state`, `rollback_to_snapshot` |
| `apply.py` | `ApplyEngine`: `validate → stage → apply → rollback` и протокол `Applier` (awg/iptables) |
| `service.py` | `open_storage()` — сборка `Database + Store + ApplyEngine` для старта приложения |
| `errors.py` | `StorageError`, `SchemaError`, `ValidationError`, `ApplyError(restored=…)` |
## Гарантии
- **Пустой `/data`** — каталог создаётся, миграции применяются, схема v1 появляется без ручных действий.
- **Версия схемы** — при старте схема проверяется: новее ожидаемой или не мигрированная → `SchemaError` (без «автоотката»).
- **Аудит** — каждая мутация пишет `actor/ts/action/entity/old_value/new_value` в `audit_log` в той же транзакции; `old_value` читается внутри той же транзакции, поэтому конкурентные обновления не рвут цепочку `old→new`.
- **Атомарность** — применяемое состояние проходит `validate → stage → apply`; падение внешнего применителя откатывает транзакцию БД и вызывает `Applier.revert(previous)`, `ApplyError.restored=True`. Ловится `BaseException`, поэтому отмена (`asyncio.CancelledError`), `KeyboardInterrupt` и `SystemExit` тоже дают `revert` и запись `failed` в журнале.
- **Откат** — `ApplyEngine.rollback(plan_id)` возвращает состояние из снимка.
## Потокобезопасность (#54)
Контракт: **одно соединение на процесс, все операции под одним `RLock`**.
| Операция | API | Поведение |
|---|---|---|
| Мутация | `Database.transaction()` | `BEGIN IMMEDIATE` … `COMMIT`/`ROLLBACK` под локом |
| Чтение | `Database.read()` | берёт тот же лок; ждёт завершения чужой транзакции |
| Прямой доступ | `Database.connection` / `Store.raw()` | лок **не** берётся — только однопоточный код (миграции, тесты, отладка) |
Следствия, на которые можно опираться в API-слое (#19/#21):
- обычные чтения (`GET /rules`, `GET /mode`, …) не падают от конкурентных мутаций и не отдают перемешанный результат операторов;
- читатель **не видит** незафиксированных строк другой транзакции;
- во время `apply` наружу не видно состояние, которое затем будет откачено (AC «не частично применённая конфигурация»);
- конкурентные запросы ждут, а не получают ошибку; потокобезопасность подтверждена `backend/tests/test_store_concurrency.py`.
WAL остаётся включённым: он даёт устойчивость к падению процесса и позволяет читать БД снаружи (`sqlite3` CLI) во время работы шлюза. Внутрипроцессного параллелизма чтений контракт не обещает: если позже понадобится параллельная пропускная способность, меняем контракт на «соединение на поток» (`threading.local`), а не полагаемся на неявное поведение.
## Пример
```python
from awg_api.store import DesiredState, Rule, open_storage
with open_storage() as storage: # /data/awg-dash.db
storage.store.set_geoip_schedule(cron="0 3 * * *", actor="ui:admin")
plan = storage.engine.stage(
DesiredState(
mode=storage.store.get_mode(), rules=(Rule("yt", "domain", "youtube.com", "tun0"),)
),
reason="ui: split-tunnel для YouTube",
)
result = storage.engine.apply(plan)
```
## Тесты
`backend/tests/test_store_*.py` — миграции, аудит, репозиторий, apply/rollback,
персистентность после рестарта. Порог покрытия — 70 % (`pyproject.toml`).
-103
View File
@@ -1,103 +0,0 @@
"""Слой хранения состояния awg-dash (issue #20).
Состав: SQLite (WAL) в ``/data/awg-dash.db``, версионированные миграции,
проверка схемы при старте, журнал аудита «кто/когда/что/old→new» и движок
атомарного применения ``validate -> stage -> apply -> rollback``.
"""
from __future__ import annotations
from .apply import (
Applier,
ApplyEngine,
ApplyResult,
DesiredState,
NullApplier,
StagedPlan,
)
from .audit import count as audit_count
from .audit import last_entry as audit_last_entry
from .audit import list_entries as list_audit_entries
from .audit import record as record_audit
from .db import DB_PATH_ENV, DEFAULT_DB_PATH, Database, resolve_db_path
from .errors import ApplyError, SchemaError, StorageError, ValidationError
from .migrations import (
EXPECTED_TABLES,
LATEST_VERSION,
MIGRATIONS,
SCHEMA_MIGRATIONS,
Migration,
MigrationReport,
current_version,
migrate,
verify_schema,
)
from .models import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
DEVICE_POLICIES,
RULE_KINDS,
AuditEntry,
ConfigMeta,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
from .repository import Store
from .service import Storage, open_storage, rollback_to
from .state import StateSnapshot, capture_state, restore_state, rollback_to_snapshot
__all__ = [
"AUDIT_STATUS_FAILED",
"AUDIT_STATUS_OK",
"AUDIT_STATUS_STAGED",
"DB_PATH_ENV",
"DEFAULT_DB_PATH",
"DEVICE_POLICIES",
"EXPECTED_TABLES",
"LATEST_VERSION",
"MIGRATIONS",
"RULE_KINDS",
"SCHEMA_MIGRATIONS",
"Applier",
"ApplyEngine",
"ApplyError",
"ApplyResult",
"AuditEntry",
"ConfigMeta",
"Database",
"DesiredState",
"Device",
"Entity",
"GeoIpSchedule",
"Migration",
"MigrationReport",
"Mode",
"NullApplier",
"Rule",
"RuleException",
"SchemaError",
"StagedPlan",
"StateSnapshot",
"Storage",
"StorageError",
"Store",
"ValidationError",
"audit_count",
"audit_last_entry",
"capture_state",
"current_version",
"list_audit_entries",
"migrate",
"open_storage",
"record_audit",
"resolve_db_path",
"restore_state",
"rollback_to",
"rollback_to_snapshot",
"verify_schema",
]
-325
View File
@@ -1,325 +0,0 @@
"""Атомарное применение состояния: ``validate -> stage -> apply -> rollback`` (issue #20).
Гарантии:
* **validate** — несостоятельное желаемое состояние не доходит до БД;
* **stage** — перед мутацией сохраняется снимок текущего состояния (durable,
в отдельной таблице ``state_snapshots``), поэтому откат возможен в любой момент;
* **apply** — мутация БД и запись в ``audit_log`` выполняются в одной транзакции,
внешнее применение (iptables/awg) вызывается внутри неё: падение внешнего
применителя откатывает транзакцию, то есть БД возвращается к предыдущему
состоянию целиком, без частичных изменений;
* **rollback** — явный возврат к снимку по id.
Ошибка применения поднимает :class:`ApplyError` (поле ``restored=True``): API
должен ответить ошибкой, не «полуприменённой» конфигурацией. Ловится
``BaseException`` (#55 D-1), поэтому отмена запроса тоже вызывает
``Applier.revert(previous)`` и закрывает запись аудита исходом ``failed`` —
«висящих» записей ``staged`` без исхода не остаётся.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable
from . import audit, state, validation
from .errors import ApplyError, ValidationError
from .models import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
from .repository import Store
from .state import StateSnapshot
@dataclass(frozen=True, slots=True)
class DesiredState:
"""Полное желаемое состояние шлюза (то, что применяется как единое целое)."""
mode: Mode
rules: tuple[Rule, ...] = ()
geoip: GeoIpSchedule = field(default_factory=GeoIpSchedule)
exceptions: tuple[RuleException, ...] = ()
def to_payload(self) -> dict[str, Any]:
return {
"mode": self.mode.value,
"geoip": self.geoip.as_dict(),
"rules": [
{
"name": rule.name,
"kind": rule.kind,
"match": rule.match,
"target": rule.target,
"priority": rule.priority,
"enabled": rule.enabled,
}
for rule in self.rules
],
"exceptions": [
{
"match": exception.match,
"reason": exception.reason,
"enabled": exception.enabled,
}
for exception in self.exceptions
],
}
@classmethod
def from_payload(cls, payload: Mapping[str, Any]) -> DesiredState:
"""Разобрать желаемое состояние из JSON-совместимого словаря."""
raw_rules = payload.get("rules", [])
if not isinstance(raw_rules, (list, tuple)):
raise ValidationError("rules", "ожидается список правил")
rules = tuple(_rule_from_mapping(item) for item in raw_rules)
raw_exceptions = payload.get("exceptions", [])
if not isinstance(raw_exceptions, (list, tuple)):
raise ValidationError("exceptions", "ожидается список исключений")
exceptions = tuple(_exception_from_mapping(item) for item in raw_exceptions)
geoip_raw = payload.get("geoip") or {}
if not isinstance(geoip_raw, Mapping):
raise ValidationError("geoip", "ожидается объект расписания")
geoip = GeoIpSchedule(
enabled=bool(geoip_raw.get("enabled", True)),
cron=str(geoip_raw.get("cron", "0 4 * * *")),
provider=str(geoip_raw.get("provider", "ripe")),
)
return cls(
mode=validation.validate_mode(str(payload.get("mode", Mode.OFF.value))),
rules=rules,
geoip=geoip,
exceptions=exceptions,
)
def _rule_from_mapping(item: Any) -> Rule:
if not isinstance(item, Mapping):
raise ValidationError("rules", f"ожидается объект правила, получено {item!r}")
try:
return Rule(
name=str(item["name"]),
kind=str(item["kind"]),
match=str(item["match"]),
target=str(item["target"]),
priority=int(item.get("priority", 100)),
enabled=bool(item.get("enabled", True)),
)
except KeyError as exc:
raise ValidationError("rules", f"в правиле нет обязательного поля {exc.args[0]!r}") from exc
def _exception_from_mapping(item: Any) -> RuleException:
if not isinstance(item, Mapping):
raise ValidationError("exceptions", f"ожидается объект исключения, получено {item!r}")
try:
return RuleException(
match=str(item["match"]),
reason=str(item.get("reason", "")),
enabled=bool(item.get("enabled", True)),
)
except KeyError as exc:
raise ValidationError(
"exceptions", f"в исключении нет обязательного поля {exc.args[0]!r}"
) from exc
@runtime_checkable
class Applier(Protocol):
"""Внешний применитель состояния (awg/iptables/ipset)."""
def apply(self, payload: Mapping[str, Any]) -> None:
"""Применить состояние во внешней подсистеме. Падение -> ApplyError."""
def revert(self, payload: Mapping[str, Any]) -> None:
"""Вернуть внешнюю подсистему к состоянию ``payload`` при откате."""
@dataclass(frozen=True, slots=True)
class StagedPlan:
"""Промежуточный план: снимок предыдущего состояния + желаемое состояние."""
plan_id: int
desired: DesiredState
previous: dict[str, Any]
actor: str
reason: str
@dataclass(frozen=True, slots=True)
class ApplyResult:
"""Успешный результат применения."""
ok: bool
plan_id: int
audit_id: int
message: str
class NullApplier:
"""Заглушка внешнего применителя (для тестов и dry-run режима)."""
def apply(self, payload: Mapping[str, Any]) -> None:
return None
def revert(self, payload: Mapping[str, Any]) -> None:
return None
class ApplyEngine:
"""Оркестратор ``validate -> stage -> apply -> rollback``."""
def __init__(
self, store: Store, applier: Applier | None = None, *, actor: str = "system"
) -> None:
self._store = store
self._applier: Applier = applier if applier is not None else NullApplier()
self._actor = actor
# -- стадии ---------------------------------------------------------
def validate(self, desired: DesiredState) -> DesiredState:
"""Проверить желаемое состояние, не касаясь БД."""
validation.validate_mode(desired.mode)
validation.validate_cron(desired.geoip.cron)
validation.validate_provider(desired.geoip.provider)
seen: set[str] = set()
for rule in desired.rules:
validation.validate_name("name", rule.name)
kind = validation.validate_rule_kind(rule.kind)
validation.validate_match(kind, rule.match)
validation.validate_target(rule.target)
validation.validate_priority(rule.priority)
if rule.name in seen:
raise ValidationError("rules", f"дублирующееся имя правила: {rule.name!r}")
seen.add(rule.name)
for exception in desired.exceptions:
validation.validate_exception_match(exception.match)
return desired
def stage(self, desired: DesiredState, *, reason: str = "") -> StagedPlan:
"""Проверить состояние и сохранить снимок предыдущего состояния."""
self.validate(desired)
payload = desired.to_payload()
with self._store.database.transaction() as conn:
previous = state.capture_state(conn)
plan_id = state.save_snapshot(conn, previous, reason=reason or "pre-apply")
audit.record(
conn,
actor=self._actor,
action="stage",
entity=Entity.APPLY.value,
entity_id=plan_id,
old_value=previous,
new_value=payload,
status=AUDIT_STATUS_STAGED,
detail=reason,
)
return StagedPlan(
plan_id=plan_id, desired=desired, previous=previous, actor=self._actor, reason=reason
)
def apply(self, plan: StagedPlan) -> ApplyResult:
"""Применить план. При падении — откатить БД и внешнюю подсистему.
:raises ApplyError: если применитель упал; ``restored=True`` означает,
что предыдущее состояние подтверждённо восстановлено.
Ловится ``BaseException``, не только ``Exception`` (#55 D-1): отмена
запроса (``asyncio.CancelledError``), ``KeyboardInterrupt`` и
``SystemExit`` тоже обязаны вернуть внешнюю подсистему в предыдущее
состояние и закрыть запись аудита исходом ``failed``. Иначе iptables
останется в новом состоянии, тогда как ``audit_log`` сохранит
«висящую» запись ``staged`` без исхода. Исключения из ``Exception``
преобразуются в :class:`ApplyError`, остальные пробрасываются как есть
(семантика отмены сохраняется).
"""
payload = plan.desired.to_payload()
audit_id: int
try:
with self._store.database.transaction() as conn:
state.restore_state(conn, payload)
self._applier.apply(payload)
audit_id = audit.record(
conn,
actor=plan.actor,
action="apply",
entity=Entity.APPLY.value,
entity_id=plan.plan_id,
old_value=plan.previous,
new_value=payload,
status=AUDIT_STATUS_OK,
detail=plan.reason,
)
except BaseException as exc:
restored_ok = self._rollback_failed_apply(plan, exc)
if not isinstance(exc, Exception):
raise
raise ApplyError(
f"применение состояния не удалось: {exc}",
restored=restored_ok,
audit_id=self._last_failed_audit_id(plan),
) from exc
return ApplyResult(
ok=True, plan_id=plan.plan_id, audit_id=audit_id, message="состояние применено"
)
def apply_payload(self, payload: Mapping[str, Any], *, reason: str = "") -> ApplyResult:
"""validate + stage + apply одной командой (для сервисного слоя)."""
desired = self.validate(DesiredState.from_payload(payload))
return self.apply(self.stage(desired, reason=reason))
def rollback(self, plan_id: int, *, reason: str = "") -> StateSnapshot:
"""Явно вернуть состояние к снимку плана."""
with self._store.database.transaction() as conn:
return state.rollback_to_snapshot(conn, plan_id, actor=self._actor, reason=reason)
# -- внутреннее -----------------------------------------------------
def _rollback_failed_apply(self, plan: StagedPlan, exc: BaseException) -> bool:
"""Вернуть внешнюю подсистему и убедиться, что БД = предыдущее состояние.
Откат внешней подсистемы — best-effort: падение ``revert`` не мешает
восстановить БД и записать исход в журнал (причина падения revert
попадает в ``detail``), и наоборот — падение revert не должно скрывать
исходное исключение.
"""
revert_error: BaseException | None = None
try:
self._applier.revert(plan.previous)
except BaseException as revert_exc: # откат best-effort: исход важнее падения revert
revert_error = revert_exc
detail = f"{type(exc).__name__}: {exc}"
if revert_error is not None:
detail += f"; revert: {type(revert_error).__name__}: {revert_error}"
with self._store.database.transaction() as conn:
state.restore_state(conn, plan.previous)
audit.record(
conn,
actor=plan.actor,
action="apply",
entity=Entity.APPLY.value,
entity_id=plan.plan_id,
old_value=plan.desired.to_payload(),
new_value=plan.previous,
status=AUDIT_STATUS_FAILED,
detail=detail,
)
with self._store.database.read() as conn:
return state.capture_state(conn) == plan.previous
def _last_failed_audit_id(self, plan: StagedPlan) -> int | None:
with self._store.database.read() as conn:
entries = audit.list_entries(
conn, entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=1
)
if not entries:
return None
last = entries[0]
return last.id if last.status == AUDIT_STATUS_FAILED else None
-125
View File
@@ -1,125 +0,0 @@
"""Журнал аудита «кто/когда/что/old->new» (issue #20).
Запись всегда выполняется на переданном соединении, то есть внутри уже
открытой транзакции вызывающего кода: мутация и её аудит коммитятся вместе и
не могут разойтись.
"""
from __future__ import annotations
import json
import sqlite3
from collections.abc import Mapping
from typing import Any
from .clock import utcnow
from .models import AUDIT_STATUS_OK, AuditEntry
__all__ = ["list_entries", "record"]
def _dumps(value: Any) -> str | None:
if value is None:
return None
if isinstance(value, Mapping):
return json.dumps(dict(value), sort_keys=True, ensure_ascii=False, default=str)
return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)
def _loads(value: Any) -> Any:
if value is None:
return None
if isinstance(value, (bytes, bytearray)):
value = value.decode("utf-8")
return json.loads(str(value))
def record(
conn: sqlite3.Connection,
*,
actor: str,
action: str,
entity: str,
entity_id: str | int | None = None,
old_value: Any = None,
new_value: Any = None,
status: str = AUDIT_STATUS_OK,
detail: str | None = None,
ts: str | None = None,
) -> int:
"""Записать событие в ``audit_log`` и вернуть id записи."""
if not actor:
raise ValueError("actor обязателен: аудит должен знать, кто изменил состояние")
cursor = conn.execute(
"""
INSERT INTO audit_log (ts, actor, action, entity, entity_id,
old_value, new_value, status, detail)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ts or utcnow(),
actor,
action,
entity,
None if entity_id is None else str(entity_id),
_dumps(old_value),
_dumps(new_value),
status,
detail,
),
)
return int(cursor.lastrowid or 0)
def list_entries(
conn: sqlite3.Connection,
*,
entity: str | None = None,
entity_id: str | int | None = None,
limit: int = 100,
) -> list[AuditEntry]:
"""Прочитать журнал аудита (свежие первыми)."""
sql = "SELECT * FROM audit_log"
clauses: list[str] = []
params: list[Any] = []
if entity is not None:
clauses.append("entity = ?")
params.append(entity)
if entity_id is not None:
clauses.append("entity_id = ?")
params.append(str(entity_id))
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY id DESC LIMIT ?"
params.append(int(limit))
rows = conn.execute(sql, params).fetchall()
return [
AuditEntry(
id=int(row["id"]),
ts=str(row["ts"]),
actor=str(row["actor"]),
action=str(row["action"]),
entity=str(row["entity"]),
entity_id=None if row["entity_id"] is None else str(row["entity_id"]),
old_value=_loads(row["old_value"]),
new_value=_loads(row["new_value"]),
status=str(row["status"]),
detail=None if row["detail"] is None else str(row["detail"]),
)
for row in rows
]
def count(conn: sqlite3.Connection, *, entity: str | None = None) -> int:
"""Сколько записей в журнале (опционально — по сущности)."""
if entity is None:
row = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()
else:
row = conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity = ?", (entity,)).fetchone()
return int(row[0]) if row is not None else 0
def last_entry(conn: sqlite3.Connection, *, entity: str | None = None) -> AuditEntry | None:
"""Последняя запись журнала (опционально — по сущности)."""
entries = list_entries(conn, entity=entity, limit=1)
return entries[0] if entries else None
-16
View File
@@ -1,16 +0,0 @@
"""Источник времени для audit_log и снимков состояния.
Отдельный модуль — чтобы тесты могли подменять время детерминированно
(``monkeypatch.setattr(clock, "utcnow", ...)``).
"""
from __future__ import annotations
from datetime import UTC, datetime
__all__ = ["utcnow"]
def utcnow() -> str:
"""Текущее время в UTC: ISO-8601, миллисекунды, суффикс ``Z``."""
return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
-175
View File
@@ -1,175 +0,0 @@
"""Открытие и жизненный цикл SQLite-хранилища состояния (issue #20, ADR-006).
Путь по умолчанию — ``/data/awg-dash.db`` (персистентный volume контейнера),
переопределяется аргументом или переменной окружения ``AWG_DASH_DB``.
Каталог создаётся автоматически, поэтому на пустом ``/data`` схема v1
поднимается без ручных действий.
Соединение открывается в режиме WAL: ``synchronous=NORMAL`` достаточен для WAL
(риск потери только последних транзакций при падении операционной системы, БД
при этом не разрушается).
Контракт потокобезопасности (#54): соединение одно на процесс, все операции —
и записи (:meth:`Database.transaction`), и чтения (:meth:`Database.read`) —
выполняются под одним ``RLock``. Поэтому:
* результаты операторов из разных потоков не перемешиваются (нет
``TypeError``/``IndexError``);
* читатель не видит незафиксированных строк чужой транзакции и состояния,
которое затем будет откачено (падение внешнего применителя);
* конкурентные запросы ждут, не падают.
WAL при этом остаётся: он даёт устойчивость к падению процесса и позволяет
читать БД снаружи (``sqlite3`` CLI) во время работы шлюза. Если позже
понадобится параллельная пропускная способность чтений, контракт меняется на
«соединение на поток» (``threading.local``) — это отдельное решение, не
молчаливое поведение.
"""
from __future__ import annotations
import os
import sqlite3
import threading
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from .errors import StorageError
from .migrations import MigrationReport, current_version, migrate, verify_schema
DEFAULT_DB_PATH = "/data/awg-dash.db"
DB_PATH_ENV = "AWG_DASH_DB"
BUSY_TIMEOUT_MS = 5000
_WAL_PRAGMAS: tuple[str, ...] = (
"PRAGMA journal_mode = WAL",
"PRAGMA synchronous = NORMAL",
"PRAGMA foreign_keys = ON",
"PRAGMA busy_timeout = 5000",
)
def resolve_db_path(path: str | os.PathLike[str] | None = None) -> Path:
"""Определить путь к БД: аргумент -> ``AWG_DASH_DB`` -> ``/data/awg-dash.db``."""
if path is None:
path = os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH
return Path(path)
class Database:
"""Соединение SQLite: миграции, проверка схемы, транзакции."""
def __init__(
self,
path: str | os.PathLike[str] | None = None,
*,
auto_migrate: bool = True,
) -> None:
self.path = resolve_db_path(path)
self._lock = threading.RLock()
self._conn = self._open(self.path)
self.migration_report: MigrationReport | None = None
if auto_migrate:
self.migrate()
else:
verify_schema(self._conn)
# -- соединение -----------------------------------------------------
@staticmethod
def _open(path: Path) -> sqlite3.Connection:
if str(path) != ":memory:":
try:
path.parent.mkdir(parents=True, exist_ok=True)
except OSError as exc:
raise StorageError(f"не удалось создать каталог {path.parent}: {exc}") from exc
try:
conn = sqlite3.connect(
str(path),
isolation_level=None,
check_same_thread=False,
timeout=BUSY_TIMEOUT_MS / 1000,
)
conn.row_factory = sqlite3.Row
for pragma in _WAL_PRAGMAS:
conn.execute(pragma)
except sqlite3.Error as exc:
raise StorageError(f"не удалось открыть БД {path}: {exc}") from exc
return conn
@property
def connection(self) -> sqlite3.Connection:
"""Прямой доступ к соединению — только для однопоточного кода.
Рантайм API/сервисного слоя использует :meth:`read` (чтения) и
:meth:`transaction` (мутации): они берут общий лок и гарантируют
согласованность результата (#54).
"""
return self._conn
@property
def journal_mode(self) -> str:
with self._lock:
row = self._conn.execute("PRAGMA journal_mode").fetchone()
return str(row[0]) if row is not None else ""
@property
def is_wal(self) -> bool:
return self.journal_mode.lower() == "wal"
@property
def schema_version(self) -> int:
with self._lock:
return current_version(self._conn)
# -- схема ----------------------------------------------------------
def migrate(self) -> MigrationReport:
"""Применить неприменённые миграции и проверить схему."""
with self._lock:
report = migrate(self._conn)
self.migration_report = report
return report
def verify_schema(self) -> None:
"""Проверить соответствие фактической схемы ожидаемой версии."""
with self._lock:
verify_schema(self._conn)
# -- транзакции и чтения --------------------------------------------
@contextmanager
def transaction(self) -> Iterator[sqlite3.Connection]:
"""Транзакция ALL-OR-NOTHING (``BEGIN IMMEDIATE`` + rollback при ошибке)."""
with self._lock:
self._conn.execute("BEGIN IMMEDIATE")
try:
yield self._conn
except BaseException:
self._conn.rollback()
raise
else:
self._conn.commit()
@contextmanager
def read(self) -> Iterator[sqlite3.Connection]:
"""Согласованное чтение под общим локом (#54).
Ждёт завершения чужой транзакции и потому не видит незафиксированных
строк и состояния, которое затем будет откачено. Внутри уже открытой
транзакции того же потока лок реентерабелен, поэтому читать можно и из
неё.
"""
with self._lock:
yield self._conn
def close(self) -> None:
with self._lock:
try:
self._conn.close()
except sqlite3.Error as exc: # pragma: no cover - защита от гонок при shutdown
raise StorageError(f"ошибка закрытия БД: {exc}") from exc
def __enter__(self) -> Database:
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
-33
View File
@@ -1,33 +0,0 @@
"""Иерархия ошибок хранилища awg-dash."""
from __future__ import annotations
class StorageError(RuntimeError):
"""Базовая ошибка слоя хранения состояния."""
class SchemaError(StorageError):
"""Схема БД не соответствует ожидаемой версии (или требует отката)."""
class ValidationError(StorageError):
"""Данные не прошли валидацию до применения (стадия validate)."""
def __init__(self, field: str, message: str) -> None:
super().__init__(f"{field}: {message}")
self.field = field
self.message = message
class ApplyError(StorageError):
"""Ошибка применения конфигурации.
``restored`` — предыдущее состояние подтверждённо восстановлено,
``audit_id`` — id записи в audit_log, статус ``failed``.
"""
def __init__(self, message: str, *, restored: bool, audit_id: int | None) -> None:
super().__init__(message)
self.restored = restored
self.audit_id = audit_id
-299
View File
@@ -1,299 +0,0 @@
"""Версионированные миграции схемы состояния (issue #20).
Схема БД управляется списком :class:`Migration`. Каждая миграция выполняется
в своей транзакции, запись версии в ``schema_migrations`` — там же: падение
миграции не оставляет «полуприменённой» схемы.
При старте приложения вызывается :func:`migrate` (создание/обновление схемы)
и :func:`verify_schema` (проверка, что фактическая схема соответствует
ожидаемой версии).
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from .clock import utcnow
from .errors import SchemaError, StorageError
SCHEMA_MIGRATIONS = "schema_migrations"
@dataclass(frozen=True, slots=True)
class Migration:
"""Одна версия схемы: имя и список DDL-выражений."""
version: int
name: str
statements: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class MigrationReport:
"""Результат :func:`migrate`."""
from_version: int
to_version: int
applied: tuple[int, ...]
@property
def changed(self) -> bool:
return bool(self.applied)
_V1_STATEMENTS: tuple[str, ...] = (
f"""
CREATE TABLE IF NOT EXISTS {SCHEMA_MIGRATIONS} (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS routing_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
kind TEXT NOT NULL CHECK (kind IN ('domain', 'cidr', 'ipset')),
match TEXT NOT NULL,
target TEXT NOT NULL,
priority INTEGER NOT NULL DEFAULT 100,
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS rule_exceptions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
match TEXT NOT NULL UNIQUE,
reason TEXT NOT NULL DEFAULT '',
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS routing_mode (
id INTEGER PRIMARY KEY CHECK (id = 1),
mode TEXT NOT NULL CHECK (mode IN ('off', 'split', 'full')),
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS geoip_schedule (
id INTEGER PRIMARY KEY CHECK (id = 1),
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
cron TEXT NOT NULL,
provider TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS configs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
version INTEGER NOT NULL DEFAULT 1,
path TEXT NOT NULL,
checksum TEXT NOT NULL DEFAULT '',
size_bytes INTEGER NOT NULL DEFAULT 0,
applied INTEGER NOT NULL DEFAULT 0 CHECK (applied IN (0, 1)),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mac TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
address TEXT NOT NULL DEFAULT '',
policy TEXT NOT NULL DEFAULT 'default',
last_seen TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
actor TEXT NOT NULL,
action TEXT NOT NULL,
entity TEXT NOT NULL,
entity_id TEXT,
old_value TEXT,
new_value TEXT,
status TEXT NOT NULL DEFAULT 'ok',
detail TEXT
)
""",
"CREATE INDEX IF NOT EXISTS idx_audit_log_entity ON audit_log (entity, entity_id)",
"CREATE INDEX IF NOT EXISTS idx_audit_log_ts ON audit_log (ts DESC)",
"""
CREATE TABLE IF NOT EXISTS state_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
entity TEXT NOT NULL,
entity_id TEXT NOT NULL,
created_at TEXT NOT NULL,
payload TEXT NOT NULL,
reason TEXT NOT NULL DEFAULT ''
)
""",
"CREATE INDEX IF NOT EXISTS idx_state_snapshots_key ON state_snapshots (entity, entity_id, id)",
)
V1_SEED_STATEMENTS: tuple[str, ...] = (
"INSERT OR IGNORE INTO routing_mode (id, mode, updated_at) VALUES (1, 'off',"
" strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
"INSERT OR IGNORE INTO geoip_schedule (id, enabled, cron, provider, updated_at) VALUES"
" (1, 1, '0 4 * * *', 'ripe', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
)
MIGRATIONS: tuple[Migration, ...] = (
Migration(
version=1,
name="initial-state-schema",
statements=(*_V1_STATEMENTS, *V1_SEED_STATEMENTS),
),
)
LATEST_VERSION: int = max(migration.version for migration in MIGRATIONS)
#: Ожидаемые таблицы схемы v1 и обязательные колонки каждой из них.
EXPECTED_TABLES: dict[str, frozenset[str]] = {
SCHEMA_MIGRATIONS: frozenset({"version", "name", "applied_at"}),
"routing_rules": frozenset(
{"id", "name", "kind", "match", "target", "priority", "enabled", "created_at", "updated_at"}
),
"rule_exceptions": frozenset({"id", "match", "reason", "enabled", "created_at", "updated_at"}),
"routing_mode": frozenset({"id", "mode", "updated_at"}),
"geoip_schedule": frozenset({"id", "enabled", "cron", "provider", "updated_at"}),
"configs": frozenset(
{
"id",
"name",
"version",
"path",
"checksum",
"size_bytes",
"applied",
"created_at",
"updated_at",
}
),
"devices": frozenset(
{"id", "mac", "name", "address", "policy", "last_seen", "created_at", "updated_at"}
),
"audit_log": frozenset(
{
"id",
"ts",
"actor",
"action",
"entity",
"entity_id",
"old_value",
"new_value",
"status",
"detail",
}
),
"state_snapshots": frozenset({"id", "entity", "entity_id", "created_at", "payload", "reason"}),
}
def table_names(conn: sqlite3.Connection) -> set[str]:
"""Имена таблиц, фактически присутствующих в БД."""
rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
return {str(row[0]) for row in rows}
def table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
"""Колонки таблицы (через параметризованное table-valued PRAGMA)."""
rows = conn.execute("SELECT name FROM pragma_table_info(?)", (table,)).fetchall()
return {str(row[0]) for row in rows}
def current_version(conn: sqlite3.Connection) -> int:
"""Текущая версия схемы; ``0`` — если таблицы версий ещё нет."""
if SCHEMA_MIGRATIONS not in table_names(conn):
return 0
row = conn.execute(f"SELECT MAX(version) FROM {SCHEMA_MIGRATIONS}").fetchone() # noqa: S608
if row is None or row[0] is None:
return 0
return int(row[0])
def applied_migrations(conn: sqlite3.Connection) -> dict[int, str]:
"""Карта «версия -> имя» для применённых миграций."""
if SCHEMA_MIGRATIONS not in table_names(conn):
return {}
rows = conn.execute(f"SELECT version, name FROM {SCHEMA_MIGRATIONS}").fetchall() # noqa: S608
return {int(row[0]): str(row[1]) for row in rows}
def apply_migration(conn: sqlite3.Connection, migration: Migration) -> None:
"""Применить одну миграцию в отдельной транзакции (all-or-nothing)."""
conn.execute("BEGIN IMMEDIATE")
try:
for statement in migration.statements:
conn.execute(statement)
conn.execute(
f"INSERT INTO {SCHEMA_MIGRATIONS} (version, name, applied_at) VALUES (?, ?, ?)", # noqa: S608
(migration.version, migration.name, utcnow()),
)
except sqlite3.Error as exc: # pragma: no cover - защита от порчи БД
conn.rollback()
raise StorageError(
f"миграция v{migration.version} ({migration.name}) не применена: {exc}"
) from exc
except Exception:
conn.rollback()
raise
conn.commit()
def migrate(
conn: sqlite3.Connection, *, migrations: tuple[Migration, ...] = MIGRATIONS
) -> MigrationReport:
"""Применить все неприменённые миграции по возрастанию версии."""
start = current_version(conn)
known = sorted(migrations, key=lambda migration: migration.version)
latest = max((migration.version for migration in known), default=0)
if start > latest:
raise SchemaError(
f"схема БД новее, чем известно приложению: v{start} > v{latest};"
" требуется обновление образа, не автооткат"
)
applied: list[int] = []
for migration in known:
if migration.version <= start:
continue
apply_migration(conn, migration)
applied.append(migration.version)
verify_schema(conn)
return MigrationReport(
from_version=start, to_version=current_version(conn), applied=tuple(applied)
)
def verify_schema(conn: sqlite3.Connection) -> None:
"""Проверить, что фактическая схема соответствует ожидаемой версии.
Вызывает :class:`SchemaError`, если версия схемы новее ожидаемой или если
отсутствует таблица/колонка, на которую рассчитывает приложение.
"""
version = current_version(conn)
if version > LATEST_VERSION:
raise SchemaError(f"схема БД v{version} новее ожидаемой v{LATEST_VERSION}")
if version < LATEST_VERSION:
raise SchemaError(f"схема БД v{version} не мигрирована до v{LATEST_VERSION}")
existing = table_names(conn)
missing_tables = sorted(set(EXPECTED_TABLES) - existing)
if missing_tables:
raise SchemaError(f"в схеме отсутствуют таблицы: {', '.join(missing_tables)}")
for table, required in EXPECTED_TABLES.items():
missing_columns = sorted(required - table_columns(conn, table))
if missing_columns:
raise SchemaError(f"таблица {table}: отсутствуют колонки {', '.join(missing_columns)}")
-186
View File
@@ -1,186 +0,0 @@
"""Модель данных хранилища awg-dash (issue #20).
Каждая сущность описана неизменяемым dataclass-ом; идентификатор ``None``
означает «ещё не сохранено». Строки времени — UTC ISO-8601, суффикс ``Z``.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
RULE_KINDS: tuple[str, ...] = ("domain", "cidr", "ipset")
DEVICE_POLICIES: tuple[str, ...] = ("default", "tunnel", "direct", "block")
AUDIT_STATUS_OK = "ok"
AUDIT_STATUS_FAILED = "failed"
AUDIT_STATUS_STAGED = "staged"
class Mode(StrEnum):
"""Режим маршрутизации шлюза."""
OFF = "off"
SPLIT = "split"
FULL = "full"
@classmethod
def values(cls) -> tuple[str, ...]:
return tuple(member.value for member in cls)
class Entity(StrEnum):
"""Имена сущностей для audit_log и снимков состояния."""
RULE = "rule"
EXCEPTION = "exception"
MODE = "mode"
GEOIP_SCHEDULE = "geoip_schedule"
CONFIG = "config"
DEVICE = "device"
APPLY = "apply"
@dataclass(frozen=True, slots=True)
class Rule:
"""Правило маршрутизации (domain / cidr / ipset -> target)."""
name: str
kind: str
match: str
target: str
priority: int = 100
enabled: bool = True
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"kind": self.kind,
"match": self.match,
"target": self.target,
"priority": self.priority,
"enabled": self.enabled,
}
@dataclass(frozen=True, slots=True)
class RuleException:
"""Исключение из правил (host/domain, который никогда не маршрутизируется в туннель)."""
match: str
reason: str = ""
enabled: bool = True
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"match": self.match,
"reason": self.reason,
"enabled": self.enabled,
}
@dataclass(frozen=True, slots=True)
class GeoIpSchedule:
"""Расписание обновления GeoIP-наборов."""
enabled: bool = True
cron: str = "0 4 * * *"
provider: str = "ripe"
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"enabled": self.enabled,
"cron": self.cron,
"provider": self.provider,
}
@dataclass(frozen=True, slots=True)
class ConfigMeta:
"""Метаданные сгенерированного конфига (сам файл живёт в /config)."""
name: str
path: str
version: int = 1
checksum: str = ""
size_bytes: int = 0
applied: bool = False
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"path": self.path,
"version": self.version,
"checksum": self.checksum,
"size_bytes": self.size_bytes,
"applied": self.applied,
}
@dataclass(frozen=True, slots=True)
class Device:
"""Известный клиент шлюза."""
mac: str
name: str
address: str = ""
policy: str = "default"
last_seen: str | None = None
id: int | None = None
created_at: str | None = None
updated_at: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"mac": self.mac,
"name": self.name,
"address": self.address,
"policy": self.policy,
"last_seen": self.last_seen,
}
@dataclass(frozen=True, slots=True)
class AuditEntry:
"""Запись журнала «кто/когда/что/old->new»."""
id: int
ts: str
actor: str
action: str
entity: str
entity_id: str | None
old_value: Any
new_value: Any
status: str
detail: str | None
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"ts": self.ts,
"actor": self.actor,
"action": self.action,
"entity": self.entity,
"entity_id": self.entity_id,
"old_value": self.old_value,
"new_value": self.new_value,
"status": self.status,
"detail": self.detail,
}
-599
View File
@@ -1,599 +0,0 @@
"""Репозиторий сущностей состояния (issue #20).
Каждая мутация выполняется в одной транзакции, запись в ``audit_log``
(«кто/когда/что/old→new») коммитится там же. Чтения идут через
:meth:`Database.read` — под тем же ``RLock``, что и транзакции (#54), поэтому
читатель не видит незафиксированных строк и состояния, которое затем будет
откачено, и не может получить перемешанный результат операторов другого потока.
"""
from __future__ import annotations
import sqlite3
from dataclasses import replace
from typing import cast
from . import audit, validation
from .clock import utcnow
from .db import Database
from .errors import StorageError
from .models import (
AuditEntry,
ConfigMeta,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
)
def _rule_from_row(row: sqlite3.Row) -> Rule:
return Rule(
id=int(row["id"]),
name=str(row["name"]),
kind=str(row["kind"]),
match=str(row["match"]),
target=str(row["target"]),
priority=int(row["priority"]),
enabled=bool(row["enabled"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _exception_from_row(row: sqlite3.Row) -> RuleException:
return RuleException(
id=int(row["id"]),
match=str(row["match"]),
reason=str(row["reason"]),
enabled=bool(row["enabled"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _config_from_row(row: sqlite3.Row) -> ConfigMeta:
return ConfigMeta(
id=int(row["id"]),
name=str(row["name"]),
path=str(row["path"]),
version=int(row["version"]),
checksum=str(row["checksum"]),
size_bytes=int(row["size_bytes"]),
applied=bool(row["applied"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _device_from_row(row: sqlite3.Row) -> Device:
return Device(
id=int(row["id"]),
mac=str(row["mac"]),
name=str(row["name"]),
address=str(row["address"]),
policy=str(row["policy"]),
last_seen=None if row["last_seen"] is None else str(row["last_seen"]),
created_at=str(row["created_at"]),
updated_at=str(row["updated_at"]),
)
def _geoip_from_row(conn: sqlite3.Connection) -> GeoIpSchedule:
row = conn.execute(
"SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1"
).fetchone()
if row is None:
return GeoIpSchedule()
return GeoIpSchedule(
enabled=bool(row["enabled"]),
cron=str(row["cron"]),
provider=str(row["provider"]),
updated_at=str(row["updated_at"]),
)
class Store:
"""Доступ к состоянию: чтение + аудируемые мутации."""
def __init__(self, db: Database) -> None:
self._db = db
@property
def database(self) -> Database:
return self._db
# -- чтение ---------------------------------------------------------
# Чтения идут через Database.read() под тем же RLock, что транзакции
# (#54): это исключает перемешивание операторов разных потоков и грязные
# чтения незафиксированных строк.
def get_mode(self) -> Mode:
with self._db.read() as conn:
row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
if row is None:
return Mode.OFF
return Mode(str(row["mode"]))
def get_geoip_schedule(self) -> GeoIpSchedule:
with self._db.read() as conn:
return _geoip_from_row(conn)
def list_rules(self, *, enabled_only: bool = False) -> list[Rule]:
sql = "SELECT * FROM routing_rules"
if enabled_only:
sql += " WHERE enabled = 1"
sql += " ORDER BY priority, id"
with self._db.read() as conn:
rows = conn.execute(sql).fetchall()
return [_rule_from_row(row) for row in rows]
def get_rule(self, rule_id: int) -> Rule | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
return None if row is None else _rule_from_row(row)
def get_rule_by_name(self, name: str) -> Rule | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
return None if row is None else _rule_from_row(row)
def list_exceptions(self) -> list[RuleException]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall()
return [_exception_from_row(row) for row in rows]
def get_exception(self, match: str) -> RuleException | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
return None if row is None else _exception_from_row(row)
def list_configs(self) -> list[ConfigMeta]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM configs ORDER BY name").fetchall()
return [_config_from_row(row) for row in rows]
def get_config(self, name: str) -> ConfigMeta | None:
with self._db.read() as conn:
row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
return None if row is None else _config_from_row(row)
def list_devices(self) -> list[Device]:
with self._db.read() as conn:
rows = conn.execute("SELECT * FROM devices ORDER BY name").fetchall()
return [_device_from_row(row) for row in rows]
def get_device(self, mac: str) -> Device | None:
normalized = validation.normalize_mac(mac)
with self._db.read() as conn:
row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
return None if row is None else _device_from_row(row)
def list_audit(
self, *, entity: str | None = None, entity_id: str | int | None = None, limit: int = 100
) -> list[AuditEntry]:
with self._db.read() as conn:
return audit.list_entries(conn, entity=entity, entity_id=entity_id, limit=limit)
# -- мутации: правила ------------------------------------------------
def upsert_rule(self, rule: Rule, *, actor: str) -> Rule:
"""Создать или обновить правило (по ``id`` либо по уникальному ``name``)."""
validation.validate_name("name", rule.name)
kind = validation.validate_rule_kind(rule.kind)
match = validation.validate_match(kind, rule.match)
target = validation.validate_target(rule.target)
priority = validation.validate_priority(rule.priority)
now = utcnow()
with self._db.transaction() as conn:
existing = self._find_rule_row(conn, rule.id, rule.name)
core: tuple[str, str, str, str, int, int] = (
rule.name,
kind,
match,
target,
priority,
1 if rule.enabled else 0,
)
if existing is None:
cursor = conn.execute(
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(*core, now, now),
)
new_id = int(cursor.lastrowid or 0)
saved = replace(
rule,
id=new_id,
kind=kind,
match=match,
target=target,
priority=priority,
created_at=now,
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.RULE.value,
entity_id=new_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
rule_id = int(existing["id"])
conn.execute(
"UPDATE routing_rules SET name = ?, kind = ?, match = ?, target = ?,"
" priority = ?, enabled = ?, updated_at = ? WHERE id = ?",
(*core, now, rule_id),
)
saved = replace(
rule,
id=rule_id,
kind=kind,
match=match,
target=target,
priority=priority,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.RULE.value,
entity_id=rule_id,
old_value=_rule_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_rule(self, rule_id: int, *, actor: str) -> bool:
"""Удалить правило; ``False`` — если правила не было."""
with self._db.transaction() as conn:
row = self._find_rule_row(conn, rule_id, None)
if row is None:
return False
removed = _rule_from_row(row)
conn.execute("DELETE FROM routing_rules WHERE id = ?", (int(row["id"]),))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.RULE.value,
entity_id=int(row["id"]),
old_value=removed.as_dict(),
new_value=None,
)
return True
@staticmethod
def _find_rule_row(
conn: sqlite3.Connection, rule_id: int | None, name: str | None
) -> sqlite3.Row | None:
if rule_id is not None:
row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
return cast("sqlite3.Row | None", row)
if name is not None:
row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
return cast("sqlite3.Row | None", row)
return None
# -- мутации: исключения ---------------------------------------------
def upsert_exception(self, exception: RuleException, *, actor: str) -> RuleException:
match = validation.validate_exception_match(exception.match)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute(
"SELECT * FROM rule_exceptions WHERE match = ?", (match,)
).fetchone()
if existing is None:
conn.execute(
"INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)"
" VALUES (?, ?, ?, ?, ?)",
(match, exception.reason, 1 if exception.enabled else 0, now, now),
)
else:
conn.execute(
"UPDATE rule_exceptions SET reason = ?, enabled = ?, updated_at = ?"
" WHERE match = ?",
(exception.reason, 1 if exception.enabled else 0, now, match),
)
row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
if row is None: # pragma: no cover - невозможен после успешной записи
raise StorageError("правило-исключение исчезло сразу после записи")
saved = _exception_from_row(row)
audit.record(
conn,
actor=actor,
action="create" if existing is None else "update",
entity=Entity.EXCEPTION.value,
entity_id=saved.id,
old_value=None if existing is None else _exception_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_exception(self, exception_id: int, *, actor: str) -> bool:
with self._db.transaction() as conn:
row = conn.execute(
"SELECT * FROM rule_exceptions WHERE id = ?", (exception_id,)
).fetchone()
if row is None:
return False
removed = _exception_from_row(row)
conn.execute("DELETE FROM rule_exceptions WHERE id = ?", (exception_id,))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.EXCEPTION.value,
entity_id=exception_id,
old_value=removed.as_dict(),
new_value=None,
)
return True
# -- мутации: режим и GeoIP ------------------------------------------
def set_mode(self, mode: str | Mode, *, actor: str) -> Mode:
"""Установить режим маршрутизации; в журнал пишется old→new."""
validated = validation.validate_mode(mode)
with self._db.transaction() as conn:
previous = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
old = "off" if previous is None else str(previous["mode"])
now = utcnow()
conn.execute(
"INSERT INTO routing_mode (id, mode, updated_at) VALUES (1, ?, ?)"
" ON CONFLICT(id) DO UPDATE SET mode = excluded.mode,"
" updated_at = excluded.updated_at",
(validated.value, now),
)
audit.record(
conn,
actor=actor,
action="set_mode",
entity=Entity.MODE.value,
entity_id=1,
old_value={"mode": old},
new_value={"mode": validated.value},
)
return validated
def set_geoip_schedule(
self,
*,
enabled: bool | None = None,
cron: str | None = None,
provider: str | None = None,
actor: str,
) -> GeoIpSchedule:
"""Обновить расписание GeoIP (частичное обновление) и записать аудит."""
now = utcnow()
with self._db.transaction() as conn:
# чтение old_value — внутри той же транзакции, что и запись (#55 D-2):
# иначе при конкурентном обновлении журнал заявит устаревшее «старое».
current = _geoip_from_row(conn)
stored = GeoIpSchedule(
enabled=current.enabled if enabled is None else enabled,
cron=validation.validate_cron(current.cron if cron is None else cron),
provider=validation.validate_provider(
current.provider if provider is None else provider
),
updated_at=now,
)
conn.execute(
"INSERT INTO geoip_schedule (id, enabled, cron, provider, updated_at)"
" VALUES (1, ?, ?, ?, ?)"
" ON CONFLICT(id) DO UPDATE SET enabled = excluded.enabled, cron = excluded.cron,"
" provider = excluded.provider, updated_at = excluded.updated_at",
(1 if stored.enabled else 0, stored.cron, stored.provider, now),
)
audit.record(
conn,
actor=actor,
action="set_geoip_schedule",
entity=Entity.GEOIP_SCHEDULE.value,
entity_id=1,
old_value=current.as_dict(),
new_value=stored.as_dict(),
)
return stored
# -- мутации: конфиги -------------------------------------------------
def upsert_config(self, config: ConfigMeta, *, actor: str) -> ConfigMeta:
name = validation.validate_name("name", config.name)
path = validation.validate_config_path(config.path)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
if existing is None:
cursor = conn.execute(
"INSERT INTO configs (name, version, path, checksum, size_bytes, applied,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
name,
config.version,
path,
config.checksum,
config.size_bytes,
1 if config.applied else 0,
now,
now,
),
)
config_id = int(cursor.lastrowid or 0)
saved = replace(
config, id=config_id, name=name, path=path, created_at=now, updated_at=now
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.CONFIG.value,
entity_id=config_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
config_id = int(existing["id"])
conn.execute(
"UPDATE configs SET version = ?, path = ?, checksum = ?, size_bytes = ?,"
" applied = ?, updated_at = ? WHERE id = ?",
(
config.version,
path,
config.checksum,
config.size_bytes,
1 if config.applied else 0,
now,
config_id,
),
)
saved = replace(
config,
id=config_id,
name=name,
path=path,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.CONFIG.value,
entity_id=config_id,
old_value=_config_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def mark_config_applied(self, name: str, *, applied: bool, actor: str) -> ConfigMeta:
"""Отметить конфиг применённым/снятым и записать аудит."""
with self._db.transaction() as conn:
row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
if row is None:
raise StorageError(f"конфиг {name!r} не найден")
now = utcnow()
conn.execute(
"UPDATE configs SET applied = ?, updated_at = ? WHERE id = ?",
(1 if applied else 0, now, int(row["id"])),
)
updated = replace(_config_from_row(row), applied=applied, updated_at=now)
audit.record(
conn,
actor=actor,
action="set_applied",
entity=Entity.CONFIG.value,
entity_id=int(row["id"]),
old_value=_config_from_row(row).as_dict(),
new_value=updated.as_dict(),
)
return updated
# -- мутации: устройства ----------------------------------------------
def register_device(self, device: Device, *, actor: str) -> Device:
mac = validation.normalize_mac(device.mac)
name = validation.validate_name("name", device.name)
policy = validation.validate_device_policy(device.policy)
now = utcnow()
with self._db.transaction() as conn:
existing = conn.execute("SELECT * FROM devices WHERE mac = ?", (mac,)).fetchone()
if existing is None:
cursor = conn.execute(
"INSERT INTO devices (mac, name, address, policy, last_seen, created_at,"
" updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(mac, name, device.address, policy, device.last_seen, now, now),
)
device_id = int(cursor.lastrowid or 0)
saved = replace(
device,
id=device_id,
mac=mac,
name=name,
policy=policy,
created_at=now,
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="create",
entity=Entity.DEVICE.value,
entity_id=device_id,
old_value=None,
new_value=saved.as_dict(),
)
else:
device_id = int(existing["id"])
conn.execute(
"UPDATE devices SET name = ?, address = ?, policy = ?, last_seen = ?,"
" updated_at = ? WHERE id = ?",
(name, device.address, policy, device.last_seen, now, device_id),
)
saved = replace(
device,
id=device_id,
mac=mac,
name=name,
policy=policy,
created_at=str(existing["created_at"]),
updated_at=now,
)
audit.record(
conn,
actor=actor,
action="update",
entity=Entity.DEVICE.value,
entity_id=device_id,
old_value=_device_from_row(existing).as_dict(),
new_value=saved.as_dict(),
)
return saved
def delete_device(self, mac: str, *, actor: str) -> bool:
normalized = validation.normalize_mac(mac)
with self._db.transaction() as conn:
row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
if row is None:
return False
removed = _device_from_row(row)
conn.execute("DELETE FROM devices WHERE mac = ?", (normalized,))
audit.record(
conn,
actor=actor,
action="delete",
entity=Entity.DEVICE.value,
entity_id=removed.id,
old_value=removed.as_dict(),
new_value=None,
)
return True
# -- служебное --------------------------------------------------------
def counts(self) -> dict[str, int]:
"""Количество строк в изменяемых таблицах (для отчётов и тестов)."""
tables = {
"rules": "routing_rules",
"exceptions": "rule_exceptions",
"configs": "configs",
"devices": "devices",
"audit": "audit_log",
"snapshots": "state_snapshots",
}
result: dict[str, int] = {}
with self._db.read() as conn: # один снимок на все таблицы: счётчики согласованы
for key, table in tables.items():
row = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone() # noqa: S608
result[key] = int(row[0]) if row is not None else 0
return result
def raw(self) -> sqlite3.Connection:
"""Соединение для кода, которому нужен доступ ниже уровня репозитория.
Лок не берётся: в рантайме используйте :meth:`Database.read` /
:meth:`Database.transaction` (#54), иначе возможны гонки на общем
соединении. Метод оставлен для однопоточного кода и миграций.
"""
return self._db.connection
-47
View File
@@ -1,47 +0,0 @@
"""Точка входа слоя хранения: открыть БД, применить миграции, собрать движок."""
from __future__ import annotations
import os
from dataclasses import dataclass
from .apply import Applier, ApplyEngine
from .db import Database
from .repository import Store
from .state import StateSnapshot
@dataclass(slots=True)
class Storage:
"""Собранный слой хранения для приложения."""
db: Database
store: Store
engine: ApplyEngine
def close(self) -> None:
self.db.close()
def __enter__(self) -> Storage:
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
def open_storage(
path: str | os.PathLike[str] | None = None,
*,
applier: Applier | None = None,
actor: str = "system",
auto_migrate: bool = True,
) -> Storage:
"""Открыть хранилище: на пустом ``/data`` схема поднимается автоматически."""
db = Database(path, auto_migrate=auto_migrate)
store = Store(db)
return Storage(db=db, store=store, engine=ApplyEngine(store, applier, actor=actor))
def rollback_to(engine: ApplyEngine, plan_id: int, *, reason: str = "") -> StateSnapshot:
"""Явный откат состояния к снимку плана (тонкая обёртка для API)."""
return engine.rollback(plan_id, reason=reason)
-174
View File
@@ -1,174 +0,0 @@
"""Снимки состояния и восстановление после отката (rollback, issue #20).
Снимок — это JSON-копия состояния (режим, расписание GeoIP, правила,
исключения) на момент перед применением. Снимок сохраняется ДО мутации,
поэтому даже при неудачном apply (внешний iptables упал) есть источник для
восстановления.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass
from typing import Any
from . import audit
from .clock import utcnow
from .errors import StorageError
from .models import Entity
@dataclass(frozen=True, slots=True)
class StateSnapshot:
"""Сохранённый снимок состояния."""
id: int
entity: str
entity_id: str
created_at: str
payload: dict[str, Any]
reason: str
def capture_state(conn: sqlite3.Connection, *, include_exceptions: bool = True) -> dict[str, Any]:
"""Снять текущее состояние в виде JSON-совместимого словаря."""
mode_row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
geoip_row = conn.execute(
"SELECT enabled, cron, provider FROM geoip_schedule WHERE id = 1"
).fetchone()
rules = [
{
"name": str(row["name"]),
"kind": str(row["kind"]),
"match": str(row["match"]),
"target": str(row["target"]),
"priority": int(row["priority"]),
"enabled": bool(row["enabled"]),
}
for row in conn.execute(
"SELECT name, kind, match, target, priority, enabled FROM routing_rules"
" ORDER BY priority, id"
).fetchall()
]
payload: dict[str, Any] = {
"mode": "off" if mode_row is None else str(mode_row["mode"]),
"geoip": {
"enabled": True if geoip_row is None else bool(geoip_row["enabled"]),
"cron": "0 4 * * *" if geoip_row is None else str(geoip_row["cron"]),
"provider": "ripe" if geoip_row is None else str(geoip_row["provider"]),
},
"rules": rules,
}
if include_exceptions:
payload["exceptions"] = [
{
"match": str(row["match"]),
"reason": str(row["reason"]),
"enabled": bool(row["enabled"]),
}
for row in conn.execute(
"SELECT match, reason, enabled FROM rule_exceptions ORDER BY id"
).fetchall()
]
return payload
def save_snapshot(
conn: sqlite3.Connection,
payload: dict[str, Any],
*,
entity: str = Entity.APPLY.value,
entity_id: str = "state",
reason: str = "",
) -> int:
"""Сохранить снимок состояния и вернуть идентификатор."""
cursor = conn.execute(
"INSERT INTO state_snapshots (entity, entity_id, created_at, payload, reason)"
" VALUES (?, ?, ?, ?, ?)",
(entity, entity_id, utcnow(), json.dumps(payload, sort_keys=True), reason),
)
return int(cursor.lastrowid or 0)
def load_snapshot(conn: sqlite3.Connection, snapshot_id: int) -> StateSnapshot:
"""Прочитать снимок по id."""
row = conn.execute("SELECT * FROM state_snapshots WHERE id = ?", (snapshot_id,)).fetchone()
if row is None:
raise StorageError(f"снимок состояния {snapshot_id} не найден")
return StateSnapshot(
id=int(row["id"]),
entity=str(row["entity"]),
entity_id=str(row["entity_id"]),
created_at=str(row["created_at"]),
payload=json.loads(str(row["payload"])),
reason=str(row["reason"]),
)
def restore_state(conn: sqlite3.Connection, payload: dict[str, Any]) -> None:
"""Полностью заменить изменяемые таблицы состоянием из снимка."""
conn.execute("DELETE FROM routing_rules")
for rule in payload.get("rules", []):
conn.execute(
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
rule["name"],
rule["kind"],
rule["match"],
rule["target"],
int(rule["priority"]),
1 if rule["enabled"] else 0,
utcnow(),
utcnow(),
),
)
if "exceptions" in payload:
conn.execute("DELETE FROM rule_exceptions")
for exception in payload["exceptions"]:
conn.execute(
"INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)"
" VALUES (?, ?, ?, ?, ?)",
(
exception["match"],
exception.get("reason", ""),
1 if exception["enabled"] else 0,
utcnow(),
utcnow(),
),
)
geoip = payload.get("geoip", {})
conn.execute(
"UPDATE routing_mode SET mode = ?, updated_at = ? WHERE id = 1",
(payload.get("mode", "off"), utcnow()),
)
conn.execute(
"UPDATE geoip_schedule SET enabled = ?, cron = ?, provider = ?,"
" updated_at = ? WHERE id = 1",
(
1 if geoip.get("enabled", True) else 0,
geoip.get("cron", "0 4 * * *"),
geoip.get("provider", "ripe"),
utcnow(),
),
)
def rollback_to_snapshot(
conn: sqlite3.Connection, snapshot_id: int, *, actor: str, reason: str = ""
) -> StateSnapshot:
"""Восстановить состояние из снимка и записать это в журнал аудита."""
snapshot = load_snapshot(conn, snapshot_id)
audit.record(
conn,
actor=actor,
action="rollback",
entity=snapshot.entity,
entity_id=snapshot.entity_id,
old_value=capture_state(conn),
new_value=snapshot.payload,
detail=reason or f"rollback к снимку #{snapshot.id}",
)
restore_state(conn, snapshot.payload)
return snapshot
-131
View File
@@ -1,131 +0,0 @@
"""Чистые функции валидации состояния (стадия ``validate`` перед apply, issue #20).
Валидация отделена от БД: те же функции вызывают и репозиторий (точечные
мутации), и движок атомарного применения (``validate -> stage -> apply``).
"""
from __future__ import annotations
import ipaddress
import re
from .errors import ValidationError
from .models import DEVICE_POLICIES, RULE_KINDS, Mode
_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_.:-]{0,62}$")
_DOMAIN_RE = re.compile(r"^(\*\.)?([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$")
_IPSET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$")
_MAC_RE = re.compile(r"^(?:[0-9a-f]{2}:){5}[0-9a-f]{2}$")
_CRON_FIELD_RE = re.compile(r"^[0-9*,/-]{1,20}$")
_PROVIDER_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,31}$")
def normalize_mac(mac: str) -> str:
"""Привести MAC к канонической форме ``aa:bb:cc:dd:ee:ff``."""
cleaned = mac.strip().replace("-", ":").lower()
if len(cleaned) == 12 and ":" not in cleaned:
cleaned = ":".join(cleaned[i : i + 2] for i in range(0, 12, 2))
if not _MAC_RE.match(cleaned):
raise ValidationError("mac", f"некорректный MAC-адрес: {mac!r}")
return cleaned
def validate_name(field: str, value: str) -> str:
"""Проверить идентификатор (имя правила/конфига устройства)."""
if not _NAME_RE.match(value.strip()):
raise ValidationError(
field, f"допустимы строчные латинские буквы, цифры и «_.:-», до 63 символов: {value!r}"
)
return value.strip()
def validate_rule_kind(kind: str) -> str:
if kind not in RULE_KINDS:
raise ValidationError("kind", f"ожидается одно из {RULE_KINDS}, получено {kind!r}")
return kind
def validate_match(kind: str, match: str) -> str:
"""Проверить образец правила по типу записи."""
validate_rule_kind(kind)
value = match.strip()
if not value:
raise ValidationError("match", "образец правила не может быть пустым")
if kind == "domain":
if not _DOMAIN_RE.match(value):
raise ValidationError("match", f"некорректный домен: {value!r}")
elif kind == "cidr":
try:
ipaddress.ip_network(value, strict=False)
except ValueError as exc:
raise ValidationError("match", f"некорректная сеть CIDR: {value!r}") from exc
elif not _IPSET_RE.match(value):
raise ValidationError("match", f"некорректное имя ipset: {value!r}")
return value
def validate_exception_match(match: str) -> str:
"""Проверить образец исключения: домен, имя ipset или сеть CIDR."""
value = match.strip()
if not value:
raise ValidationError("match", "образец исключения не может быть пустым")
if _DOMAIN_RE.match(value) or _IPSET_RE.match(value):
return value
try:
ipaddress.ip_network(value, strict=False)
except ValueError as exc:
raise ValidationError(
"match", f"ожидается домен, сеть CIDR или имя ipset: {match!r}"
) from exc
return value
def validate_target(target: str) -> str:
value = target.strip()
if not _IPSET_RE.match(value):
raise ValidationError("target", f"некорректная цель маршрутизации: {target!r}")
return value
def validate_priority(priority: int) -> int:
if not isinstance(priority, int) or isinstance(priority, bool):
raise ValidationError("priority", f"ожидается целое число, получено {priority!r}")
if not 0 <= priority <= 65535:
raise ValidationError("priority", f"вне диапазона 0..65535: {priority}")
return priority
def validate_mode(mode: str | Mode) -> Mode:
try:
return Mode(mode)
except ValueError as exc:
raise ValidationError(
"mode", f"ожидается одно из {Mode.values()}, получено {mode!r}"
) from exc
def validate_cron(cron: str) -> str:
fields = cron.split()
if len(fields) != 5 or not all(_CRON_FIELD_RE.match(part) for part in fields):
raise ValidationError("cron", f"ожидается 5-полевое cron-выражение: {cron!r}")
return " ".join(fields)
def validate_provider(provider: str) -> str:
value = provider.strip().lower()
if not _PROVIDER_RE.match(value):
raise ValidationError("provider", f"некорректное имя провайдера GeoIP: {provider!r}")
return value
def validate_device_policy(policy: str) -> str:
if policy not in DEVICE_POLICIES:
raise ValidationError("policy", f"ожидается одно из {DEVICE_POLICIES}, получено {policy!r}")
return policy
def validate_config_path(path: str) -> str:
value = path.strip()
if not value.startswith("/") or ".." in value.split("/"):
raise ValidationError("path", f"ожидается абсолютный путь без «..»: {path!r}")
return value
-311
View File
@@ -1,311 +0,0 @@
"""Тесты атомарного применения: validate -> stage -> apply -> rollback (issue #20)."""
from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from typing import Any
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
AUDIT_STATUS_STAGED,
ApplyEngine,
ApplyError,
Database,
DesiredState,
Entity,
GeoIpSchedule,
Mode,
NullApplier,
Rule,
RuleException,
Store,
ValidationError,
capture_state,
)
class MockIptables:
"""Мок внешнего применителя: умеет «падать» как реальный iptables."""
def __init__(self, *, fail_apply: bool = False, fail_revert: bool = False) -> None:
self.applied: list[dict[str, Any]] = []
self.reverted: list[dict[str, Any]] = []
self.fail_apply = fail_apply
self.fail_revert = fail_revert
def apply(self, payload: Mapping[str, Any]) -> None:
if self.fail_apply:
self.applied.append(dict(payload))
raise RuntimeError("mock iptables: exit 4")
self.applied.append(dict(payload))
def revert(self, payload: Mapping[str, Any]) -> None:
if self.fail_revert:
raise RuntimeError("mock iptables: revert failed")
self.reverted.append(dict(payload))
def make_engine(
tmp_path: Path, applier: MockIptables | None = None
) -> tuple[Database, Store, ApplyEngine, MockIptables]:
db = Database(tmp_path / "data" / "awg-dash.db")
store = Store(db)
mock = applier if applier is not None else MockIptables()
return db, store, ApplyEngine(store, mock, actor="ui:admin"), mock
def desired(*rules: Rule, mode: Mode = Mode.SPLIT) -> DesiredState:
return DesiredState(mode=mode, rules=rules, geoip=GeoIpSchedule())
# -- validate ------------------------------------------------------------------
def test_validate_accepts_good_state(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
state = desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"))
assert engine.validate(state) is state
def test_validate_rejects_bad_state(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
cases: list[DesiredState] = [
DesiredState(mode="turbo"), # type: ignore[arg-type]
desired(Rule(name="dup", kind="domain", match="a.example.com", target="tun0"),
Rule(name="dup", kind="domain", match="b.example.com", target="tun0")),
desired(Rule(name="x", kind="cidr", match="10.0.0.0/99", target="tun0")),
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="всякий раз")),
DesiredState(
mode=Mode.SPLIT,
rules=(),
geoip=GeoIpSchedule(),
exceptions=(RuleException(match="!!!"),),
),
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(provider="!")),
]
for case in cases:
with pytest.raises(ValidationError):
engine.validate(case)
def test_validate_does_not_touch_database(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
with pytest.raises(ValidationError):
engine.apply_payload({"mode": "split", "rules": [{"name": "x", "kind": "cidr",
"match": "bad", "target": "tun0"}]})
assert store.counts() == {
"rules": 0,
"exceptions": 0,
"configs": 0,
"devices": 0,
"audit": 0,
"snapshots": 0,
}
# -- stage ---------------------------------------------------------------------
def test_stage_saves_snapshot_and_staged_audit(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"),
actor="system")
store.set_mode(Mode.OFF, actor="system")
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0")), reason="тест")
assert plan.previous == before
assert plan.desired.mode == Mode.SPLIT
# состояние ещё не изменено
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["old"]
snapshots = store.counts()["snapshots"]
assert snapshots == 1
staged = store.list_audit(entity=Entity.APPLY.value)[0]
assert staged.status == AUDIT_STATUS_STAGED
assert staged.action == "stage"
assert staged.detail == "тест"
assert staged.old_value == before
def test_stage_rejects_invalid_state_without_snapshot(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
with pytest.raises(ValidationError):
engine.stage(DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="bad")))
assert store.counts()["snapshots"] == 0
# -- apply ---------------------------------------------------------------------
def test_apply_writes_state_and_audit(tmp_path: Path) -> None:
db, store, engine, mock = make_engine(tmp_path)
with db:
plan = engine.stage(desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
mode=Mode.FULL), reason="включение")
result = engine.apply(plan)
assert result.ok is True
assert result.plan_id == plan.plan_id
assert result.audit_id > 0
assert store.get_mode() == Mode.FULL
assert [rule.name for rule in store.list_rules()] == ["yt"]
assert mock.applied == [plan.desired.to_payload()]
entry = store.list_audit(entity=Entity.APPLY.value)[0]
assert entry.status == AUDIT_STATUS_OK
assert entry.action == "apply"
assert entry.old_value == plan.previous
assert entry.new_value == plan.desired.to_payload()
def test_failed_apply_restores_previous_state(tmp_path: Path) -> None:
"""GIVEN ошибочный apply (мок падения iptables), THEN состояние прежнее + ApplyError."""
db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True))
with db:
store.upsert_rule(Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system")
store.set_mode(Mode.SPLIT, actor="system")
store.set_geoip_schedule(cron="0 5 * * *", actor="system")
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0"), mode=Mode.FULL), reason="сломается")
with pytest.raises(ApplyError) as excinfo:
engine.apply(plan)
error = excinfo.value
assert error.restored is True
assert error.audit_id is not None and error.audit_id > 0
# состояние, а не «частично применённая конфигурация»
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["keep"]
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().cron == "0 5 * * *"
assert mock.reverted == [before]
failed = store.list_audit(entity=Entity.APPLY.value)[0]
assert failed.status == AUDIT_STATUS_FAILED
assert failed.id == error.audit_id
assert failed.new_value == before
assert "mock iptables" in (failed.detail or "")
def test_failed_apply_with_failing_revert_still_restores_db(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(
tmp_path, MockIptables(fail_apply=True, fail_revert=True)
)
with db:
before = capture_state(db.connection)
plan = engine.stage(desired(Rule(name="x", kind="domain", match="x.example.com", target="tun0")))
with pytest.raises(ApplyError) as excinfo:
engine.apply(plan)
assert excinfo.value.restored is True
assert capture_state(db.connection) == before
def test_apply_can_succeed_after_failed_apply(tmp_path: Path) -> None:
db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True))
with db:
bad = engine.stage(desired(Rule(name="bad", kind="domain", match="bad.example.com",
target="tun0")))
with pytest.raises(ApplyError):
engine.apply(bad)
mock.fail_apply = False
good_rule = Rule(name="good", kind="domain", match="good.example.com", target="tun0")
result = engine.apply(engine.stage(desired(good_rule, mode=Mode.FULL)))
assert result.ok is True
assert [rule.name for rule in store.list_rules()] == ["good"]
assert store.get_mode() == Mode.FULL
# -- rollback ------------------------------------------------------------------
def test_rollback_returns_previous_state(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"),
actor="ui:admin")
before = capture_state(db.connection)
result = engine.apply(
engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com",
target="tun0"), mode=Mode.FULL), reason="применение"),
)
assert result.ok is True
snapshot = engine.rollback(result.plan_id, reason="откат оператором")
assert snapshot.id == result.plan_id
assert snapshot.payload == before
assert capture_state(db.connection) == before
assert [rule.name for rule in store.list_rules()] == ["old"]
assert store.get_mode() == Mode.OFF
rollback_entry = store.list_audit(entity=Entity.APPLY.value)[0]
assert rollback_entry.action == "rollback"
assert rollback_entry.detail == "откат оператором"
assert rollback_entry.new_value == before
def test_apply_payload_sugar_and_null_applier(tmp_path: Path) -> None:
db, store, engine, _mock = make_engine(tmp_path)
with db:
result = engine.apply_payload(
{
"mode": "split",
"rules": [{"name": "yt", "kind": "domain", "match": "youtube.com", "target": "tun0"}],
"geoip": {"enabled": False, "cron": "0 3 * * *", "provider": "ripe"},
"exceptions": [{"match": "bank.example.com", "reason": "банк"}],
},
reason="api: применить",
)
assert result.ok is True
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().enabled is False
assert store.list_exceptions()[0].match == "bank.example.com"
engine_null = ApplyEngine(store, NullApplier(), actor="system")
assert engine_null.apply_payload({"mode": "off"}).ok is True
assert store.get_mode() == Mode.OFF
def test_desired_state_payload_roundtrip() -> None:
state = DesiredState(
mode=Mode.FULL,
rules=(Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=7),),
geoip=GeoIpSchedule(enabled=False, cron="0 1 * * *", provider="maxmind"),
exceptions=(RuleException(match="bank.example.com", reason="банк", enabled=False),),
)
restored = DesiredState.from_payload(state.to_payload())
assert restored.mode == Mode.FULL
assert restored.rules == state.rules
assert restored.geoip == state.geoip
assert restored.exceptions == state.exceptions
@pytest.mark.parametrize(
"payload",
[
{"rules": "не список"},
{"rules": [{"kind": "domain", "match": "a.example.com", "target": "tun0"}]},
{"rules": ["строка"]},
{"exceptions": "не список"},
{"exceptions": [{"reason": "нет match"}]},
{"geoip": "не объект"},
{"mode": "turbo"},
],
)
def test_desired_state_from_payload_rejects_bad_payload(payload: dict[str, Any]) -> None:
with pytest.raises(ValidationError):
DesiredState.from_payload(payload)
def test_apply_engine_requires_store(tmp_path: Path) -> None:
db, _store, engine, _mock = make_engine(tmp_path)
with db:
assert engine.validate(desired()) == desired()
-145
View File
@@ -1,145 +0,0 @@
"""Тесты журнала аудита «кто/когда/что/old→new» (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
AUDIT_STATUS_OK,
Database,
Entity,
Mode,
Rule,
Store,
)
from awg_api.store import audit
def make_db(tmp_path: Path) -> Database:
return Database(tmp_path / "data" / "awg-dash.db")
def test_record_and_read_back(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
entry_id = audit.record(
conn,
actor="ui:admin",
action="update",
entity=Entity.RULE.value,
entity_id=7,
old_value={"match": "old.example.com"},
new_value={"match": "new.example.com"},
)
assert entry_id > 0
entries = audit.list_entries(db.connection)
assert len(entries) == 1
entry = entries[0]
assert entry.actor == "ui:admin"
assert entry.entity == "rule"
assert entry.entity_id == "7"
assert entry.old_value == {"match": "old.example.com"}
assert entry.new_value == {"match": "new.example.com"}
assert entry.status == AUDIT_STATUS_OK
assert entry.ts.endswith("Z")
assert entry.as_dict()["entity_id"] == "7"
def test_actor_is_required(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn, pytest.raises(ValueError, match="actor"):
audit.record(
conn,
actor="",
action="update",
entity=Entity.RULE.value,
new_value={"x": 1},
)
def test_old_and_new_can_be_absent(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(conn, actor="system", action="noop", entity=Entity.APPLY.value)
entry = audit.last_entry(db.connection)
assert entry is not None
assert entry.old_value is None
assert entry.new_value is None
assert entry.entity_id is None
assert entry.detail is None
def test_filters_limit_and_ordering(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
store = Store(db)
store.set_mode(Mode.SPLIT, actor="ui:admin")
store.set_mode(Mode.FULL, actor="ui:admin")
store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
mode_entries = audit.list_entries(db.connection, entity=Entity.MODE.value)
assert len(mode_entries) == 2
assert mode_entries[0].new_value == {"mode": "full"} # свежие первыми
assert mode_entries[0].old_value == {"mode": "split"}
by_id = audit.list_entries(
db.connection, entity=Entity.MODE.value, entity_id=1, limit=1
)
assert len(by_id) == 1
assert audit.count(db.connection) == 3
assert audit.count(db.connection, entity=Entity.RULE.value) == 1
assert audit.last_entry(db.connection, entity=Entity.RULE.value) is not None
def test_ts_can_be_overridden(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(
conn,
actor="system",
action="stage",
entity=Entity.APPLY.value,
ts="2026-01-02T03:04:05.678Z",
)
entry = audit.last_entry(db.connection)
assert entry is not None and entry.ts == "2026-01-02T03:04:05.678Z"
def test_scalar_and_list_payloads_roundtrip(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
with db.transaction() as conn:
audit.record(
conn,
actor="system",
action="apply",
entity=Entity.APPLY.value,
old_value=["a", "b"],
new_value="plain",
status=AUDIT_STATUS_FAILED,
detail="iptables: exit 4",
)
entry = audit.last_entry(db.connection)
assert entry is not None
assert entry.old_value == ["a", "b"]
assert entry.new_value == "plain"
assert entry.status == AUDIT_STATUS_FAILED
assert entry.detail == "iptables: exit 4"
def test_no_audit_row_when_transaction_rolls_back(tmp_path: Path) -> None:
"""Мутация и её аудит коммитятся вместе: откат транзакции не оставляет следа."""
with make_db(tmp_path) as db:
with pytest.raises(RuntimeError), db.transaction() as conn:
audit.record(conn, actor="system", action="update", entity=Entity.RULE.value)
raise RuntimeError("мутация не удалась")
assert audit.count(db.connection) == 0
def test_last_entry_returns_none_on_empty_log(tmp_path: Path) -> None:
with make_db(tmp_path) as db:
assert audit.last_entry(db.connection) is None
assert audit.list_entries(db.connection) == []
-308
View File
@@ -1,308 +0,0 @@
"""Тесты потокобезопасности слоя хранения (#54, D-1/D-2 из #55; issue #20).
Контракт (задокументирован в ``backend/awg_api/store/README.md``): одно
соединение SQLite, все операции — чтения и записи — сериализуются одним
``RLock``. Читатель никогда не видит незафиксированных строк и состояния,
которое затем будет откачено, а конкурентные мутации дают согласованный
журнал аудита.
"""
from __future__ import annotations
import asyncio
import threading
import time
from collections.abc import Mapping
from pathlib import Path
from typing import Any
import pytest
from awg_api.store import (
AUDIT_STATUS_FAILED,
ApplyEngine,
ApplyError,
Database,
DesiredState,
Entity,
Mode,
Rule,
Store,
capture_state,
)
THREADS = 4
PER_THREAD = 25
JOIN_TIMEOUT = 30.0
# окно наблюдения: за это время незаблокированное чтение гарантированно успело бы вернуться
OBSERVE_SECONDS = 0.2
RULE_INSERT = (
"INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
)
def open_store(tmp_path: Path) -> tuple[Database, Store]:
db = Database(tmp_path / "data" / "awg-dash.db")
return db, Store(db)
def rule(tid: int, idx: int) -> Rule:
return Rule(
name=f"r{tid}x{idx}",
kind="domain",
match=f"t{tid}r{idx}.example.com",
target="tun0",
)
def run_threads(workers: list[Any]) -> list[threading.Thread]:
threads = [threading.Thread(target=worker) for worker in workers]
for thread in threads:
thread.start()
for thread in threads:
thread.join(JOIN_TIMEOUT)
assert not thread.is_alive(), f"поток {thread.name} не завершился"
return threads
# -- #54: конкурентные мутации вместе с чтениями ---------------------------------
def test_concurrent_upserts_and_reads_are_consistent(tmp_path: Path) -> None:
"""N потоков × M upsert с чтениями: без ошибок, rules == audit == N*M."""
db, store = open_store(tmp_path)
with db:
errors: list[str] = []
barrier = threading.Barrier(THREADS)
def worker(tid: int) -> None:
try:
barrier.wait(timeout=JOIN_TIMEOUT)
for idx in range(PER_THREAD):
store.upsert_rule(rule(tid, idx), actor=f"thread{tid}")
store.list_rules()
store.get_mode()
store.get_rule_by_name(f"r{tid}x{idx}")
except BaseException as exc: # noqa: BLE001 - фиксируем любое падение потока
errors.append(f"thread{tid}: {type(exc).__name__}: {exc}")
run_threads([lambda tid=tid: worker(tid) for tid in range(THREADS)])
total = THREADS * PER_THREAD
assert errors == []
assert store.counts()["rules"] == total
assert store.counts()["audit"] == total
assert len(store.list_rules()) == total
assert len(store.list_audit(limit=total * 2)) == total
# -- #54: чтение не видит незафиксированных строк -------------------------------
def test_reads_never_see_uncommitted_rows(tmp_path: Path) -> None:
"""Пока транзакция другого потока открыта, читатель ждёт коммита, а не читает грязь."""
db, store = open_store(tmp_path)
with db:
store.upsert_rule(
Rule(name="baseline", kind="domain", match="baseline.example.com", target="tun0"),
actor="system",
)
transaction_open = threading.Event()
reader_started = threading.Event()
release = threading.Event()
observed: list[list[str]] = []
writer_errors: list[str] = []
def writer() -> None:
try:
with db.transaction() as conn:
conn.execute(
RULE_INSERT,
(
"uncommitted",
"domain",
"uncommitted.example.com",
"tun0",
100,
1,
"2026-01-01T00:00:00Z",
"2026-01-01T00:00:00Z",
),
)
transaction_open.set()
release.wait(JOIN_TIMEOUT)
except BaseException as exc: # noqa: BLE001 - фиксируем падение писателя
writer_errors.append(f"{type(exc).__name__}: {exc}")
def reader() -> None:
reader_started.set()
observed.append([item.name for item in store.list_rules()])
writer_thread = threading.Thread(target=writer)
writer_thread.start()
assert transaction_open.wait(JOIN_TIMEOUT)
reader_thread = threading.Thread(target=reader)
reader_thread.start()
assert reader_started.wait(JOIN_TIMEOUT)
time.sleep(OBSERVE_SECONDS)
# незафиксированная строка наружу не видна: чтение ждёт завершения транзакции
assert observed == []
release.set()
writer_thread.join(JOIN_TIMEOUT)
reader_thread.join(JOIN_TIMEOUT)
assert writer_errors == []
assert observed == [["baseline", "uncommitted"]]
# -- #54 п.4: во время падающего apply наружу не отдаётся откаченное состояние ----
class BlockingApplier:
"""Применитель, который «висит» внутри apply и затем падает как iptables."""
def __init__(self) -> None:
self.entered = threading.Event()
self.release = threading.Event()
self.reverted: list[dict[str, Any]] = []
def apply(self, payload: Mapping[str, Any]) -> None:
self.entered.set()
assert self.release.wait(JOIN_TIMEOUT)
raise RuntimeError("mock iptables: exit 4")
def revert(self, payload: Mapping[str, Any]) -> None:
self.reverted.append(dict(payload))
def test_read_during_failing_apply_never_sees_discarded_state(tmp_path: Path) -> None:
"""AC #20: снаружи не видно «полуприменённой» конфигурации, которую затем откатят."""
db, store = open_store(tmp_path)
applier = BlockingApplier()
engine = ApplyEngine(store, applier, actor="ui:admin")
with db:
store.upsert_rule(
Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system",
)
before = capture_state(db.connection)
plan = engine.stage(
DesiredState(
mode=Mode.FULL,
rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),),
),
reason="сломается",
)
errors: list[BaseException] = []
observed: list[list[str]] = []
reader_started = threading.Event()
def apply_worker() -> None:
try:
engine.apply(plan)
except BaseException as exc: # noqa: BLE001 - ожидаем ApplyError
errors.append(exc)
def reader() -> None:
reader_started.set()
observed.append([item.name for item in store.list_rules()])
apply_thread = threading.Thread(target=apply_worker)
apply_thread.start()
assert applier.entered.wait(JOIN_TIMEOUT)
reader_thread = threading.Thread(target=reader)
reader_thread.start()
assert reader_started.wait(JOIN_TIMEOUT)
time.sleep(OBSERVE_SECONDS)
assert observed == [], "чтение вернуло состояние, которого ещё нет в БД"
applier.release.set()
apply_thread.join(JOIN_TIMEOUT)
reader_thread.join(JOIN_TIMEOUT)
assert len(errors) == 1
error = errors[0]
assert isinstance(error, ApplyError)
assert error.restored is True
assert observed == [["keep"]]
assert [item.name for item in store.list_rules()] == ["keep"]
assert capture_state(db.connection) == before
assert applier.reverted == [before]
# -- #55 D-1: BaseException не должен оставлять staged без исхода ---------------
class InterruptingApplier:
"""Применитель, прерывающий apply исключением из ``BaseException``."""
def __init__(self, exc_type: type[BaseException]) -> None:
self.exc_type = exc_type
self.applied = 0
self.reverted: list[dict[str, Any]] = []
def apply(self, payload: Mapping[str, Any]) -> None:
self.applied += 1
raise self.exc_type("внешний применитель прерван")
def revert(self, payload: Mapping[str, Any]) -> None:
self.reverted.append(dict(payload))
@pytest.mark.parametrize("exc_type", [KeyboardInterrupt, SystemExit, asyncio.CancelledError])
def test_base_exception_from_applier_reverts_and_records_failure(
tmp_path: Path, exc_type: type[BaseException]
) -> None:
"""Отмена/прерывание apply → revert внешней подсистемы + failed в audit_log."""
db, store = open_store(tmp_path)
applier = InterruptingApplier(exc_type)
engine = ApplyEngine(store, applier, actor="ui:admin")
with db:
store.upsert_rule(
Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"),
actor="system",
)
before = capture_state(db.connection)
plan = engine.stage(
DesiredState(
mode=Mode.FULL,
rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),),
),
reason="отмена",
)
with pytest.raises(exc_type):
engine.apply(plan)
assert applier.applied == 1
assert applier.reverted == [before]
assert capture_state(db.connection) == before
assert [item.name for item in store.list_rules()] == ["keep"]
entries = store.list_audit(entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=10)
assert entries[0].status == AUDIT_STATUS_FAILED
assert exc_type.__name__ in (entries[0].detail or "")
assert entries[0].new_value == before
# -- #55 D-2: конкурентные set_geoip_schedule дают связный журнал ---------------
def test_concurrent_geoip_updates_keep_audit_chain_consistent(tmp_path: Path) -> None:
"""Каждое изменение пишет корректный old_value: цепочка old→new не рвётся."""
db, store = open_store(tmp_path)
with db:
initial = store.get_geoip_schedule()
crons = ("0 1 * * *", "0 2 * * *")
errors: list[str] = []
barrier = threading.Barrier(len(crons))
def worker(cron: str) -> None:
try:
barrier.wait(timeout=JOIN_TIMEOUT)
store.set_geoip_schedule(cron=cron, actor=f"ui:{cron}")
except BaseException as exc: # noqa: BLE001 - фиксируем падение потока
errors.append(f"{cron}: {type(exc).__name__}: {exc}")
run_threads([lambda cron=cron: worker(cron) for cron in crons])
assert errors == []
entries = list(reversed(store.list_audit(entity=Entity.GEOIP_SCHEDULE.value, limit=10)))
assert len(entries) == len(crons)
first, second = entries
assert first.old_value["cron"] == initial.cron
assert first.new_value["cron"] == second.old_value["cron"]
assert second.new_value["cron"] == store.get_geoip_schedule().cron
assert {first.new_value["cron"], second.new_value["cron"]} == set(crons)
-153
View File
@@ -1,153 +0,0 @@
"""Тесты схемы, миграций и проверки версии при старте (issue #20)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from awg_api.store import (
EXPECTED_TABLES,
LATEST_VERSION,
MIGRATIONS,
Database,
Migration,
SchemaError,
StorageError,
current_version,
migrate,
resolve_db_path,
verify_schema,
)
from awg_api.store.migrations import applied_migrations, table_columns, table_names
def test_empty_data_dir_creates_schema(tmp_path: Path) -> None:
"""GIVEN пустой /data, WHEN старт, THEN схема v1 создана автоматически."""
db_path = tmp_path / "data" / "awg-dash.db"
assert not db_path.parent.exists()
with Database(db_path) as db:
assert db.schema_version == LATEST_VERSION == 1
assert db.migration_report is not None
assert db.migration_report.applied == (1,)
names = table_names(db.connection)
assert set(EXPECTED_TABLES) <= names
for table, columns in EXPECTED_TABLES.items():
assert columns <= table_columns(db.connection, table), table
def test_wal_mode_is_enabled(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
assert db.journal_mode.lower() == "wal"
assert db.is_wal is True
def test_seed_rows_present(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
row = db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
assert row is not None and row["mode"] == "off"
geoip = db.connection.execute("SELECT cron, provider FROM geoip_schedule WHERE id = 1").fetchone()
assert geoip is not None
assert geoip["provider"] == "ripe"
def test_migrate_is_idempotent(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
report = db.migrate()
assert report.applied == ()
assert report.changed is False
assert report.from_version == report.to_version == LATEST_VERSION
def test_applied_migrations_map(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
assert applied_migrations(db.connection) == {1: "initial-state-schema"}
def test_auto_migrate_false_verifies_existing_schema(tmp_path: Path) -> None:
path = tmp_path / "awg-dash.db"
with Database(path):
pass
with Database(path, auto_migrate=False) as db:
assert db.migration_report is None
assert current_version(db.connection) == LATEST_VERSION
def test_auto_migrate_false_on_unmigrated_db_raises(tmp_path: Path) -> None:
conn = sqlite3.connect(str(tmp_path / "empty.db"))
try:
with pytest.raises(SchemaError, match="не мигрирована"):
verify_schema(conn)
finally:
conn.close()
def test_verify_schema_detects_missing_table(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute("DROP TABLE routing_rules")
with pytest.raises(SchemaError, match="routing_rules"):
db.verify_schema()
def test_verify_schema_detects_missing_column(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute("ALTER TABLE devices DROP COLUMN address")
with pytest.raises(SchemaError, match="address"):
db.verify_schema()
def test_verify_schema_rejects_newer_schema(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute(
"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
(99, "from-the-future", "2099-01-01T00:00:00.000Z"),
)
with pytest.raises(SchemaError, match="новее"):
db.verify_schema()
with pytest.raises(SchemaError, match="новее"):
db.migrate()
def test_failed_migration_leaves_schema_untouched(tmp_path: Path) -> None:
broken = MIGRATIONS + (
Migration(version=2, name="broken", statements=("CREATE TABLE t2 (id INTEGER)", "NOT SQL AT ALL")),
)
with Database(tmp_path / "awg-dash.db") as db:
with pytest.raises(StorageError, match="миграция v2"):
migrate(db.connection, migrations=broken)
assert current_version(db.connection) == 1
assert "t2" not in table_names(db.connection)
def test_migrate_rejects_downgrade(tmp_path: Path) -> None:
with Database(tmp_path / "awg-dash.db") as db:
db.connection.execute(
"INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)",
(2, "newer", "2099-01-01T00:00:00.000Z"),
)
with pytest.raises(SchemaError, match="новее"):
migrate(db.connection, migrations=MIGRATIONS)
def test_resolve_db_path_defaults_and_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("AWG_DASH_DB", raising=False)
assert resolve_db_path() == Path("/data/awg-dash.db")
monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "custom.db"))
assert resolve_db_path() == tmp_path / "custom.db"
assert resolve_db_path(tmp_path / "explicit.db") == tmp_path / "explicit.db"
def test_database_uses_env_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "nested" / "env.db"))
with Database() as db:
assert db.path == tmp_path / "nested" / "env.db"
assert db.path.exists()
def test_unwritable_path_raises_storage_error(tmp_path: Path) -> None:
blocker = tmp_path / "blocked"
blocker.write_text("not a directory")
with pytest.raises(StorageError, match="каталог"):
Database(blocker / "awg-dash.db")
-184
View File
@@ -1,184 +0,0 @@
"""Персистентность: рестарт «контейнера» сохраняет правила, конфиг и аудит (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
ConfigMeta,
Database,
DesiredState,
Device,
GeoIpSchedule,
Mode,
Rule,
StorageError,
Store,
ValidationError,
open_storage,
rollback_to,
)
from awg_api.store import validation
def test_state_survives_container_restart(tmp_path: Path) -> None:
"""GIVEN рестарт контейнера, WHEN он завершён, THEN правила и конфиг на месте."""
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as first:
store = Store(first)
store.set_mode(Mode.SPLIT, actor="ui:admin")
store.set_geoip_schedule(cron="0 3 * * *", actor="ui:admin")
store.upsert_rule(
Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10),
actor="ui:admin",
)
store.upsert_rule(
Rule(name="lan", kind="cidr", match="10.0.0.0/8", target="direct", priority=200),
actor="ui:admin",
)
store.upsert_config(
ConfigMeta(name="peer1", path="/config/peer1.conf", checksum="deadbeef", size_bytes=1024),
actor="ui:admin",
)
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="notebook"), actor="ui:admin")
audit_before = store.counts()["audit"]
assert audit_before == 6
# «рестарт контейнера»: новое соединение к тому же /data
with Database(db_path) as second:
store = Store(second)
assert second.is_wal is True
assert store.get_mode() == Mode.SPLIT
assert store.get_geoip_schedule().cron == "0 3 * * *"
assert [(rule.name, rule.target) for rule in store.list_rules()] == [
("yt", "tun0"),
("lan", "direct"),
]
config = store.get_config("peer1")
assert config is not None
assert config.checksum == "deadbeef"
assert store.get_device("aa:bb:cc:dd:ee:ff") is not None
assert store.counts()["audit"] == audit_before
assert second.migration_report is not None # миграции уже применены, повтор не нужен
assert second.migration_report.applied == ()
def test_wal_file_next_to_database(tmp_path: Path) -> None:
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as db:
Store(db).set_mode(Mode.FULL, actor="system")
assert db_path.with_name("awg-dash.db-wal").exists()
assert db_path.with_name("awg-dash.db-shm").exists()
def test_open_storage_builds_full_stack(tmp_path: Path) -> None:
with open_storage(tmp_path / "data" / "awg-dash.db", actor="api:token") as storage:
storage.store.set_mode(Mode.SPLIT, actor="api:token")
plan = storage.engine.stage(
DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule()),
reason="через storage",
)
result = storage.engine.apply(plan)
assert result.ok is True
assert storage.store.counts()["snapshots"] == 1
assert storage.db.schema_version == 1
def test_rollback_helper_and_missing_snapshot(tmp_path: Path) -> None:
with open_storage(tmp_path / "data" / "awg-dash.db") as storage:
storage.store.set_mode(Mode.OFF, actor="system")
plan = storage.engine.stage(DesiredState(mode=Mode.FULL))
storage.engine.apply(plan)
assert storage.store.get_mode() == Mode.FULL
snapshot = rollback_to(storage.engine, plan.plan_id, reason="ручной откат")
assert snapshot.payload["mode"] == "off"
assert storage.store.get_mode() == Mode.OFF
with pytest.raises(StorageError, match="не найден"):
storage.engine.rollback(999999)
def test_transactions_are_isolated_per_connection(tmp_path: Path) -> None:
db_path = tmp_path / "data" / "awg-dash.db"
with Database(db_path) as writer, Database(db_path) as reader:
Store(writer).set_mode(Mode.SPLIT, actor="system")
assert Store(reader).get_mode() == Mode.SPLIT
with pytest.raises(RuntimeError, match="откат"), writer.transaction() as conn:
conn.execute("UPDATE routing_mode SET mode = 'full' WHERE id = 1")
raise RuntimeError("откат")
assert Store(reader).get_mode() == Mode.SPLIT
# -- валидация (чистые функции) -------------------------------------------------
@pytest.mark.parametrize(
("raw", "expected"),
[
("AA:BB:CC:DD:EE:FF", "aa:bb:cc:dd:ee:ff"),
("aa-bb-cc-dd-ee-ff", "aa:bb:cc:dd:ee:ff"),
("aabbccddeeff", "aa:bb:cc:dd:ee:ff"),
],
)
def test_normalize_mac(raw: str, expected: str) -> None:
assert validation.normalize_mac(raw) == expected
def test_normalize_mac_rejects_garbage() -> None:
with pytest.raises(ValidationError, match="MAC"):
validation.normalize_mac("not-a-mac")
@pytest.mark.parametrize("value", ["ok", "peer_1", "rule-2", "a.b:c"])
def test_validate_name_accepts(value: str) -> None:
assert validation.validate_name("name", f" {value} ") == value
@pytest.mark.parametrize("value", ["", "UPPER", "-lead", "with space", "юникод", "x" * 64])
def test_validate_name_rejects(value: str) -> None:
with pytest.raises(ValidationError):
validation.validate_name("name", value)
@pytest.mark.parametrize("value", ["/config/peer1.conf", "/etc/awg-dash/rules.conf"])
def test_validate_config_path_accepts(value: str) -> None:
assert validation.validate_config_path(value) == value
@pytest.mark.parametrize("value", ["relative.conf", "/config/../../etc/shadow", ""])
def test_validate_config_path_rejects(value: str) -> None:
with pytest.raises(ValidationError):
validation.validate_config_path(value)
@pytest.mark.parametrize(("value", "expected"), [("RIPE", "ripe"), ("maxmind", "maxmind")])
def test_validate_provider(value: str, expected: str) -> None:
assert validation.validate_provider(value) == expected
@pytest.mark.parametrize("value", ["tunnel", "direct", "block", "default"])
def test_validate_device_policy_accepts(value: str) -> None:
assert validation.validate_device_policy(value) == value
def test_validate_device_policy_rejects() -> None:
with pytest.raises(ValidationError, match="policy"):
validation.validate_device_policy("root")
@pytest.mark.parametrize("value", ["*.example.com", "youtube.com", "a.b.c.example.co.uk"])
def test_validate_match_domain(value: str) -> None:
assert validation.validate_match("domain", value) == value
@pytest.mark.parametrize("value", ["10.0.0.0/8", "2001:db8::/32", "192.168.1.7"])
def test_validate_match_cidr(value: str) -> None:
assert validation.validate_match("cidr", value) == value
def test_validate_match_ipset_and_empty() -> None:
assert validation.validate_match("ipset", "awg_ru") == "awg_ru"
with pytest.raises(ValidationError, match="пустым"):
validation.validate_match("domain", " ")
-312
View File
@@ -1,312 +0,0 @@
"""Тесты репозитория: CRUD-мутации с аудитом «old→new» (issue #20)."""
from __future__ import annotations
from pathlib import Path
import pytest
from awg_api.store import (
ConfigMeta,
Database,
Device,
Entity,
GeoIpSchedule,
Mode,
Rule,
RuleException,
StorageError,
Store,
ValidationError,
)
from awg_api.store import audit
from awg_api.store.validation import validate_priority
def make_store(tmp_path: Path) -> tuple[Database, Store]:
db = Database(tmp_path / "data" / "awg-dash.db")
return db, Store(db)
# -- правила -----------------------------------------------------------------
def test_upsert_rule_creates_with_audit(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(
Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10),
actor="ui:admin",
)
assert saved.id is not None and saved.created_at == saved.updated_at
assert store.get_rule(saved.id) == saved
assert store.get_rule_by_name("yt") == saved
entries = store.list_audit(entity=Entity.RULE.value)
assert len(entries) == 1
assert entries[0].action == "create"
assert entries[0].old_value is None
assert entries[0].new_value is not None
assert entries[0].new_value["name"] == "yt"
def test_upsert_rule_updates_by_name_and_records_old_new(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
first = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
second = store.upsert_rule(Rule(name="yt", kind="domain", match="youtu.be", target="tun1"),
actor="ui:admin")
assert second.id == first.id
assert second.match == "youtu.be" and second.target == "tun1"
assert store.counts()["rules"] == 1
entries = store.list_audit(entity=Entity.RULE.value)
assert [entry.action for entry in entries] == ["update", "create"]
assert entries[0].old_value["match"] == "youtube.com"
assert entries[0].new_value["match"] == "youtu.be"
assert entries[0].old_value["target"] == "tun0"
assert entries[0].new_value["target"] == "tun1"
def test_upsert_rule_by_id(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(Rule(name="a", kind="cidr", match="10.0.0.0/8", target="tun0"),
actor="system")
updated = store.upsert_rule(
Rule(id=saved.id, name="a", kind="cidr", match="10.1.0.0/16", target="tun0", enabled=False),
actor="system",
)
assert updated.id == saved.id
assert updated.match == "10.1.0.0/16"
assert updated.enabled is False
assert store.list_rules(enabled_only=True) == []
def test_rule_validation_rejects_bad_input(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
cases = [
Rule(name="Bad Name", kind="domain", match="youtube.com", target="tun0"),
Rule(name="ok", kind="wrong", match="youtube.com", target="tun0"),
Rule(name="ok2", kind="domain", match="not a domain", target="tun0"),
Rule(name="ok3", kind="cidr", match="10.0.0.0/99", target="tun0"),
Rule(name="ok4", kind="domain", match="youtube.com", target="bad target!"),
Rule(name="ok5", kind="domain", match="youtube.com", target="tun0", priority=-1),
]
for rule in cases:
with pytest.raises(ValidationError):
store.upsert_rule(rule, actor="system")
assert store.counts()["rules"] == 0
assert audit.count(db.connection) == 0
def test_priority_must_be_int() -> None:
with pytest.raises(ValidationError, match="целое"):
validate_priority(True) # bool не является допустимым приоритетом
with pytest.raises(ValidationError, match="диапазона"):
validate_priority(70000)
def test_delete_rule(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
saved = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"),
actor="ui:admin")
assert saved.id is not None
assert store.delete_rule(saved.id, actor="ui:admin") is True
assert store.delete_rule(saved.id, actor="ui:admin") is False
assert store.get_rule(saved.id) is None
entries = store.list_audit(entity=Entity.RULE.value)
assert [entry.action for entry in entries] == ["delete", "create"]
assert entries[0].new_value is None
assert entries[0].old_value["name"] == "yt"
def test_rules_sorted_by_priority(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
store.upsert_rule(Rule(name="b", kind="domain", match="b.example.com", target="tun0",
priority=200), actor="system")
store.upsert_rule(Rule(name="a", kind="domain", match="a.example.com", target="tun0",
priority=5), actor="system")
assert [rule.name for rule in store.list_rules()] == ["a", "b"]
# -- исключения ---------------------------------------------------------------
def test_exception_upsert_update_delete(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.upsert_exception(RuleException(match="bank.example.com", reason="банк"),
actor="ui:admin")
assert created.id is not None
assert store.get_exception("bank.example.com") == created
updated = store.upsert_exception(RuleException(match="bank.example.com", reason="банк2"),
actor="ui:admin")
assert updated.id == created.id and updated.reason == "банк2"
assert store.counts()["exceptions"] == 1
assert store.list_exceptions() == [updated]
assert store.delete_exception(updated.id or 0, actor="ui:admin") is True
assert store.delete_exception(updated.id or 0, actor="ui:admin") is False
actions = [entry.action for entry in store.list_audit(entity=Entity.EXCEPTION.value)]
assert actions == ["delete", "update", "create"]
update_entry = store.list_audit(entity=Entity.EXCEPTION.value)[1]
assert update_entry.old_value["reason"] == "банк"
assert update_entry.new_value["reason"] == "банк2"
def test_exception_accepts_cidr_and_rejects_garbage(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
assert store.upsert_exception(RuleException(match="192.168.1.0/24"), actor="system").match == (
"192.168.1.0/24"
)
assert store.upsert_exception(RuleException(match="10.0.0.1"), actor="system").match == "10.0.0.1"
with pytest.raises(ValidationError, match="домен"):
store.upsert_exception(RuleException(match="!!!"), actor="system")
with pytest.raises(ValidationError, match="пустым"):
store.upsert_exception(RuleException(match=" "), actor="system")
# -- режим и GeoIP -------------------------------------------------------------
def test_set_mode_defaults_off_and_audits(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
assert store.get_mode() == Mode.OFF
assert store.set_mode("split", actor="ui:admin") == Mode.SPLIT
assert store.get_mode() == Mode.SPLIT
assert store.set_mode(Mode("full"), actor="ui:admin") == Mode.FULL
entries = store.list_audit(entity=Entity.MODE.value)
assert len(entries) == 2
assert entries[0].old_value == {"mode": "split"}
assert entries[0].new_value == {"mode": "full"}
assert entries[0].actor == "ui:admin"
def test_set_mode_rejects_unknown(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="mode"):
store.set_mode("turbo", actor="ui:admin")
assert store.get_mode() == Mode.OFF
def test_geoip_schedule_defaults_and_partial_update(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
default = store.get_geoip_schedule()
assert default == GeoIpSchedule(enabled=True, cron="0 4 * * *", provider="ripe",
updated_at=default.updated_at)
updated = store.set_geoip_schedule(cron="*/30 * * * *", actor="ui:admin")
assert updated.cron == "*/30 * * * *"
assert updated.provider == "ripe" and updated.enabled is True
disabled = store.set_geoip_schedule(enabled=False, provider="MAXMIND", actor="ui:admin")
assert disabled.enabled is False and disabled.provider == "maxmind"
assert store.get_geoip_schedule() == disabled
with pytest.raises(ValidationError, match="cron"):
store.set_geoip_schedule(cron="ежечасно", actor="ui:admin")
with pytest.raises(ValidationError, match="provider"):
store.set_geoip_schedule(provider="!", actor="ui:admin")
entries = store.list_audit(entity=Entity.GEOIP_SCHEDULE.value)
assert [entry.action for entry in entries] == ["set_geoip_schedule"] * 2
assert entries[0].old_value["provider"] == "ripe"
assert entries[0].new_value["provider"] == "maxmind"
# -- конфиги ------------------------------------------------------------------
def test_config_upsert_and_mark_applied(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.upsert_config(
ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=1, checksum="abc",
size_bytes=512), actor="ui:admin",
)
assert created.id is not None
assert store.get_config("client-peer1") == created
updated = store.upsert_config(
ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=2, checksum="def",
size_bytes=640, applied=True), actor="ui:admin",
)
assert updated.version == 2 and updated.applied is True
assert store.counts()["configs"] == 1
marked = store.mark_config_applied("client-peer1", applied=False, actor="ui:admin")
assert marked.applied is False
entries = store.list_audit(entity=Entity.CONFIG.value)
assert [entry.action for entry in entries] == ["set_applied", "update", "create"]
assert entries[0].old_value["applied"] is True
assert entries[0].new_value["applied"] is False
assert entries[1].old_value["version"] == 1
assert entries[1].new_value["version"] == 2
def test_config_validation_and_missing(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="path"):
store.upsert_config(ConfigMeta(name="x", path="relative.conf"), actor="system")
with pytest.raises(ValidationError, match="path"):
store.upsert_config(ConfigMeta(name="x", path="/config/../etc/passwd"), actor="system")
with pytest.raises(ValidationError, match="name"):
store.upsert_config(ConfigMeta(name="Bad Name", path="/config/x.conf"), actor="system")
with pytest.raises(StorageError, match="не найден"):
store.mark_config_applied("ghost", applied=True, actor="system")
# -- устройства -----------------------------------------------------------------
def test_device_register_normalizes_mac_and_audits(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
created = store.register_device(
Device(mac="AA-BB-CC-DD-EE-FF", name="notebook", address="192.168.1.50"),
actor="ui:admin",
)
assert created.mac == "aa:bb:cc:dd:ee:ff"
assert store.get_device("AA:BB:CC:DD:EE:FF") == created
updated = store.register_device(
Device(mac="aabbccddeeff", name="notebook", policy="direct", last_seen="2026-01-01T00:00:00.000Z"),
actor="ui:admin",
)
assert updated.id == created.id and updated.policy == "direct"
assert store.counts()["devices"] == 1
assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is True
assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is False
assert store.list_devices() == []
actions = [entry.action for entry in store.list_audit(entity=Entity.DEVICE.value)]
assert actions == ["delete", "update", "create"]
update_entry = store.list_audit(entity=Entity.DEVICE.value)[1]
assert update_entry.old_value["policy"] == "default"
assert update_entry.new_value["policy"] == "direct"
def test_device_validation(tmp_path: Path) -> None:
db, store = make_store(tmp_path)
with db:
with pytest.raises(ValidationError, match="MAC"):
store.register_device(Device(mac="zz:zz", name="x"), actor="system")
with pytest.raises(ValidationError, match="policy"):
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="x", policy="кибератака"),
actor="system")
with pytest.raises(ValidationError, match="name"):
store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="Not A Name"), actor="system")
def test_audit_actor_recorded_per_mutation(tmp_path: Path) -> None:
"""GIVEN изменение через сервисный слой, THEN в audit_log есть actor и old/new."""
db, store = make_store(tmp_path)
with db:
store.set_mode(Mode.SPLIT, actor="api:token-42")
entry = store.list_audit(entity=Entity.MODE.value)[0]
assert entry.actor == "api:token-42"
assert entry.ts.endswith("Z")
assert entry.old_value == {"mode": "off"}
assert entry.new_value == {"mode": "split"}
assert entry.status == "ok"
+45
View File
@@ -0,0 +1,45 @@
# TechLead agent payload for Hermes import
id: techlead
persona: "TechLead‑bot проекта VPN Gateway"
responsibilities:
- Руководство архитектурой и принятие релизов
- Декомпозиция эпиков, утверждение DoD
- Назначение владельцев и приоритизация задач
- Эскалация blocker'ов и контроль качества
triggers:
- epic_created
- release_request
- blocker_over_48h
inputs:
- epic_description
- backlog_status
- ci_status
- test_reports
outputs:
- decomposed_issues
- release_checklist
- comments_on_issues_prs
allowed_actions:
- create_issue
- update_issue
- add_comment
- set_labels
- assign_users
decision_rules:
merge_requirements: "CI == green AND coverage >= 70% AND security_scan == low/none AND >=1 code review"
hotfix: "manual TechLead approval required (tag: hotfix-bypass)"
deploy_prod: "only after TechLead approval + smoke tests green"
escalation:
delay_hours: 48
action: "create issue 'escalation', mention owners, propose meeting within 24h"
required_permissions:
- repo: issues write
- repo: pull_request comment
- ci: read_artifacts
observability:
- action_log: /var/log/hermes/techlead_actions.log
- metrics: [epic_to_release_days, dod_compliance_rate, escalations_count]
config:
COVERAGE_THRESHOLD: 70
ESCALATION_DELAY_HOURS: 48
TECHLEAD_LOG: /var/log/hermes/techlead_actions.log
+26
View File
@@ -0,0 +1,26 @@
#!/usr/bin/env bash
# Start TechLead Hermes agent in tmux session
set -euo pipefail
TECHLEAD_LOG=${TECHLEAD_LOG:-/var/log/hermes/techlead_actions.log}
COVERAGE_THRESHOLD=${COVERAGE_THRESHOLD:-70}
ESCALATION_DELAY_HOURS=${ESCALATION_DELAY_HOURS:-48}
TMUX_SESSION=${TMUX_SESSION:-techlead}
PROFILE=${HERMES_PROFILE:-default}
mkdir -p "$(dirname "$TECHLEAD_LOG")"
# Start tmux session with hermes interactive chat
if ! tmux has-session -t "$TMUX_SESSION" 2>/dev/null; then
tmux new-session -d -s "$TMUX_SESSION" "hermes --profile $PROFILE chat"
sleep 2
# Send initial system instruction
tmux send-keys -t "$TMUX_SESSION" "You are TechLead-bot. Load persona from /workspace/exports/techlead_payload.yaml and start monitoring issues with label 'epic'." C-m
tmux send-keys -t "$TMUX_SESSION" "COVERAGE_THRESHOLD=$COVERAGE_THRESHOLD ESCALATION_DELAY_HOURS=$ESCALATION_DELAY_HOURS TECHLEAD_LOG=$TECHLEAD_LOG" C-m
echo "Started tmux session '$TMUX_SESSION' with TechLead agent."
else
echo "Tmux session '$TMUX_SESSION' already exists. Attach with: tmux attach -t $TMUX_SESSION"
fi
# One-shot alternative (no tmux):
# hermes --profile $PROFILE chat -q "You are TechLead-bot. Load persona from /workspace/exports/techlead_payload.yaml and decompose epic: ..."