polyc_eventlog/

integrity.rs

//! Per-conversation tamper-evidence over the event log (#799).
//!
//! [`crate::EventLog`] gives durability and ordering but — on its own — no
//! way for a reader to tell that an operator (or a storage-layer bit flip)
//! hasn't quietly rewritten a persisted event between when it was appended
//! and when it was replayed. This module closes that gap by threading a
//! [`polyc_mmr::VerifiableLog`] alongside the journal:
//!
//! 1. Every conversation event becomes one MMR leaf (see
//!    [`rebuild_from_events`] for how a cold log's tree is reconstructed).
//! 2. After extending the tree with a batch of new events, the caller signs
//!    the current root and appends it to the SAME partition as one more
//!    event ([`MMR_SIGNED_ROOT_KIND`]) — see [`extend_and_sign`]. Because it
//!    lands in the same commit as the turn's other events, the root is
//!    signed atomically with the content it covers.
//! 3. A later reader replays the whole partition and calls [`verify_replay`],
//!    which rebuilds the tree from scratch and checks every signed root it
//!    finds along the way — catching a tampered event, a forged or
//!    substituted root marker, or a root signed under the wrong key.
//!
//! This is intentionally decoupled from the journal itself: everything here
//! operates on plain `(kind, payload)` / [`Event`] sequences, so it is
//! testable without a live journal and reusable by anything that replays a
//! partition (the eventlog host, the CLI's `conversation repair`, forensics).

// Several doc summaries here need two sentences to state both the "what"
// and the atomicity/ordering contract in one place, rather than splitting
// across a line an editor of this module would have to re-join to reread.
#![allow(clippy::too_long_first_doc_paragraph)]

use polyc_crypto::approval::ApprovalSigner;
use polyc_mmr::{SignedRoot, VerifiableLog, verify_root_signature};

use crate::Event;

/// Event kind naming a persisted [`SignedRoot`] marker. Namespaced so it
/// cannot collide with any conversation-content kind (every real kind in
/// `polyc-proto`'s `events.proto` is a bare identifier with no `__`
/// wrapping).
pub const MMR_SIGNED_ROOT_KIND: &str = "__mmr_signed_root__";

/// Failures from extending, signing, or verifying a partition's MMR.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum IntegrityError {
    /// The underlying MMR operation failed (lock poisoned, proof error).
    #[error("mmr: {0}")]
    Mmr(#[from] polyc_mmr::MmrError),
    /// A signed-root marker's payload was not the JSON [`SignedRoot`] this
    /// module writes — a corrupted or foreign event landed under the
    /// reserved kind.
    #[error("malformed signed-root marker at leaf count {leaf_count_hint}: {source}")]
    MalformedRoot {
        /// Running leaf count at the point the malformed marker was found,
        /// for locating it in the replay.
        leaf_count_hint: u64,
        /// The JSON decode error.
        source: serde_json::Error,
    },
    /// A persisted root's ed25519 signature does not verify under the
    /// expected signer public key — the marker was forged, or signed by a
    /// different key than the caller trusts.
    #[error("signed root at leaf count {leaf_count} does not verify under the expected signer")]
    SignatureInvalid {
        /// The root's claimed leaf count.
        leaf_count: u64,
    },
    /// The recomputed MMR root (or leaf count) at a checkpoint does not
    /// match what was signed — the tamper-evidence violation this whole
    /// module exists to catch.
    #[error(
        "integrity violation: at leaf count {leaf_count}, replay computed root {computed_root_hex} \
         but the signed marker recorded {expected_root_hex}"
    )]
    RootMismatch {
        /// Leaf count at the point of the mismatch.
        leaf_count: u64,
        /// The root the signed marker claims.
        expected_root_hex: String,
        /// The root replay actually computed.
        computed_root_hex: String,
    },
}

/// Extend `log` with each of `new_events`'s `(kind, payload)` leaves, sign
/// the resulting root with `signer`, and return the [`Event`] to append
/// (kind [`MMR_SIGNED_ROOT_KIND`]) — the caller places it in the SAME
/// journal batch as `new_events` (e.g. right before `turn_complete`) so the
/// signature is atomic with the content it covers.
///
/// # Errors
///
/// Returns [`IntegrityError::Mmr`] if extending the tree or signing fails.
///
/// # Panics
///
/// Never in practice: [`SignedRoot`] always serializes (plain strings and
/// integers), so the internal `expect` cannot fail for any value this
/// module produces.
pub fn extend_and_sign(
    log: &VerifiableLog,
    new_events: &[Event],
    signer: &ApprovalSigner,
) -> Result<Event, IntegrityError> {
    for event in new_events {
        log.append(&event.kind, &event.payload)?;
    }
    let root = log.sign_root(signer)?;
    let payload = serde_json::to_vec(&root).expect("SignedRoot serializes");
    Ok(Event::new(MMR_SIGNED_ROOT_KIND, payload))
}

/// Reconstruct a partition's running MMR from a full replay, for a caller
/// that wants to keep extending it (the eventlog host, on first touching a
/// partition after a restart). Root-marker events themselves are not MMR
/// leaves — only real conversation events are — so this filters them out
/// before delegating to [`VerifiableLog::rebuild`].
///
/// # Errors
///
/// Returns [`IntegrityError::Mmr`] if the rebuild fails.
pub fn rebuild_from_events(events: &[Event]) -> Result<VerifiableLog, IntegrityError> {
    let log = VerifiableLog::rebuild(
        events
            .iter()
            .filter(|e| e.kind != MMR_SIGNED_ROOT_KIND)
            .map(|e| (e.kind.as_str(), e.payload.as_slice())),
    )?;
    Ok(log)
}

/// Verify a full partition replay's tamper-evidence: rebuild the MMR leaf by
/// leaf in append order, and at every [`MMR_SIGNED_ROOT_KIND`] marker check
/// that (a) its signature verifies under `expected_signer_pk_hex` and (b)
/// the tree's root and leaf count at that point match what the marker
/// claims. Returns on the FIRST violation found, naming exactly where it
/// occurred.
///
/// A partition with no signed-root markers at all verifies trivially — this
/// is the transitional state before the first turn completes.
///
/// # Errors
///
/// Returns [`IntegrityError`] describing the first tamper/forgery/mismatch
/// encountered.
pub fn verify_replay(events: &[Event], expected_signer_pk_hex: &str) -> Result<(), IntegrityError> {
    let log = VerifiableLog::new();
    for event in events {
        if event.kind == MMR_SIGNED_ROOT_KIND {
            let leaf_count = log.leaf_count()?;
            let root: SignedRoot = serde_json::from_slice(&event.payload).map_err(|source| {
                IntegrityError::MalformedRoot {
                    leaf_count_hint: leaf_count,
                    source,
                }
            })?;
            let sig_ok = verify_root_signature(&root, expected_signer_pk_hex).unwrap_or(false);
            if !sig_ok {
                return Err(IntegrityError::SignatureInvalid { leaf_count });
            }
            let computed_root = log.root()?;
            let computed_root_hex = hex::encode(computed_root.as_ref());
            if root.leaf_count != leaf_count || root.root_hex != computed_root_hex {
                return Err(IntegrityError::RootMismatch {
                    leaf_count,
                    expected_root_hex: root.root_hex,
                    computed_root_hex,
                });
            }
        } else {
            log.append(&event.kind, &event.payload)?;
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]
    use super::*;

    fn signer() -> ApprovalSigner {
        ApprovalSigner::from_seed(7)
    }

    fn pk_hex(signer: &ApprovalSigner) -> String {
        hex::encode(signer.public_key_bytes())
    }

    /// The pinning test (#799): append a turn's worth of events, sign the
    /// root, replay, flip one byte of one persisted event's payload, and
    /// assert replay reports an integrity violation. Before this module
    /// existed nothing checked this at all — `verify_replay` didn't exist.
    #[test]
    fn mmr_verify_replay_detects_tampered_event() {
        let log = VerifiableLog::new();
        let s = signer();
        let turn_events = vec![
            Event::new("user_msg", b"what is 2+2?".to_vec()),
            Event::new("output_msg", b"4".to_vec()),
        ];
        let marker = extend_and_sign(&log, &turn_events, &s).expect("sign");

        let mut persisted = turn_events.clone();
        persisted.push(marker);

        // Untampered: verifies cleanly.
        verify_replay(&persisted, &pk_hex(&s)).expect("untampered replay must verify");

        // Flip one byte of a persisted event's payload — the "torn write /
        // quiet rewrite" the audit describes.
        persisted[1].payload[0] ^= 0xFF;
        let err = verify_replay(&persisted, &pk_hex(&s))
            .expect_err("tampered replay must report an integrity violation");
        assert!(
            matches!(err, IntegrityError::RootMismatch { .. }),
            "expected a root mismatch, got {err:?}"
        );
    }

    #[test]
    fn mmr_verify_replay_accepts_multi_turn_untampered_log() {
        let log = VerifiableLog::new();
        let s = signer();
        let mut persisted = Vec::new();

        for turn in 0..3u8 {
            let events = vec![
                Event::new("user_msg", vec![turn]),
                Event::new("output_msg", vec![turn, turn]),
            ];
            let marker = extend_and_sign(&log, &events, &s).expect("sign");
            persisted.extend(events);
            persisted.push(marker);
        }

        verify_replay(&persisted, &pk_hex(&s)).expect("three untampered turns must verify");
    }

    #[test]
    fn mmr_verify_replay_rejects_root_signed_by_a_different_key() {
        let log = VerifiableLog::new();
        let s = signer();
        let events = vec![Event::new("user_msg", b"hi".to_vec())];
        let marker = extend_and_sign(&log, &events, &s).expect("sign");
        let mut persisted = events;
        persisted.push(marker);

        let other = ApprovalSigner::from_seed(999);
        let err = verify_replay(&persisted, &pk_hex(&other))
            .expect_err("a root signed under a different key must not verify");
        assert!(matches!(err, IntegrityError::SignatureInvalid { .. }));
    }

    #[test]
    fn mmr_verify_replay_accepts_partition_with_no_signed_roots_yet() {
        let events = vec![Event::new("user_msg", b"no marker yet".to_vec())];
        verify_replay(&events, &pk_hex(&signer())).expect("no markers is trivially fine");
    }

    #[test]
    fn mmr_rebuild_from_events_skips_marker_events() {
        let log = VerifiableLog::new();
        let s = signer();
        let events = vec![
            Event::new("user_msg", b"a".to_vec()),
            Event::new("output_msg", b"b".to_vec()),
        ];
        let marker = extend_and_sign(&log, &events, &s).expect("sign");
        let mut persisted = events;
        persisted.push(marker);

        let rebuilt = rebuild_from_events(&persisted).expect("rebuild");
        assert_eq!(rebuilt.leaf_count().unwrap(), 2, "markers are not leaves");
        assert_eq!(rebuilt.root().unwrap(), log.root().unwrap());
    }
}