256 lines
No EOL
13 KiB
Python
256 lines
No EOL
13 KiB
Python
"""Simbiot Harness — криптографический аудит (t-260930083716).
|
||
|
||
Alex (m-516173240): у харнеса свой приватный ключ; каждое событие стора и
|
||
каждая команда подписываются ним. Цели: полная аудируемость, доказуемое
|
||
авторство («это сделал харнес»), защита стора/командной шины от инъекций
|
||
(подделка ≠ валид).
|
||
|
||
Схема:
|
||
- ed25519 (cryptography), приватный ключ на хосте 0600 —
|
||
<repo>/.audit/ed25519.key, публичный рядом, key_id = отпечаток ключа.
|
||
- Каждое событие при записи (es.append_event) получает sig, key_id и
|
||
prev_sig — подписываются ВСЕ поля канонического представления
|
||
(aggregate|aggregate_id|event_type|canonical_json(data)|timestamp|version)
|
||
+ prev_sig предыдущей строки → тампер-цепочка (переписать середину стора,
|
||
не сломав голову цепочки, практически нельзя).
|
||
- Существующие события ДО внедрения — ретро-маркер генезиса:
|
||
sig='genesis:<id>' (не подписаны криптоключом, честно оговорено;
|
||
после включения всё новое подписано).
|
||
- Верификация: `cli.py verify` — полный аудит всей цепочки; каждая
|
||
проекция (run_projections) проверяет хвостовую дельту и цепочку-голову;
|
||
КРИПТО-ошибка подписи ⇒ RuntimeError — проекции останавливаются
|
||
(подделанное событие не проектируется). Без-подписные «послегенезис»
|
||
строки (писались старым кодом до рестарта демонов) — учитываются
|
||
как legacy, ошибка не поднимается.
|
||
- Команды: cli.py audit-command '<json>' — подписывает конверт команды;
|
||
cli.py verify-command — проверяет конверт (бинарная шина команд).
|
||
|
||
Внешний аудит: cli.py verify (выход 0/1) + audit.py.verify_events —
|
||
может запускаться любой стороной с публичным ключом.
|
||
"""
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import tempfile
|
||
from datetime import datetime, timezone
|
||
|
||
from cryptography.exceptions import InvalidSignature
|
||
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
|
||
Ed25519PrivateKey, Ed25519PublicKey)
|
||
from cryptography.hazmat.primitives import serialization
|
||
|
||
HARNESS_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
KEY_DIR = os.environ.get("SIMBIOT_AUDIT_KEY_DIR", os.path.join(HARNESS_ROOT, ".audit"))
|
||
PRIV_PATH = os.path.join(KEY_DIR, "ed25519.key")
|
||
PUB_PATH = os.path.join(KEY_DIR, "ed25519.pub")
|
||
|
||
KEY_PUB_B64 = None # публичный ключ (base64 raw 32б) — для внешних аудиторов
|
||
_caches = {"priv": None, "key_id": None}
|
||
|
||
|
||
# ────────────────────────── ключи ──────────────────────────
|
||
|
||
def _load_or_create():
|
||
"""Приватный ключ: создать один раз, 0600, атомарно (tmp+rename)."""
|
||
if _caches["priv"] is not None:
|
||
return _caches["priv"]
|
||
os.makedirs(KEY_DIR, mode=0o700, exist_ok=True)
|
||
if os.path.exists(PRIV_PATH):
|
||
with open(PRIV_PATH, "rb") as f:
|
||
priv = serialization.load_pem_private_key(f.read(), password=None)
|
||
if not os.path.exists(PUB_PATH): # ключ создан до появления .pub-логики
|
||
pub = priv.public_key().public_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PublicFormat.SubjectPublicKeyInfo)
|
||
with open(PUB_PATH, "wb") as f:
|
||
f.write(pub)
|
||
os.chmod(PUB_PATH, 0o644)
|
||
else:
|
||
priv = Ed25519PrivateKey.generate()
|
||
pem = priv.private_bytes(
|
||
serialization.Encoding.PEM,
|
||
serialization.PrivateFormat.PKCS8,
|
||
serialization.NoEncryption())
|
||
fd, tmp = tempfile.mkstemp(dir=KEY_DIR, prefix=".key-", suffix=".tmp")
|
||
try:
|
||
os.chmod(tmp, 0o600)
|
||
os.write(fd, pem)
|
||
os.close(fd)
|
||
os.replace(tmp, PRIV_PATH) # атомарно даже при гонке процессов
|
||
finally:
|
||
if os.path.exists(tmp):
|
||
os.unlink(tmp)
|
||
os.chmod(PRIV_PATH, 0o600)
|
||
# публичный ключ рядом — для внешних аудиторов (0644, читать можно всем)
|
||
pub = priv.public_key().public_bytes(
|
||
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
|
||
with open(PUB_PATH, "wb") as f:
|
||
f.write(pub)
|
||
os.chmod(PUB_PATH, 0o644)
|
||
_caches["priv"] = priv
|
||
return priv
|
||
|
||
|
||
def public_key_b64() -> str:
|
||
"""Публичный ключ base64(32б) — публикуется, им проверяют внешние аудиторы."""
|
||
global KEY_PUB_B64
|
||
if KEY_PUB_B64 is None:
|
||
pub = _load_or_create().public_key().public_bytes(
|
||
serialization.Encoding.Raw, serialization.PublicFormat.Raw)
|
||
KEY_PUB_B64 = base64.b64encode(pub).decode()
|
||
return KEY_PUB_B64
|
||
|
||
|
||
def key_id() -> str:
|
||
"""Отпечаток ключа (sha256 публичного, 16 hex) — попадает в каждое событие."""
|
||
if _caches["key_id"] is None:
|
||
digest = hashlib.sha256(base64.b64decode(public_key_b64())).hexdigest()
|
||
_caches["key_id"] = digest[:16]
|
||
return _caches["key_id"]
|
||
|
||
|
||
def key_info() -> dict:
|
||
return {"key_id": key_id(), "public_b64": public_key_b64(),
|
||
"priv_path": PRIV_PATH,
|
||
"priv_mode": (os.stat(PRIV_PATH).st_mode & 0o777) if os.path.exists(PRIV_PATH) else None}
|
||
|
||
|
||
# ────────────────────────── каноника + подпись ──────────────────────────
|
||
|
||
def canonical_data(data_str: str) -> str:
|
||
"""Канонический JSON данных: отсортированные ключи, без пробелов."""
|
||
try:
|
||
return json.dumps(json.loads(data_str or "{}"), ensure_ascii=False,
|
||
sort_keys=True, separators=(",", ":"))
|
||
except Exception:
|
||
return data_str or "" # битые данные не проектируются верификатором
|
||
|
||
|
||
def payload_bytes(aggregate, aggregate_id, event_type, data_str, timestamp,
|
||
version, prev_sig, key_id_) -> bytes:
|
||
parts = [aggregate, aggregate_id, event_type,
|
||
canonical_data(data_str), timestamp, str(version),
|
||
prev_sig or "", key_id_ or ""]
|
||
return "\x1f".join(parts).encode("utf-8")
|
||
|
||
|
||
def sign_bytes(b: bytes) -> str:
|
||
return base64.b64encode(_load_or_create().sign(b)).decode()
|
||
|
||
|
||
def verify_bytes(pub_b64: str, b: bytes, sig_b64: str) -> bool:
|
||
pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64))
|
||
try:
|
||
pub.verify(base64.b64decode(sig_b64), b)
|
||
return True
|
||
except (InvalidSignature, ValueError):
|
||
return False
|
||
|
||
|
||
# ────────────────────────── события ──────────────────────────
|
||
|
||
def sign_event_row(aggregate, aggregate_id, event_type, data_str, timestamp,
|
||
version, prev_sig) -> dict:
|
||
"""Свежие поля подписи для записи события (es.append_event вызывает)."""
|
||
kid = key_id()
|
||
payload = payload_bytes(aggregate, aggregate_id, event_type, data_str,
|
||
timestamp, version, prev_sig, kid)
|
||
return {"sig": sign_bytes(payload), "key_id": kid, "prev_sig": prev_sig or ""}
|
||
|
||
|
||
def verify_rows(rows: list, pub_b64: str = None) -> dict:
|
||
"""Проверка списка событий (dict'и sqlite-строк, id-порядок).
|
||
|
||
- genesis:<id> — ретро-маркер строк ДО внедрения подписи (доверенная
|
||
граница): маркер сверяется со своим id, chain-link НЕ проверяется
|
||
(у подписанных соседей prev мог указывать на дого-генезисное состояние);
|
||
- sig IS NULL — legacy (неподписанная послегенезисная строка старого
|
||
кода): тоже граница цепочки;
|
||
- иначе подписано: payload (включая СОХРАНЁННЫЙ prev_sig — подпись
|
||
связывает предыдущую строку) — криптопроверка; link-check prev_sig ==
|
||
sig предыдущей ПОДПИСАННОЙ строки (после границы skip).
|
||
Контент тамперится → подпись не пройдёт; цепочку двигают → link-check.
|
||
"""
|
||
pub_b64 = pub_b64 or public_key_b64()
|
||
checked = genesis = legacy = link_checks = 0
|
||
first_bad = None
|
||
last_signed_sig = None # sig последней проверенной подписанной строки
|
||
for r in rows:
|
||
sig = r.get("sig")
|
||
if sig and sig.startswith("genesis:"):
|
||
if sig != f"genesis:{r['id']}":
|
||
first_bad = r["id"] if first_bad is None else first_bad
|
||
break
|
||
genesis += 1
|
||
last_signed_sig = None # граница: link-check для следующего — skip
|
||
continue
|
||
if not sig:
|
||
legacy += 1
|
||
last_signed_sig = None
|
||
continue
|
||
checked += 1
|
||
prev = r.get("prev_sig") or None
|
||
if last_signed_sig is not None:
|
||
link_checks += 1
|
||
if prev != last_signed_sig:
|
||
first_bad = r["id"] if first_bad is None else first_bad
|
||
break
|
||
payload = payload_bytes(r["aggregate"], r["aggregate_id"],
|
||
r["event_type"], r["data"], r["timestamp"],
|
||
r["version"], prev, r.get("key_id"))
|
||
if not verify_bytes(pub_b64, payload, sig):
|
||
first_bad = r["id"] if first_bad is None else first_bad
|
||
break
|
||
last_signed_sig = sig
|
||
return {"checked": checked, "genesis": genesis, "legacy": legacy,
|
||
"link_checks": link_checks, "first_bad": first_bad,
|
||
"ok": first_bad is None}
|
||
|
||
|
||
def verify_events(db_path: str = None, after_id: int = 0, limit: int = 10**9,
|
||
pub_b64: str = None) -> dict:
|
||
"""Полный аудит цепочки стора: все события в id-порядке (для cli verify)."""
|
||
if db_path is None:
|
||
from . import es
|
||
db_path = es.DB_PATH
|
||
db = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=10)
|
||
db.row_factory = sqlite3.Row
|
||
rows = [dict(r) for r in db.execute(
|
||
"SELECT * FROM events WHERE id > ? ORDER BY id LIMIT ?",
|
||
(after_id, limit))]
|
||
db.close()
|
||
rep = verify_rows(rows, pub_b64)
|
||
rep["total"] = len(rows)
|
||
return rep
|
||
|
||
|
||
# ────────────────────────── конверты команд (шина CLI) ──────────────────────────
|
||
|
||
def sign_command(command: dict) -> dict:
|
||
"""Подписи команды: canonical JSON конверта + signature (t-260930083716)."""
|
||
envelope = dict(command)
|
||
envelope.setdefault("ts", datetime.now(timezone.utc).isoformat())
|
||
payload = json.dumps(envelope, ensure_ascii=False, sort_keys=True,
|
||
separators=(",", ":")).encode("utf-8")
|
||
kid = key_id()
|
||
return {"envelope": envelope, "key_id": kid,
|
||
"payload_b64": base64.b64encode(payload).decode(),
|
||
"signature": sign_bytes(payload)}
|
||
|
||
|
||
def verify_command(signed: dict) -> dict:
|
||
"""Проверка подписанного конверта команды."""
|
||
try:
|
||
payload = base64.b64decode(signed["payload_b64"])
|
||
ok = verify_bytes(public_key_b64(), payload, signed["signature"])
|
||
recomputed = json.dumps(json.loads(payload.decode()), ensure_ascii=False,
|
||
sort_keys=True, separators=(",", ":"))
|
||
return {"ok": ok, "key_id_match": signed.get("key_id") == key_id(),
|
||
"payload": json.loads(payload.decode()),
|
||
"canonical_match": recomputed == json.dumps(
|
||
json.loads(payload.decode()), ensure_ascii=False,
|
||
sort_keys=True, separators=(",", ":"))}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)[:200]} |