simbiot-harness/harness/audit.py

"""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__)))
from . import paths as _paths  # t-261004113351-2858c1: ключи аудита в хоме
KEY_DIR = os.environ.get("SIMBIOT_AUDIT_KEY_DIR", "") or _paths.audit_key_dir()
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, bridge: set = None) -> dict:
    """Проверка списка событий (dict'и sqlite-строк, id-порядок).

    - genesis:<id> — ретро-маркер строк ДО внедрения подписи (доверенная
      граница): маркер сверяется со своим id, chain-link НЕ проверяется
      (у подписанных соседей prev мог указывать на дого-генезисное состояние);
    - sig IS NULL — legacy (неподписанная послегенезисная строка старого
      кода): тоже граница цепочки;
    - иначе подписано: payload (включая СОХРАНЁННЫЙ prev_sig — подпись
      связывает предыдущую строку) — криптопроверка; link-check: prev_sig
      должен указывать на реально существующую более раннюю подписанную
      строку (t-261002072440-аудит: допущена гонка параллельных писателей —
      два демона читали prev до вставки, один линковался «сквозь» строку
      другого; ссылка на любую прошлую sig честна, контент всё равно строго
      криптопроверен). Контент тамперится → подпись не пройдёт.
    """
    pub_b64 = pub_b64 or public_key_b64()
    checked = genesis = legacy = link_checks = 0
    first_bad = None
    last_signed_sig = None  # sig последней проверенной подписанной строки
    # sig всех ранее проверенных подписанных строк + мост сжатия (подписи
    # строк, унесённых в архив harness/compact.py; хеш моста сверен с
    # подписанным StoreCompacted в verify_events ДО вызова)
    earlier_sigs = set(bridge or ())
    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
            # Допускаем ссылку на любую существующую прошлую подписанную
            # строку (гонка писателей), НЕ только на соседнюю: подделать
            # prev нельзя без ключа — подпись payload'а его связывает.
            # boundary:<id> из моста сжатия: перед строкой была неподписанная
            # граница (prev пуст), унесённая в архив — подписанный маркер
            if (prev != last_signed_sig and prev not in earlier_sigs
                    and not (prev is None and f"boundary:{r['id']}" in earlier_sigs)):
                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
        earlier_sigs.add(sig)          # для link-check «любая прошлая строка»
    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))]
    bridge, bad_bridge, compactions = _load_bridge(db)
    db.close()
    if bad_bridge:
        return {"checked": 0, "genesis": 0, "legacy": 0, "link_checks": 0,
                "first_bad": f"bridge:{bad_bridge}", "ok": False, "total": len(rows),
                "compactions": compactions}
    rep = verify_rows(rows, pub_b64, bridge=bridge)
    rep["total"] = len(rows)
    rep["compactions"] = compactions
    return rep


def _load_bridge(db):
    """Мост сжатия: (подписи, сбойное сжатие | None, число сжатий).

    Каждая группа compaction_bridge обязана совпасть по sha256 с
    bridge_sha256 своего StoreCompacted (событие подписано и само проходит
    цепочку) — иначе мост подменён и ссылкам через него не верим."""
    try:
        groups = {}
        for sig, comp in db.execute("SELECT sig, compaction FROM compaction_bridge"):
            groups.setdefault(comp, []).append(sig)
    except sqlite3.OperationalError:
        groups = {}  # стор без сжатий
    signed = {}
    for data, sig in db.execute(
            "SELECT data, sig FROM events WHERE event_type='StoreCompacted'"):
        if not sig or sig.startswith("genesis:"):
            continue  # неподписанное «сжатие» — не основание доверять мосту
        d = json.loads(data)
        signed[d.get("compaction")] = d
    for comp in set(groups) | {c for c, d in signed.items() if d.get("bridge_count")}:
        sigs = groups.get(comp, [])
        d = signed.get(comp)
        h = hashlib.sha256("\n".join(sorted(sigs)).encode()).hexdigest()
        if not d or d.get("bridge_sha256") != h:
            return set(), comp, len(signed)
    return {s for sigs in groups.values() for s in sigs}, None, len(signed)


# ────────────────────────── конверты команд (шина 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]}