139 lines
No EOL
6.7 KiB
Python
139 lines
No EOL
6.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""t-260930083716: криптографическая подпись событий и команд (hermetic).
|
||
|
||
Запуск: ./venv/bin/python tests/test_audit.py
|
||
|
||
Проверяет: ключ генерируется один раз (0600, key_id стабилен), события
|
||
подписываются при записи (sig/key_id/prev_sig), тампер данных события
|
||
ловится аудитом (first_bad), тампер НОВОГО (не спроецированного) события
|
||
останавливает проекции (RuntimeError), genesis-маркер ретро-строк,
|
||
unsigned-послегенезис строки = legacy (не падают), сцепка prev_sig
|
||
включена, конверты команд подпись/проверка/тампер, CLI verify/audit-key.
|
||
"""
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
|
||
# ── изоляция: своя БД в /tmp, свой ключ в /tmp ──
|
||
_DST = "/tmp/simbiot-test-audit.db"
|
||
for p in (_DST, _DST + "-wal", _DST + "-shm"):
|
||
if os.path.exists(p):
|
||
os.unlink(p)
|
||
os.environ["SIMBIOT_DB"] = _DST
|
||
os.environ["SIMBIOT_AUDIT_KEY_DIR"] = "/tmp/simbiot-test-audit-keys"
|
||
sys.path.insert(0, "/root/simbiot-harness")
|
||
|
||
from harness import audit, es # noqa: E402
|
||
|
||
fails = []
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
print(("PASS" if cond else "FAIL"), name, ("| " + extra) if not cond else "")
|
||
if not cond:
|
||
fails.append(name)
|
||
|
||
|
||
# ── 1. ключи ──
|
||
info = audit.key_info()
|
||
check("ключ создан", os.path.exists(audit.PRIV_PATH))
|
||
check("приватный ключ 0600", info["priv_mode"] == 0o600, str(info["priv_mode"]))
|
||
kid = audit.key_id()
|
||
check("key_id 16 hex", len(kid) == 16 and all(c in "0123456789abcdef" for c in kid))
|
||
|
||
# ── 2. подпись при записи ──
|
||
es.init_db()
|
||
es.append_event("task", "a-1", "TaskSubmitted", {"title": "тест"})
|
||
es.append_event("task", "a-1", "TaskClaimed", {"level": "day"})
|
||
es.append_event("task", "a-1", "TaskCompleted", {"answer": "годно"})
|
||
db = sqlite3.connect(_DST)
|
||
db.row_factory = sqlite3.Row
|
||
rows = [dict(r) for r in db.execute("SELECT * FROM events ORDER BY id")]
|
||
db.close()
|
||
check("3 события", len(rows) == 3)
|
||
check("все подписаны при записи",
|
||
all(r["sig"] and not r["sig"].startswith("genesis:") for r in rows))
|
||
check("key_id совпадает", all(r["key_id"] == kid for r in rows))
|
||
check("prev_sig = sig предыдущего",
|
||
all(rows[i]["prev_sig"] == rows[i - 1]["sig"] for i in (1, 2)))
|
||
|
||
# ── 3. аудит чистого стора ──
|
||
rep = audit.verify_events()
|
||
check("verify: чисто", rep["ok"] and rep["checked"] == 3 and rep["total"] == 3, str(rep))
|
||
|
||
# ── 4. тампер уже спроецированного события: ловит ПОЛНЫЙ аудит ──
|
||
es.run_projections()
|
||
db = sqlite3.connect(_DST)
|
||
db.execute("UPDATE events SET data='{\"title\":\"TAMPERED\"}' WHERE id=2")
|
||
db.commit(); db.close()
|
||
rep = audit.verify_events()
|
||
check("тампер ловится аудитом", not rep["ok"] and rep["first_bad"] == 2, str(rep))
|
||
|
||
# ── 5. тампер хвоста останавливает проекции ──
|
||
os.unlink(_DST) # свежий стор
|
||
for p in (_DST + "-wal", _DST + "-shm"):
|
||
if os.path.exists(p):
|
||
os.unlink(p)
|
||
es._signature_schema_ready = False # новый инстанс схемы в этой же БД
|
||
es.init_db()
|
||
es.append_event("task", "b-1", "TaskSubmitted", {"title": "ok"})
|
||
es.run_projections() # спроецировали id=1
|
||
es.append_event("task", "b-1", "TaskClaimed", {"level": "day"}) # id=2 — в хвосте
|
||
db = sqlite3.connect(_DST)
|
||
db.execute("UPDATE events SET data='{\"level\":\"FAKE\"}' WHERE id=2")
|
||
db.commit(); db.close()
|
||
try:
|
||
es.run_projections()
|
||
check("подделка хвоста останавливает проекции", False, "RuntimeError не брошен")
|
||
except RuntimeError as e:
|
||
check("подделка хвоста останавливает проекции", "audit" in str(e), str(e)[:80])
|
||
|
||
# ── 6. genesis-маркер ретро-строк ──
|
||
db = sqlite3.connect(_DST)
|
||
db.execute("UPDATE events SET data='{\"level\":\"day\"}' WHERE id=2") # снять тампер секции 5
|
||
db.execute("UPDATE events SET data='{\"title\":\"pre-audit\"}', sig=NULL WHERE id=1") # «старая» строка
|
||
db.execute("INSERT INTO projection_meta (name,last_event,updated_at) "
|
||
"VALUES ('sig_genesis_upto', 1, '2026-01-01') "
|
||
"ON CONFLICT(name) DO UPDATE SET last_event=1") # генезис до id=1
|
||
db.commit()
|
||
db.close()
|
||
es._signature_schema_ready = False
|
||
es.init_db() # миграция: строка id=1 ≤ watermark и без sig → genesis
|
||
db = sqlite3.connect(_DST); db.row_factory = sqlite3.Row
|
||
r1 = dict(db.execute("SELECT sig FROM events WHERE id=1").fetchone()); db.close()
|
||
check("genesis-маркер ретро-строки", (r1["sig"] or "").startswith("genesis:"), str(r1))
|
||
rep = audit.verify_events()
|
||
check("genesis-строки не ломают аудит",
|
||
rep["ok"] and rep["genesis"] == 1 and rep["checked"] == 1, str(rep))
|
||
|
||
# ── 7. unsigned-послегенезис строка = legacy, не краш ──
|
||
db = sqlite3.connect(_DST)
|
||
db.execute("INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version) "
|
||
"VALUES ('task','c-1','TaskSubmitted','{\"title\":\"x\"}','2026-01-01T00:00:00Z',1)")
|
||
db.commit(); db.close()
|
||
rep = audit.verify_events()
|
||
check("unsigned-послегенезис = legacy (аудит ок)",
|
||
rep["ok"] and rep["legacy"] == 1, str(rep))
|
||
|
||
# ── 8. конверты команд ──
|
||
signed = audit.sign_command({"op": "task", "title": "переписка"})
|
||
vc = audit.verify_command(signed)
|
||
check("конверт подписан и валиден", vc["ok"] and vc["key_id_match"])
|
||
bad = dict(signed); bad["signature"] = signed["signature"][:-4] + "AAAA"
|
||
check("конверт-тампер ловится", not audit.verify_command(bad)["ok"])
|
||
|
||
# ── 9. CLI verify / audit-key (hermetic) ──
|
||
env = dict(os.environ)
|
||
r = subprocess.run(["/root/simbiot-harness/venv/bin/python", "/root/simbiot-harness/cli.py", "verify"],
|
||
capture_output=True, text=True, timeout=60, env=env)
|
||
check("cli verify exit0", r.returncode == 0, r.stdout + r.stderr)
|
||
r = subprocess.run(["/root/simbiot-harness/venv/bin/python", "/root/simbiot-harness/cli.py", "audit-key"],
|
||
capture_output=True, text=True, timeout=60, env=env)
|
||
check("cli audit-key показывает ключ",
|
||
r.returncode == 0 and "public_b64" in r.stdout, r.stdout + r.stderr)
|
||
|
||
print()
|
||
print("ИТОГ:", "ВСЕ ОК" if not fails else f"ПРОВАЛЫ: {fails}")
|
||
sys.exit(1 if fails else 0) |