diff --git a/Cargo.lock b/Cargo.lock index 539d92b..17c11b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -82,6 +82,7 @@ dependencies = [ "cfr-core", "cfr-crypto", "cfr-media", + "fs2", "hex", "thiserror", ] @@ -248,6 +249,16 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -531,6 +542,28 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "x25519-dalek" version = "3.0.0" diff --git a/Cargo.toml b/Cargo.toml index ab6502a..a807087 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,7 @@ getrandom = { version = "0.4", default-features = false } thiserror = { version = "2.0", default-features = false } arbitrary = { version = "1.4", features = ["derive"] } hex = { version = "0.4", default-features = false, features = ["alloc"] } +fs2 = { version = "0.4.3", default-features = false } [workspace.lints.rust] missing_docs = "warn" diff --git a/README.md b/README.md index b09ba5d..532f401 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ build rather than the call. | feature | effect | |---|---| -| `std` *(default)* | standard library; without it the crates are `no_std` + `alloc` | +| `std` *(default)* | standard library and `cfr::persistence`; without it the crates are `no_std` + `alloc` | | `hwaes` *(default)* | AEGIS-256 with hardware AES; needs a C compiler | | `portable` | AEGIS-256 in pure Rust, no C toolchain | | `pq` | X25519 + ML-KEM-768 hybrid | diff --git a/crates/cfr-core/Cargo.toml b/crates/cfr-core/Cargo.toml index 0cbb9e0..17d5b02 100644 --- a/crates/cfr-core/Cargo.toml +++ b/crates/cfr-core/Cargo.toml @@ -23,6 +23,7 @@ thiserror = { workspace = true } default = ["std"] std = ["thiserror/std", "cfr-crypto/std"] pq = ["cfr-crypto/pq"] +persistence = ["cfr-crypto/persistence"] [lints] workspace = true diff --git a/crates/cfr-core/src/channel.rs b/crates/cfr-core/src/channel.rs index 6b0afab..8ef83b7 100644 --- a/crates/cfr-core/src/channel.rs +++ b/crates/cfr-core/src/channel.rs @@ -16,8 +16,8 @@ pub const MAX_BUFFERED: usize = 64; /// One direction of a pairwise channel. pub struct Chan { - chain: Secret, - next: u64, + pub(crate) chain: Secret, + pub(crate) next: u64, } impl Chan { @@ -74,8 +74,8 @@ impl Chan { /// The receiving side: a ratchet plus a bounded reordering buffer. pub struct RecvChan { - chan: Chan, - buffer: BTreeMap>, + pub(crate) chan: Chan, + pub(crate) buffer: BTreeMap>, } impl RecvChan { diff --git a/crates/cfr-core/src/dag.rs b/crates/cfr-core/src/dag.rs index e07d5dd..cc4fdb3 100644 --- a/crates/cfr-core/src/dag.rs +++ b/crates/cfr-core/src/dag.rs @@ -19,11 +19,11 @@ pub const MAX_OPS: usize = 4096; #[derive(Default)] pub struct Dag { /// Increments on mutation to invalidate derived caches. - epoch: u64, - ops: BTreeMap, - parents: BTreeMap>, + pub(crate) epoch: u64, + pub(crate) ops: BTreeMap, + pub(crate) parents: BTreeMap>, /// Cached transitive ancestor sets, cleared on mutation. - cache: RefCell>>, + pub(crate) cache: RefCell>>, } impl Dag { diff --git a/crates/cfr-core/src/keys.rs b/crates/cfr-core/src/keys.rs index ccbd8ec..8786e40 100644 --- a/crates/cfr-core/src/keys.rs +++ b/crates/cfr-core/src/keys.rs @@ -53,9 +53,9 @@ pub fn version_of(nodes: &BTreeSet, membership_root: &[u8; 32]) -> [u8; 8] /// /// Eviction erases the key material and ends local derivability of that version. pub struct NodeKeys { - keys: BTreeMap>, - order: Vec, - capacity: usize, + pub(crate) keys: BTreeMap>, + pub(crate) order: Vec, + pub(crate) capacity: usize, } impl Default for NodeKeys { diff --git a/crates/cfr-core/src/lib.rs b/crates/cfr-core/src/lib.rs index 2471a5d..50834bd 100644 --- a/crates/cfr-core/src/lib.rs +++ b/crates/cfr-core/src/lib.rs @@ -35,6 +35,9 @@ pub mod op; pub mod prekey; pub mod wire; +#[cfg(feature = "persistence")] +mod state; + pub use checkpoint::{ media_context_id, Capabilities, CheckpointCertificate, CheckpointSignature, ProtocolProfile, ResumptionRecord, PROTOCOL_ID, diff --git a/crates/cfr-core/src/member.rs b/crates/cfr-core/src/member.rs index b5a29e4..b351250 100644 --- a/crates/cfr-core/src/member.rs +++ b/crates/cfr-core/src/member.rs @@ -70,16 +70,16 @@ pub enum Beacon { Unknown, } -struct SendState { - chan: Chan, +pub(crate) struct SendState { + pub(crate) chan: Chan, } -struct RecvState { - eph: DhPublic, - chan: RecvChan, +pub(crate) struct RecvState { + pub(crate) eph: DhPublic, + pub(crate) chan: RecvChan, } -struct Derived { +pub(crate) struct Derived { epoch: u64, guilty: usize, frontier: BTreeSet, @@ -91,7 +91,7 @@ struct Derived { /// Cache key for membership evaluated over one causal dependency set. type AuthorizationCacheKey = ([u8; 32], u64, usize); /// Memoized membership rosters keyed by causal past and derived-state epoch. -type AuthorizationCache = BTreeMap>; +pub(crate) type AuthorizationCache = BTreeMap>; /// A participant that has generated identity material but not yet joined. pub struct PendingJoin { @@ -207,36 +207,36 @@ impl PendingJoin { /// A conference participant. pub struct Participant { - identity: SigSecret, - ipk: SigPublic, - sid: SessionId, - policy: Policy, - seed0: Secret, + pub(crate) identity: SigSecret, + pub(crate) ipk: SigPublic, + pub(crate) sid: SessionId, + pub(crate) policy: Policy, + pub(crate) seed0: Secret, - dag: Dag, - guilty: BTreeSet, + pub(crate) dag: Dag, + pub(crate) guilty: BTreeSet, - prekeys: PrekeyPool, - peer_prekeys: BTreeMap, - send: BTreeMap, - recv: BTreeMap, + pub(crate) prekeys: PrekeyPool, + pub(crate) peer_prekeys: BTreeMap, + pub(crate) send: BTreeMap, + pub(crate) recv: BTreeMap, - nodekeys: NodeKeys, - cparents: BTreeMap>, - absorbed: BTreeSet, - missing: BTreeSet, + pub(crate) nodekeys: NodeKeys, + pub(crate) cparents: BTreeMap>, + pub(crate) absorbed: BTreeSet, + pub(crate) missing: BTreeSet, - pending: Vec, - open_accusations: Vec, + pub(crate) pending: Vec, + pub(crate) open_accusations: Vec, - seen_versions: BTreeMap<[u8; 8], (BTreeSet, [u8; 32])>, - last_version: [u8; 8], + pub(crate) seen_versions: BTreeMap<[u8; 8], (BTreeSet, [u8; 32])>, + pub(crate) last_version: [u8; 8], /// Memoized membership rosters for equivalent causal dependency sets. - authz: RefCell, + pub(crate) authz: RefCell, /// Graph-derived values, invalidated when graph or accusation state changes. - derived: RefCell>, + pub(crate) derived: RefCell>, } impl Participant { @@ -1158,7 +1158,7 @@ impl Participant { } } - fn check_accusation(&self, acc: &Op) -> bool { + pub(crate) fn check_accusation(&self, acc: &Op) -> bool { let Body::Accuse { who, coid, mk, seq } = &acc.body else { return false; }; diff --git a/crates/cfr-core/src/prekey.rs b/crates/cfr-core/src/prekey.rs index 98ad1bb..210cac6 100644 --- a/crates/cfr-core/src/prekey.rs +++ b/crates/cfr-core/src/prekey.rs @@ -15,11 +15,11 @@ pub const SEAL_AFTER: u32 = 64; /// A single prekey generation. pub struct PrekeyPool { - generation: u32, - secret: Option, - public: DhPublic, - established: BTreeSet, - age: u32, + pub(crate) generation: u32, + pub(crate) secret: Option, + pub(crate) public: DhPublic, + pub(crate) established: BTreeSet, + pub(crate) age: u32, } impl PrekeyPool { diff --git a/crates/cfr-core/src/state.rs b/crates/cfr-core/src/state.rs new file mode 100644 index 0000000..6a4799f --- /dev/null +++ b/crates/cfr-core/src/state.rs @@ -0,0 +1,617 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +//! Private, bounded participant-state codec used by the application crate. + +use crate::channel::{Chan, RecvChan, MAX_BUFFERED}; +use crate::codec::{Reader, Writer, MAX_FIELD}; +use crate::dag::{Dag, MAX_OPS}; +use crate::error::{Error, Result}; +use crate::keys::{NodeKeys, OVERLAP}; +use crate::member::{Participant, RecvState, SendState, MAX_PENDING}; +use crate::membership::Policy; +use crate::op::{Kind, Oid, Op, MAX_RECIPIENTS}; +use crate::prekey::{PrekeyPool, SEAL_AFTER}; +use alloc::collections::{BTreeMap, BTreeSet}; +use alloc::vec::Vec; +use cfr_crypto::{DhPublic, DhSecret, Secret, SigPublic, SigSecret, KEY_LEN}; +use core::cell::RefCell; + +type SeenVersion = (BTreeSet, [u8; 32]); +type SeenVersions = BTreeMap<[u8; 8], SeenVersion>; + +fn invalid(message: &'static str) -> Error { + Error::Encoding(message) +} + +fn usize_to_u64(value: usize) -> Result { + u64::try_from(value).map_err(|_| invalid("state integer exceeds u64")) +} + +fn read_usize(reader: &mut Reader<'_>) -> Result { + usize::try_from(reader.u64()?).map_err(|_| invalid("state integer exceeds usize")) +} + +fn write_sig_set(writer: &mut Writer, values: &BTreeSet) { + let values: Vec = values.iter().copied().collect(); + writer.set(&values, |writer, value| { + writer.bytes(value.as_bytes()); + }); +} + +fn read_sig_set(reader: &mut Reader<'_>, limit: usize) -> Result> { + let values: Vec = + reader.set(|reader| Ok(SigPublic::from_bytes(reader.array::<32>()?)))?; + if values.len() > limit { + return Err(invalid("state identity set exceeds limit")); + } + Ok(values.into_iter().collect()) +} + +fn write_oid_set(writer: &mut Writer, values: &BTreeSet) { + let values: Vec = values.iter().copied().collect(); + writer.set(&values, |writer, value| { + writer.bytes(value); + }); +} + +fn read_oid_set(reader: &mut Reader<'_>, limit: usize) -> Result> { + let values: Vec = reader.set(Reader::array::<32>)?; + if values.len() > limit { + return Err(invalid("state operation set exceeds limit")); + } + Ok(values.into_iter().collect()) +} + +fn ensure_strictly_ordered(previous: &mut Option, value: T) -> Result<()> { + if previous.as_ref().is_some_and(|prior| prior >= &value) { + return Err(invalid("state map is not strictly ordered")); + } + *previous = Some(value); + Ok(()) +} + +fn write_policy(writer: &mut Writer, policy: &Policy) -> Result<()> { + write_sig_set(writer, &policy.admins); + writer + .u64(usize_to_u64(policy.quorum)?) + .u64(usize_to_u64(policy.max_ops_per_author)?); + Ok(()) +} + +fn read_policy(reader: &mut Reader<'_>) -> Result { + let policy = Policy { + admins: read_sig_set(reader, MAX_RECIPIENTS)?, + quorum: read_usize(reader)?, + max_ops_per_author: read_usize(reader)?, + }; + if policy.quorum == 0 || policy.max_ops_per_author == 0 { + return Err(invalid("state contains an invalid policy")); + } + Ok(policy) +} + +fn write_prekeys(writer: &mut Writer, prekeys: &PrekeyPool) { + writer.u32(prekeys.generation); + match &prekeys.secret { + Some(secret) => { + writer.u32(1).bytes(&secret.persistence_bytes()); + } + None => { + writer.u32(0); + } + } + writer.bytes(prekeys.public.as_bytes()); + write_sig_set(writer, &prekeys.established); + writer.u32(prekeys.age); +} + +fn read_prekeys(reader: &mut Reader<'_>) -> Result { + let generation = reader.u32()?; + let secret = match reader.u32()? { + 0 => None, + 1 => Some(DhSecret::from_bytes(reader.array::<32>()?)), + _ => return Err(invalid("state prekey presence flag is invalid")), + }; + let public = DhPublic::from_bytes(reader.array::<32>()?); + let established = read_sig_set(reader, MAX_RECIPIENTS)?; + let age = reader.u32()?; + if secret + .as_ref() + .is_some_and(|secret| secret.public() != public) + { + return Err(invalid("state prekey public and private halves differ")); + } + if secret.is_some() && age >= SEAL_AFTER { + return Err(invalid("state retains an expired prekey secret")); + } + Ok(PrekeyPool { + generation, + secret, + public, + established, + age, + }) +} + +fn write_chan(writer: &mut Writer, chan: &Chan) { + writer.bytes(chan.chain.as_bytes()).u64(chan.next); +} + +fn read_chan(reader: &mut Reader<'_>) -> Result { + Ok(Chan { + chain: Secret::from(reader.array::()?), + next: reader.u64()?, + }) +} + +fn write_dag(writer: &mut Writer, dag: &Dag) -> Result<()> { + let order = dag.topological(); + if order.len() != dag.len() { + return Err(invalid("state operation graph contains a cycle")); + } + let operations: Vec<&Op> = order + .iter() + .map(|oid| { + dag.get(oid) + .ok_or_else(|| invalid("state operation graph is inconsistent")) + }) + .collect::>()?; + writer.list(&operations, |writer, operation| { + operation.write(writer); + }); + Ok(()) +} + +fn read_dag(reader: &mut Reader<'_>, sid: &[u8; 32]) -> Result { + let operations: Vec = reader.list(|reader| Op::from_wire(reader.bytes()?))?; + if operations.is_empty() || operations.len() > MAX_OPS { + return Err(invalid("state operation graph size is invalid")); + } + let mut dag = Dag::new(); + let mut seen = BTreeSet::new(); + for operation in operations { + if &operation.sid != sid { + return Err(invalid("state operation belongs to another session")); + } + operation.verify()?; + let oid = operation.oid(); + if !seen.insert(oid) { + return Err(invalid("state operation graph contains a duplicate")); + } + if !operation + .deps + .iter() + .all(|dependency| seen.contains(dependency)) + { + return Err(invalid("state operation graph is not causally ordered")); + } + dag.add(operation)?; + } + Ok(dag) +} + +fn write_operations(writer: &mut Writer, operations: &[Op]) { + writer.list(operations, |writer, operation| { + operation.write(writer); + }); +} + +fn read_operations(reader: &mut Reader<'_>, sid: &[u8; 32], limit: usize) -> Result> { + let operations: Vec = reader.list(|reader| Op::from_wire(reader.bytes()?))?; + if operations.len() > limit { + return Err(invalid("state pending operation list exceeds limit")); + } + let mut ids = BTreeSet::new(); + for operation in &operations { + if &operation.sid != sid { + return Err(invalid( + "state pending operation belongs to another session", + )); + } + operation.verify()?; + if !ids.insert(operation.oid()) { + return Err(invalid("state pending operation list contains a duplicate")); + } + } + Ok(operations) +} + +fn read_peer_prekeys( + reader: &mut Reader<'_>, + ipk: SigPublic, +) -> Result> { + let entries: Vec<(SigPublic, u32, DhPublic)> = reader.list(|reader| { + Ok(( + SigPublic::from_bytes(reader.array::<32>()?), + reader.u32()?, + DhPublic::from_bytes(reader.array::<32>()?), + )) + })?; + if entries.len() > MAX_RECIPIENTS { + return Err(invalid("state peer prekey map exceeds limit")); + } + let mut values = BTreeMap::new(); + let mut previous = None; + for (peer, generation, public) in entries { + ensure_strictly_ordered(&mut previous, peer)?; + if peer == ipk { + return Err(invalid("state contains a self peer prekey")); + } + values.insert(peer, (generation, public)); + } + Ok(values) +} + +fn read_send_channels( + reader: &mut Reader<'_>, + ipk: SigPublic, +) -> Result> { + let entries: Vec<(SigPublic, Chan)> = reader.list(|reader| { + Ok(( + SigPublic::from_bytes(reader.array::<32>()?), + read_chan(reader)?, + )) + })?; + if entries.len() > MAX_RECIPIENTS { + return Err(invalid("state send channel map exceeds limit")); + } + let mut values = BTreeMap::new(); + let mut previous = None; + for (peer, chan) in entries { + ensure_strictly_ordered(&mut previous, peer)?; + if peer == ipk { + return Err(invalid("state contains a self send channel")); + } + values.insert(peer, SendState { chan }); + } + Ok(values) +} + +fn read_recv_chan(reader: &mut Reader<'_>) -> Result { + let chan = read_chan(reader)?; + let entries: Vec<(u64, Vec)> = + reader.list(|reader| Ok((reader.u64()?, reader.bytes()?.to_vec())))?; + if entries.len() > MAX_BUFFERED { + return Err(invalid("state receive buffer exceeds limit")); + } + let mut buffer = BTreeMap::new(); + let mut previous = None; + for (sequence, payload) in entries { + ensure_strictly_ordered(&mut previous, sequence)?; + if sequence <= chan.next || payload.len() < 32 { + return Err(invalid("state receive buffer entry is invalid")); + } + buffer.insert(sequence, payload); + } + Ok(RecvChan { chan, buffer }) +} + +fn read_recv_channels( + reader: &mut Reader<'_>, + ipk: SigPublic, +) -> Result> { + let entries: Vec<(SigPublic, DhPublic, RecvChan)> = reader.list(|reader| { + Ok(( + SigPublic::from_bytes(reader.array::<32>()?), + DhPublic::from_bytes(reader.array::<32>()?), + read_recv_chan(reader)?, + )) + })?; + if entries.len() > MAX_RECIPIENTS { + return Err(invalid("state receive channel map exceeds limit")); + } + let mut values = BTreeMap::new(); + let mut previous = None; + for (peer, eph, chan) in entries { + ensure_strictly_ordered(&mut previous, peer)?; + if peer == ipk { + return Err(invalid("state contains a self receive channel")); + } + values.insert(peer, RecvState { eph, chan }); + } + Ok(values) +} + +fn read_nodekeys(reader: &mut Reader<'_>) -> Result { + let capacity = read_usize(reader)?; + if capacity == 0 || capacity > OVERLAP { + return Err(invalid("state node-key capacity is invalid")); + } + let entries: Vec<(Oid, Secret)> = reader.list(|reader| { + Ok(( + reader.array::<32>()?, + Secret::from(reader.array::()?), + )) + })?; + if entries.len() > capacity { + return Err(invalid("state node-key map exceeds capacity")); + } + let mut keys = BTreeMap::new(); + let mut previous = None; + for (oid, key) in entries { + ensure_strictly_ordered(&mut previous, oid)?; + keys.insert(oid, key); + } + let order: Vec = reader.list(Reader::array::<32>)?; + if order.len() != keys.len() { + return Err(invalid("state node-key order length differs from map")); + } + let mut order_ids = BTreeSet::new(); + for oid in &order { + if !keys.contains_key(oid) || !order_ids.insert(*oid) { + return Err(invalid("state node-key order is inconsistent")); + } + } + Ok(NodeKeys { + keys, + order, + capacity, + }) +} + +fn read_versions(reader: &mut Reader<'_>) -> Result<(SeenVersions, [u8; 8])> { + let entries: Vec<([u8; 8], BTreeSet, [u8; 32])> = reader.list(|reader| { + Ok(( + reader.array::<8>()?, + read_oid_set(reader, MAX_OPS)?, + reader.array::<32>()?, + )) + })?; + if entries.is_empty() || entries.len() > OVERLAP { + return Err(invalid("state version history size is invalid")); + } + let mut versions = BTreeMap::new(); + let mut previous = None; + for (version, nodes, root) in entries { + ensure_strictly_ordered(&mut previous, version)?; + if crate::keys::version_of(&nodes, &root) != version { + return Err(invalid("state version binding is invalid")); + } + versions.insert(version, (nodes, root)); + } + let last = reader.array::<8>()?; + if !versions.contains_key(&last) { + return Err(invalid("state last version is not retained")); + } + Ok((versions, last)) +} + +fn validate_contribution_state(participant: &Participant) -> Result<()> { + let contribution_ids: BTreeSet = participant + .dag + .iter() + .filter(|(_, operation)| operation.kind() == Kind::Contrib) + .map(|(oid, _)| *oid) + .collect(); + if !participant.absorbed.is_disjoint(&participant.missing) + || !participant.absorbed.is_subset(&contribution_ids) + || !participant.missing.is_subset(&contribution_ids) + || !participant + .nodekeys + .keys + .keys() + .all(|oid| contribution_ids.contains(oid)) + { + return Err(invalid("state contribution tracking is inconsistent")); + } + if participant.pending.iter().any(|operation| { + participant.dag.contains(&operation.oid()) + || operation + .deps + .iter() + .all(|dependency| participant.dag.contains(dependency)) + }) { + return Err(invalid("state pending operation is not pending")); + } + Ok(()) +} + +fn validate_participant(participant: &Participant) -> Result<()> { + validate_contribution_state(participant)?; + for guilty in &participant.guilty { + let proven = participant.dag.iter().any(|(_, operation)| { + matches!( + &operation.body, + crate::op::Body::Accuse { who, .. } if who == guilty + ) && participant.check_accusation(operation) + }); + if !proven { + return Err(invalid("state guilty set lacks valid evidence")); + } + } + Ok(()) +} + +impl Participant { + /// Encodes all non-derived participant state for the application boundary. + #[doc(hidden)] + pub fn export_persistence_state(&self) -> Result> { + let mut writer = Writer::new(); + writer + .bytes(&self.identity.persistence_seed()) + .bytes(self.ipk.as_bytes()) + .bytes(&self.sid); + write_policy(&mut writer, &self.policy)?; + writer.bytes(self.seed0.as_bytes()); + write_dag(&mut writer, &self.dag)?; + write_sig_set(&mut writer, &self.guilty); + write_prekeys(&mut writer, &self.prekeys); + + let peer_prekeys: Vec<_> = self.peer_prekeys.iter().collect(); + writer.list(&peer_prekeys, |writer, (peer, (generation, public))| { + writer + .bytes(peer.as_bytes()) + .u32(*generation) + .bytes(public.as_bytes()); + }); + + let send: Vec<_> = self.send.iter().collect(); + writer.list(&send, |writer, (peer, state)| { + writer.bytes(peer.as_bytes()); + write_chan(writer, &state.chan); + }); + + let recv: Vec<_> = self.recv.iter().collect(); + writer.list(&recv, |writer, (peer, state)| { + writer.bytes(peer.as_bytes()).bytes(state.eph.as_bytes()); + write_chan(writer, &state.chan.chan); + let buffered: Vec<_> = state.chan.buffer.iter().collect(); + writer.list(&buffered, |writer, (sequence, payload)| { + writer.u64(**sequence).bytes(payload); + }); + }); + + writer.u64(usize_to_u64(self.nodekeys.capacity)?); + let keys: Vec<_> = self.nodekeys.keys.iter().collect(); + writer.list(&keys, |writer, (oid, key)| { + writer.bytes(*oid).bytes(key.as_bytes()); + }); + writer.list(&self.nodekeys.order, |writer, oid| { + writer.bytes(oid); + }); + + write_oid_set(&mut writer, &self.absorbed); + write_oid_set(&mut writer, &self.missing); + write_operations(&mut writer, &self.pending); + write_operations(&mut writer, &self.open_accusations); + + let versions: Vec<_> = self.seen_versions.iter().collect(); + writer.list(&versions, |writer, (version, (nodes, root))| { + writer.bytes(*version); + write_oid_set(writer, nodes); + writer.bytes(root); + }); + writer.bytes(&self.last_version); + let bytes = writer.finish(); + if bytes.len() > MAX_FIELD { + return Err(Error::LimitExceeded( + "participant state exceeds codec limit", + )); + } + Ok(bytes) + } + + /// Reconstructs participant state and rejects malformed or inconsistent input. + #[doc(hidden)] + pub fn import_persistence_state(bytes: &[u8]) -> Result { + if bytes.len() > MAX_FIELD { + return Err(Error::LimitExceeded( + "participant state exceeds codec limit", + )); + } + let mut reader = Reader::new(bytes); + let identity = SigSecret::from_seed(&reader.array::<32>()?); + let ipk = SigPublic::from_bytes(reader.array::<32>()?); + if identity.public() != ipk { + return Err(invalid("state identity binding is invalid")); + } + let sid = reader.array::<32>()?; + let policy = read_policy(&mut reader)?; + let seed0 = Secret::from(reader.array::()?); + let dag = read_dag(&mut reader, &sid)?; + let guilty = read_sig_set(&mut reader, MAX_RECIPIENTS)?; + let prekeys = read_prekeys(&mut reader)?; + let peer_prekeys = read_peer_prekeys(&mut reader, ipk)?; + let send = read_send_channels(&mut reader, ipk)?; + let recv = read_recv_channels(&mut reader, ipk)?; + let nodekeys = read_nodekeys(&mut reader)?; + let absorbed = read_oid_set(&mut reader, MAX_OPS)?; + let missing = read_oid_set(&mut reader, MAX_OPS)?; + let pending = read_operations(&mut reader, &sid, MAX_PENDING)?; + let open_accusations = read_operations(&mut reader, &sid, MAX_PENDING)?; + let (seen_versions, last_version) = read_versions(&mut reader)?; + reader.finish()?; + + let mut cparents = BTreeMap::new(); + for (oid, operation) in dag.iter() { + if let crate::op::Body::Contrib { + cparents: parents, .. + } = &operation.body + { + cparents.insert(*oid, parents.clone()); + } + } + + let participant = Self { + identity, + ipk, + sid, + policy, + seed0, + dag, + guilty, + prekeys, + peer_prekeys, + send, + recv, + nodekeys, + cparents, + absorbed, + missing, + pending, + open_accusations, + seen_versions, + last_version, + authz: RefCell::new(BTreeMap::new()), + derived: RefCell::new(None), + }; + validate_participant(&participant)?; + Ok(participant) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn participant_state_roundtrips_canonically() { + let (mut participant, _) = Participant::create(Policy::leaderless(2)).unwrap(); + participant.tick(); + participant.tick(); + let original = participant.export_persistence_state().unwrap(); + let restored = Participant::import_persistence_state(&original).unwrap(); + assert_eq!(restored.identity(), participant.identity()); + assert_eq!(restored.session_id(), participant.session_id()); + assert_eq!(restored.members(), participant.members()); + assert_eq!(restored.version(), participant.version()); + assert_eq!(restored.export_persistence_state().unwrap(), original); + } + + #[test] + fn participant_state_rejects_trailing_bytes_and_identity_substitution() { + let (participant, _) = Participant::create(Policy::leaderless(2)).unwrap(); + let state = participant.export_persistence_state().unwrap(); + + let mut trailing = state.clone(); + trailing.push(0); + assert!(Participant::import_persistence_state(&trailing).is_err()); + + let mut substituted = state; + // The first TLV field is a 32-byte identity seed: tag + u32 length. + substituted[5] ^= 1; + assert!(Participant::import_persistence_state(&substituted).is_err()); + } + + #[test] + fn departed_prekey_peers_do_not_make_valid_state_unloadable() { + let (mut participant, _) = Participant::create(Policy::leaderless(2)).unwrap(); + for value in 1..=8 { + participant + .prekeys + .established + .insert(SigPublic::from_bytes([value; 32])); + } + let state = participant.export_persistence_state().unwrap(); + let restored = Participant::import_persistence_state(&state).unwrap(); + assert_eq!( + restored.prekeys.established, + participant.prekeys.established + ); + } +} diff --git a/crates/cfr-crypto/Cargo.toml b/crates/cfr-crypto/Cargo.toml index 051097c..4c7a76b 100644 --- a/crates/cfr-crypto/Cargo.toml +++ b/crates/cfr-crypto/Cargo.toml @@ -32,6 +32,9 @@ hex = { workspace = true } [features] default = ["std", "hwaes"] std = ["thiserror/std", "blake3/std", "aegis/std"] +# Internal secret-state hooks used by the application persistence boundary. +# This feature remains `no_std`; it does not add storage or serialization deps. +persistence = [] # Hardware-accelerated AEGIS-256 (AES-NI / ARM crypto extensions) via the # reference C backend. Requires a C compiler at build time. hwaes = [] diff --git a/crates/cfr-crypto/src/aead.rs b/crates/cfr-crypto/src/aead.rs index db13b23..1bc3b5f 100644 --- a/crates/cfr-crypto/src/aead.rs +++ b/crates/cfr-crypto/src/aead.rs @@ -22,7 +22,8 @@ type CipherShort = Aegis256; /// Encrypts `plaintext`, returning `ciphertext || tag`. pub fn aead_seal(key: &[u8; 32], nonce: &[u8; 32], plaintext: &[u8], ad: &[u8]) -> Vec { - let (mut ct, tag) = Cipher::new(key, nonce).encrypt(plaintext, ad); + let mut ct = plaintext.to_vec(); + let tag = Cipher::new(key, nonce).encrypt_in_place(&mut ct, ad); ct.extend_from_slice(&tag); ct } @@ -39,9 +40,11 @@ pub fn aead_open( } let (body, tag) = ciphertext.split_at(ciphertext.len() - TAG_LEN); let tag: [u8; TAG_LEN] = tag.try_into().map_err(|_| CryptoError::Truncated)?; + let mut plaintext = body.to_vec(); Cipher::new(key, nonce) - .decrypt(body, &tag, ad) - .map_err(|_| CryptoError::BadTag) + .decrypt_in_place(&mut plaintext, &tag, ad) + .map_err(|_| CryptoError::BadTag)?; + Ok(plaintext) } /// Encrypts `buf` in place and returns the detached tag. diff --git a/crates/cfr-crypto/src/dh.rs b/crates/cfr-crypto/src/dh.rs index 3c24db3..93152a7 100644 --- a/crates/cfr-crypto/src/dh.rs +++ b/crates/cfr-crypto/src/dh.rs @@ -36,6 +36,16 @@ impl DhSecret { Self(StaticSecret::from(bytes)) } + /// Returns the raw key for the internal persistence codec. + /// + /// This deliberately secret-bearing hook is available only when the + /// application persistence feature is enabled. + #[cfg(feature = "persistence")] + #[doc(hidden)] + pub fn persistence_bytes(&self) -> [u8; 32] { + self.0.to_bytes() + } + /// The matching public key. pub fn public(&self) -> DhPublic { DhPublic(PublicKey::from(&self.0).to_bytes()) diff --git a/crates/cfr-crypto/src/sig.rs b/crates/cfr-crypto/src/sig.rs index c8e2ad0..4f94953 100644 --- a/crates/cfr-crypto/src/sig.rs +++ b/crates/cfr-crypto/src/sig.rs @@ -41,6 +41,16 @@ impl SigSecret { Self(SigningKey::from_bytes(seed)) } + /// Returns the seed for the internal persistence codec. + /// + /// This deliberately secret-bearing hook is available only when the + /// application persistence feature is enabled. + #[cfg(feature = "persistence")] + #[doc(hidden)] + pub fn persistence_seed(&self) -> [u8; 32] { + self.0.to_bytes() + } + /// The matching identity public key. pub fn public(&self) -> SigPublic { SigPublic(self.0.verifying_key().to_bytes()) diff --git a/crates/cfr-media/Cargo.toml b/crates/cfr-media/Cargo.toml index 4aa5caf..025a028 100644 --- a/crates/cfr-media/Cargo.toml +++ b/crates/cfr-media/Cargo.toml @@ -22,6 +22,7 @@ thiserror = { workspace = true } [features] default = ["std"] std = ["thiserror/std", "cfr-crypto/std"] +persistence = ["cfr-crypto/persistence"] [lints] workspace = true diff --git a/crates/cfr-media/src/error.rs b/crates/cfr-media/src/error.rs index 0b900e0..fce16ec 100644 --- a/crates/cfr-media/src/error.rs +++ b/crates/cfr-media/src/error.rs @@ -39,6 +39,9 @@ pub enum Error { /// The per-sender frame counter would wrap. Rekey instead. #[cfg_attr(feature = "std", error("frame counter exhausted"))] CounterExhausted, + /// Persisted media state was malformed or violated a security invariant. + #[cfg_attr(feature = "std", error("malformed persisted media state"))] + MalformedState, } #[cfg(not(feature = "std"))] diff --git a/crates/cfr-media/src/frame.rs b/crates/cfr-media/src/frame.rs index 2af494f..57e2821 100644 --- a/crates/cfr-media/src/frame.rs +++ b/crates/cfr-media/src/frame.rs @@ -34,7 +34,7 @@ pub fn sender_tag(id: &SigPublic) -> SenderTag { .expect("hash has at least eight bytes") } -fn context_id(version: [u8; 8]) -> ContextId { +pub(crate) fn context_id(version: [u8; 8]) -> ContextId { let digest = hash(b"cfr/media/context", &[&version]); digest[..16] .try_into() @@ -146,29 +146,29 @@ fn scatter(frame: &mut [u8], l: &Layout, data: &[u8]) { } } -struct VersionKeys { - version: [u8; 8], - key: Secret, - roster: BTreeMap, - send: SendRatchet, - recv: BTreeMap, +pub(crate) struct VersionKeys { + pub(crate) version: [u8; 8], + pub(crate) key: Secret, + pub(crate) roster: BTreeMap, + pub(crate) send: SendRatchet, + pub(crate) recv: BTreeMap, /// Frames sent under this context. /// /// The index is context-scoped to prevent nonce reuse after reselection. - counter: u64, + pub(crate) counter: u64, } /// Protects and opens media frames for one participant. /// /// Recent versions tolerate in-flight rekeys; eviction erases older key material. pub struct Protector { - sid: [u8; 32], - me: SigPublic, - my_tag: SenderTag, - versions: BTreeMap, - order: VecDeque, - current: Option, - retain: usize, + pub(crate) sid: [u8; 32], + pub(crate) me: SigPublic, + pub(crate) my_tag: SenderTag, + pub(crate) versions: BTreeMap, + pub(crate) order: VecDeque, + pub(crate) current: Option, + pub(crate) retain: usize, } impl Protector { diff --git a/crates/cfr-media/src/lib.rs b/crates/cfr-media/src/lib.rs index b99fccf..2bb504a 100644 --- a/crates/cfr-media/src/lib.rs +++ b/crates/cfr-media/src/lib.rs @@ -29,6 +29,9 @@ pub mod frame; pub mod ratchet; pub mod replay; +#[cfg(feature = "persistence")] +mod state; + pub use codec::{layout, Codec, Layout}; pub use error::{Error, Result}; pub use frame::{sender_tag, ContextId, Protector, SenderTag, Trailer, TRAILER_LEN}; diff --git a/crates/cfr-media/src/ratchet.rs b/crates/cfr-media/src/ratchet.rs index b5b5223..e704647 100644 --- a/crates/cfr-media/src/ratchet.rs +++ b/crates/cfr-media/src/ratchet.rs @@ -21,8 +21,8 @@ fn base(group: &Secret, sender: &SigPublic) -> Secret { /// The sending side of a sender-specific media ratchet. pub struct SendRatchet { - chain: Secret, - epoch: u64, + pub(crate) chain: Secret, + pub(crate) epoch: u64, } impl SendRatchet { @@ -59,8 +59,8 @@ impl SendRatchet { /// The caller commits a cloned instance only after frame authentication. #[derive(Clone)] pub struct RecvRatchet { - chain: Secret, - epoch: u64, + pub(crate) chain: Secret, + pub(crate) epoch: u64, } impl RecvRatchet { diff --git a/crates/cfr-media/src/replay.rs b/crates/cfr-media/src/replay.rs index f492b4a..3af6b17 100644 --- a/crates/cfr-media/src/replay.rs +++ b/crates/cfr-media/src/replay.rs @@ -13,9 +13,9 @@ pub const WINDOW: u64 = 64; /// A sliding replay window over authenticated frame indices. #[derive(Debug, Default, Clone)] pub struct Replay { - high: u64, - seen: u64, - started: bool, + pub(crate) high: u64, + pub(crate) seen: u64, + pub(crate) started: bool, } impl Replay { diff --git a/crates/cfr-media/src/state.rs b/crates/cfr-media/src/state.rs new file mode 100644 index 0000000..f6de40e --- /dev/null +++ b/crates/cfr-media/src/state.rs @@ -0,0 +1,386 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +//! Private, bounded media-state codec used by the application crate. + +use crate::error::{Error, Result}; +use crate::frame::{context_id, sender_tag, ContextId, Protector, SenderTag, VersionKeys}; +use crate::ratchet::{RecvRatchet, SendRatchet, EPOCH}; +use crate::replay::Replay; +use alloc::collections::{BTreeMap, BTreeSet, VecDeque}; +use alloc::vec::Vec; +use cfr_crypto::{Secret, SigPublic, KEY_LEN}; + +const MAX_MEDIA_STATE_BYTES: usize = 1 << 22; +const MAX_RETAINED_VERSIONS: usize = 64; +const MAX_ROSTER: usize = 256; + +#[derive(Default)] +struct Writer { + bytes: Vec, +} + +impl Writer { + fn u8(&mut self, value: u8) { + self.bytes.push(value); + } + + fn u32(&mut self, value: u32) { + self.bytes.extend_from_slice(&value.to_be_bytes()); + } + + fn u64(&mut self, value: u64) { + self.bytes.extend_from_slice(&value.to_be_bytes()); + } + + fn fixed(&mut self, value: &[u8]) { + self.bytes.extend_from_slice(value); + } + + fn count(&mut self, value: usize) -> Result<()> { + self.u32(u32::try_from(value).map_err(|_| Error::MalformedState)?); + Ok(()) + } + + fn finish(self) -> Vec { + self.bytes + } +} + +struct Reader<'a> { + bytes: &'a [u8], + position: usize, +} + +impl<'a> Reader<'a> { + fn new(bytes: &'a [u8]) -> Result { + if bytes.len() > MAX_MEDIA_STATE_BYTES { + return Err(Error::MalformedState); + } + Ok(Self { bytes, position: 0 }) + } + + fn take(&mut self, length: usize) -> Result<&'a [u8]> { + let end = self + .position + .checked_add(length) + .ok_or(Error::MalformedState)?; + let value = self + .bytes + .get(self.position..end) + .ok_or(Error::MalformedState)?; + self.position = end; + Ok(value) + } + + fn array(&mut self) -> Result<[u8; N]> { + self.take(N)?.try_into().map_err(|_| Error::MalformedState) + } + + fn u8(&mut self) -> Result { + self.take(1)?.first().copied().ok_or(Error::MalformedState) + } + + fn u32(&mut self) -> Result { + Ok(u32::from_be_bytes(self.array()?)) + } + + fn u64(&mut self) -> Result { + Ok(u64::from_be_bytes(self.array()?)) + } + + fn count(&mut self, limit: usize) -> Result { + let value = usize::try_from(self.u32()?).map_err(|_| Error::MalformedState)?; + if value > limit { + return Err(Error::MalformedState); + } + Ok(value) + } + + fn finish(self) -> Result<()> { + if self.position == self.bytes.len() { + Ok(()) + } else { + Err(Error::MalformedState) + } + } +} + +fn read_bool(reader: &mut Reader<'_>) -> Result { + match reader.u8()? { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(Error::MalformedState), + } +} + +fn expected_send_epoch(counter: u64) -> u64 { + counter.saturating_sub(1) / EPOCH +} + +fn read_roster(reader: &mut Reader<'_>) -> Result> { + let count = reader.count(MAX_ROSTER)?; + let mut roster = BTreeMap::new(); + let mut previous = None; + for _ in 0..count { + let tag = reader.array::<8>()?; + if previous.is_some_and(|value| value >= tag) { + return Err(Error::MalformedState); + } + previous = Some(tag); + let identity = SigPublic::from_bytes(reader.array::<32>()?); + if sender_tag(&identity) != tag { + return Err(Error::MalformedState); + } + roster.insert(tag, identity); + } + Ok(roster) +} + +fn read_receivers( + reader: &mut Reader<'_>, + roster: &BTreeMap, + my_tag: SenderTag, +) -> Result> { + let count = reader.count(roster.len())?; + let mut receivers = BTreeMap::new(); + let mut previous = None; + for _ in 0..count { + let tag = reader.array::<8>()?; + if previous.is_some_and(|value| value >= tag) || tag == my_tag || !roster.contains_key(&tag) + { + return Err(Error::MalformedState); + } + previous = Some(tag); + let ratchet = RecvRatchet { + chain: Secret::from(reader.array::()?), + epoch: reader.u64()?, + }; + let replay = Replay { + high: reader.u64()?, + seen: reader.u64()?, + started: read_bool(reader)?, + }; + if (!replay.started && (replay.high != 0 || replay.seen != 0)) + || (replay.started && (replay.seen & 1 == 0 || ratchet.epoch != replay.high / EPOCH)) + { + return Err(Error::MalformedState); + } + receivers.insert(tag, (ratchet, replay)); + } + Ok(receivers) +} + +fn read_version( + reader: &mut Reader<'_>, + context: ContextId, + my_tag: SenderTag, +) -> Result { + let version = reader.array::<8>()?; + if context_id(version) != context { + return Err(Error::MalformedState); + } + let key = Secret::from(reader.array::()?); + let roster = read_roster(reader)?; + let send = SendRatchet { + chain: Secret::from(reader.array::()?), + epoch: reader.u64()?, + }; + let recv = read_receivers(reader, &roster, my_tag)?; + let counter = reader.u64()?; + if send.epoch != expected_send_epoch(counter) { + return Err(Error::MalformedState); + } + Ok(VersionKeys { + version, + key, + roster, + send, + recv, + counter, + }) +} + +fn read_order( + reader: &mut Reader<'_>, + versions: &BTreeMap, + retain: usize, +) -> Result> { + let count = reader.count(retain)?; + if count != versions.len() { + return Err(Error::MalformedState); + } + let mut order = VecDeque::with_capacity(count); + let mut unique = BTreeSet::new(); + for _ in 0..count { + let context = reader.array::<16>()?; + if !versions.contains_key(&context) || !unique.insert(context) { + return Err(Error::MalformedState); + } + order.push_back(context); + } + Ok(order) +} + +impl Protector { + /// Encodes all media ratchets, counters, and replay windows. + #[doc(hidden)] + pub fn export_persistence_state(&self) -> Result> { + let mut writer = Writer::default(); + writer.fixed(&self.sid); + writer.fixed(self.me.as_bytes()); + writer.fixed(&self.my_tag); + writer.count(self.retain)?; + writer.count(self.versions.len())?; + for (context, version) in &self.versions { + writer.fixed(context); + writer.fixed(&version.version); + writer.fixed(version.key.as_bytes()); + writer.count(version.roster.len())?; + for (tag, identity) in &version.roster { + writer.fixed(tag); + writer.fixed(identity.as_bytes()); + } + writer.fixed(version.send.chain.as_bytes()); + writer.u64(version.send.epoch); + writer.count(version.recv.len())?; + for (tag, (ratchet, replay)) in &version.recv { + writer.fixed(tag); + writer.fixed(ratchet.chain.as_bytes()); + writer.u64(ratchet.epoch); + writer.u64(replay.high); + writer.u64(replay.seen); + writer.u8(u8::from(replay.started)); + } + writer.u64(version.counter); + } + writer.count(self.order.len())?; + for context in &self.order { + writer.fixed(context); + } + match self.current { + Some(context) => { + writer.u8(1); + writer.fixed(&context); + } + None => writer.u8(0), + } + let bytes = writer.finish(); + if bytes.len() > MAX_MEDIA_STATE_BYTES { + return Err(Error::MalformedState); + } + Ok(bytes) + } + + /// Reconstructs media state and rejects malformed or inconsistent input. + #[doc(hidden)] + pub fn import_persistence_state(bytes: &[u8]) -> Result { + let mut reader = Reader::new(bytes)?; + let sid = reader.array::<32>()?; + let me = SigPublic::from_bytes(reader.array::<32>()?); + let my_tag = reader.array::<8>()?; + if my_tag != sender_tag(&me) { + return Err(Error::MalformedState); + } + let retain = reader.count(MAX_RETAINED_VERSIONS)?; + if retain == 0 { + return Err(Error::MalformedState); + } + let version_count = reader.count(retain)?; + let mut versions = BTreeMap::new(); + let mut previous_context: Option = None; + for _ in 0..version_count { + let context = reader.array::<16>()?; + if previous_context.is_some_and(|previous| previous >= context) { + return Err(Error::MalformedState); + } + previous_context = Some(context); + versions.insert(context, read_version(&mut reader, context, my_tag)?); + } + let order = read_order(&mut reader, &versions, retain)?; + let current = match reader.u8()? { + 0 => None, + 1 => Some(reader.array::<16>()?), + _ => return Err(Error::MalformedState), + }; + reader.finish()?; + if current.is_some_and(|context| !versions.contains_key(&context)) + || (versions.is_empty() != current.is_none()) + { + return Err(Error::MalformedState); + } + Ok(Self { + sid, + me, + my_tag, + versions, + order, + current, + retain, + }) + } + + /// Validates the cross-layer session, identity, key, and roster binding. + #[doc(hidden)] + pub fn validate_persistence_binding( + &self, + sid: [u8; 32], + me: SigPublic, + current: Option<([u8; 8], &Secret, &BTreeSet)>, + ) -> Result<()> { + if self.sid != sid || self.me != me || self.my_tag != sender_tag(&me) { + return Err(Error::MalformedState); + } + if let Some((version, key, members)) = current { + let context = context_id(version); + if self.current != Some(context) { + return Err(Error::MalformedState); + } + let stored = self.versions.get(&context).ok_or(Error::MalformedState)?; + let roster: BTreeSet = stored.roster.values().copied().collect(); + if stored.version != version || !stored.key.ct_eq(key) || &roster != members { + return Err(Error::MalformedState); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Codec; + use cfr_crypto::SigSecret; + + #[test] + fn media_state_preserves_sender_counter() { + let identity = SigSecret::from_seed(&[7; 32]); + let mut protector = Protector::new([3; 32], identity.public(), 4); + protector.install([9; 8], &Secret::from([4; KEY_LEN]), [identity.public()]); + let first = protector.protect(Codec::Generic, b"frame", false).unwrap(); + assert_eq!(Protector::inspect(&first).unwrap().counter, 0); + + let bytes = protector.export_persistence_state().unwrap(); + let mut restored = Protector::import_persistence_state(&bytes).unwrap(); + assert_eq!(restored.export_persistence_state().unwrap(), bytes); + let second = restored.protect(Codec::Generic, b"frame", false).unwrap(); + assert_eq!(Protector::inspect(&second).unwrap().counter, 1); + assert_ne!(first, second); + } + + #[test] + fn media_state_rejects_trailing_bytes() { + let identity = SigSecret::from_seed(&[7; 32]); + let protector = Protector::new([3; 32], identity.public(), 4); + let mut bytes = protector.export_persistence_state().unwrap(); + bytes.push(0); + assert_eq!( + Protector::import_persistence_state(&bytes).err(), + Some(Error::MalformedState) + ); + } +} diff --git a/crates/cfr/Cargo.toml b/crates/cfr/Cargo.toml index e2567c1..f0cd5af 100644 --- a/crates/cfr/Cargo.toml +++ b/crates/cfr/Cargo.toml @@ -23,13 +23,23 @@ cfr-core = { workspace = true } cfr-crypto = { workspace = true } cfr-media = { workspace = true } thiserror = { workspace = true } +fs2 = { workspace = true, optional = true } [dev-dependencies] hex = { workspace = true } [features] default = ["std"] -std = ["thiserror/std", "cfr-core/std", "cfr-crypto/std", "cfr-media/std"] +std = [ + "thiserror/std", + "cfr-core/std", + "cfr-core/persistence", + "cfr-crypto/std", + "cfr-crypto/persistence", + "cfr-media/std", + "cfr-media/persistence", + "dep:fs2", +] # Post-quantum profile: X25519 + ML-KEM-768 hybrid. pq = ["cfr-core/pq", "cfr-crypto/pq"] diff --git a/crates/cfr/src/conference.rs b/crates/cfr/src/conference.rs index c706bd7..2c44236 100644 --- a/crates/cfr/src/conference.rs +++ b/crates/cfr/src/conference.rs @@ -259,7 +259,9 @@ impl Conference { /// Leaves the conference. pub fn leave(&mut self) -> Result { - Ok(self.core.leave()?.into()) + let message = self.core.leave()?; + self.refresh_media(); + Ok(message.into()) } /// Contributes fresh entropy, moving the key forward. @@ -323,4 +325,68 @@ impl Conference { pub fn inspect(packet: &[u8]) -> Result { Ok(Protector::inspect(packet)?) } + + #[cfg(feature = "std")] + pub(crate) fn export_persistence_state(&self) -> Result> { + let core = self.core.export_persistence_state()?; + let media = self.media.export_persistence_state()?; + let mut writer = cfr_core::codec::Writer::new(); + writer.bytes(&core).bytes(&media); + Ok(writer.finish()) + } + + #[cfg(feature = "std")] + pub(crate) fn import_persistence_state(bytes: &[u8]) -> Result { + let mut reader = cfr_core::codec::Reader::new(bytes); + let core_bytes = reader.bytes()?; + let media_bytes = reader.bytes()?; + reader.finish()?; + + let core = Participant::import_persistence_state(core_bytes)?; + let media = Protector::import_persistence_state(media_bytes)?; + let group_key = core.group_key(); + let members = core.members(); + media.validate_persistence_binding( + core.session_id(), + core.identity(), + group_key + .as_ref() + .map(|key| (core.version(), key, &members)), + )?; + Ok(Self { core, media }) + } +} + +#[cfg(all(test, feature = "std"))] +mod tests { + use super::*; + + #[test] + fn conference_state_preserves_identity_and_media_counter() { + let (mut conference, _) = Conference::create(Policy::leaderless(2)).unwrap(); + let first = conference + .protect(Codec::Generic, b"persistent frame", false) + .unwrap(); + let state = conference.export_persistence_state().unwrap(); + let mut restored = Conference::import_persistence_state(&state).unwrap(); + assert_eq!(restored.identity(), conference.identity()); + assert_eq!(restored.session_id(), conference.session_id()); + assert_eq!(restored.members(), conference.members()); + assert_eq!(restored.version(), conference.version()); + assert_eq!(restored.export_persistence_state().unwrap(), state); + + let second = restored + .protect(Codec::Generic, b"persistent frame", false) + .unwrap(); + assert_eq!(Protector::inspect(&first).unwrap().counter, 0); + assert_eq!(Protector::inspect(&second).unwrap().counter, 1); + } + + #[test] + fn conference_state_rejects_trailing_bytes() { + let (conference, _) = Conference::create(Policy::leaderless(2)).unwrap(); + let mut state = conference.export_persistence_state().unwrap(); + state.push(0); + assert!(Conference::import_persistence_state(&state).is_err()); + } } diff --git a/crates/cfr/src/lib.rs b/crates/cfr/src/lib.rs index fa280fe..3a2b1ec 100644 --- a/crates/cfr/src/lib.rs +++ b/crates/cfr/src/lib.rs @@ -20,6 +20,9 @@ extern crate alloc; mod conference; +#[cfg(feature = "std")] +pub mod persistence; + pub use conference::{Conference, Error, Joining, Message, Recipient, Result}; pub use cfr_core::{ diff --git a/crates/cfr/src/persistence/mod.rs b/crates/cfr/src/persistence/mod.rs new file mode 100644 index 0000000..1944d88 --- /dev/null +++ b/crates/cfr/src/persistence/mod.rs @@ -0,0 +1,574 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +//! Crash-safe, versioned persistence for [`crate::Conference`]. +//! +//! Every mutating operation uses a copy-on-write transaction: the candidate +//! state is validated, appended to the WAL, and synchronized before it becomes +//! the live in-memory state. Control messages leave this boundary only through +//! the durable outbox. + +mod state; +mod store; + +use crate::{ + Beacon, CheckpointCertificate, CheckpointSignature, Codec, Conference, Event, Joining, + KeyPackage, Policy, ProtocolProfile, Recipient, ResumptionRecord, SessionId, SigPublic, + Trailer, +}; +use state::LogicalState; +use std::collections::BTreeSet; +use std::path::Path; +use store::Store; + +/// Current logical schema version of a persisted conference state. +/// +/// This is independent of both the CFR wire protocol and the store envelope. +pub const CURRENT_PERSISTENCE_SCHEMA_VERSION: u32 = 1; + +pub(crate) const HARD_MAX_STATE_BYTES: usize = 4 * 1024 * 1024; +pub(crate) const HARD_MAX_WAL_BYTES: u64 = 64 * 1024 * 1024; +pub(crate) const MAX_WINDOW_ENTRIES: usize = 65_536; + +/// Identifies which internal version tag was unsupported. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum VersionKind { + /// The logical conference-state schema. + PersistenceSchema, + /// The snapshot or WAL envelope. + StoreEnvelope, +} + +/// A transport-supplied stable identifier for one inbound control message. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct InboundId([u8; 32]); + +impl InboundId { + /// Creates an identifier from transport-owned bytes. + pub const fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + /// Returns the transport-owned bytes. + pub const fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +impl From<[u8; 32]> for InboundId { + fn from(value: [u8; 32]) -> Self { + Self::from_bytes(value) + } +} + +/// A monotonically increasing durable outbox identifier. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct OutboundId(u64); + +impl OutboundId { + /// Returns the numeric identifier. + pub const fn get(self) -> u64 { + self.0 + } +} + +/// A deterministic transport retry key for one exact delivery. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DeliveryKey([u8; 32]); + +impl DeliveryKey { + /// Returns the key bytes. + pub const fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } +} + +/// One unacknowledged control message from the durable outbox. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingDelivery { + /// Monotonic outbox identifier. + pub id: OutboundId, + /// Deterministic retry key bound to the ID, recipient, and payload. + pub delivery_key: DeliveryKey, + /// Intended protocol recipient. + pub recipient: Recipient, + /// Exact CFR control-message bytes. + pub payload: Vec, +} + +/// Result of an inbound durable transaction. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct InboundResult { + /// True when the same transport ID and payload were committed previously. + pub duplicate: bool, + /// New protocol events; always empty for a duplicate. + pub events: Vec, + /// IDs added to the durable outbox by this transaction. + pub deliveries: Vec, +} + +/// Resource and compaction settings persisted with the conference. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PersistenceOptions { + /// Number of inbound IDs retained for durable idempotency. + pub inbound_window: usize, + /// Maximum number of unacknowledged control deliveries. + pub max_outbox_entries: usize, + /// Maximum encoded logical-state payload size. + pub max_state_bytes: usize, + /// Maximum payload size of one full-state WAL record. + pub max_record_bytes: usize, + /// Hard WAL file size limit. + pub max_wal_bytes: u64, + /// WAL size at which the current state is checkpointed before appending. + pub checkpoint_threshold: u64, +} + +impl Default for PersistenceOptions { + fn default() -> Self { + Self { + inbound_window: 4_096, + max_outbox_entries: 1_024, + max_state_bytes: HARD_MAX_STATE_BYTES, + max_record_bytes: HARD_MAX_STATE_BYTES, + max_wal_bytes: HARD_MAX_WAL_BYTES, + checkpoint_threshold: 32 * 1024 * 1024, + } + } +} + +impl PersistenceOptions { + pub(crate) fn validate(self) -> Result { + if self.inbound_window == 0 || self.inbound_window > MAX_WINDOW_ENTRIES { + return Err(Error::InvalidOptions("inbound window is out of range")); + } + if self.max_outbox_entries == 0 || self.max_outbox_entries > MAX_WINDOW_ENTRIES { + return Err(Error::InvalidOptions("outbox limit is out of range")); + } + if self.max_state_bytes == 0 || self.max_state_bytes > HARD_MAX_STATE_BYTES { + return Err(Error::InvalidOptions("state limit is out of range")); + } + if self.max_record_bytes == 0 || self.max_record_bytes > HARD_MAX_STATE_BYTES { + return Err(Error::InvalidOptions("record limit is out of range")); + } + if self.max_wal_bytes == 0 || self.max_wal_bytes > HARD_MAX_WAL_BYTES { + return Err(Error::InvalidOptions("WAL limit is out of range")); + } + if self.checkpoint_threshold == 0 || self.checkpoint_threshold > self.max_wal_bytes { + return Err(Error::InvalidOptions( + "checkpoint threshold exceeds WAL limit", + )); + } + Ok(self) + } +} + +/// Errors from the versioned persistence boundary. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The requested state directory does not exist. + #[error("persistent conference not found")] + NotFound, + /// Creation refused to reuse an existing path. + #[error("persistent conference already exists")] + AlreadyExists, + /// Another process or handle owns the state directory lock. + #[error("persistent conference is locked")] + Locked, + /// A complete persisted object was malformed or failed validation. + #[error("persisted conference is corrupt: {0}")] + Corrupt(&'static str), + /// A persisted internal schema or envelope version is unknown. + #[error("unsupported {kind:?} version {found}")] + UnsupportedVersion { + /// Version namespace that was rejected. + kind: VersionKind, + /// Unknown numeric version. + found: u32, + }, + /// An inbound ID was reused for different payload bytes. + #[error("inbound idempotency conflict for {id:?}")] + IdempotencyConflict { + /// Conflicting transport identifier. + id: InboundId, + }, + /// A configured durable resource bound would be exceeded. + #[error("persistence resource limit exceeded: {0}")] + LimitExceeded(&'static str), + /// Persistence options were internally inconsistent or out of range. + #[error("invalid persistence options: {0}")] + InvalidOptions(&'static str), + /// A filesystem operation failed. + #[error("persistence I/O failed: {0}")] + Io(#[from] std::io::Error), + /// The requested conference operation failed before commit. + #[error("conference operation failed: {0}")] + Protocol(#[from] crate::Error), +} + +/// Convenience alias for persistence operations. +pub type Result = std::result::Result; + +/// A conference whose complete mutable state is committed before it is exposed. +pub struct PersistentConference { + state: LogicalState, + store: Store, +} + +impl PersistentConference { + /// Creates a new founder conference with default persistence limits. + pub fn create(path: impl AsRef, policy: Policy) -> Result { + Self::create_with_options(path, policy, PersistenceOptions::default()) + } + + /// Creates a new founder conference with explicit persisted limits. + pub fn create_with_options( + path: impl AsRef, + policy: Policy, + options: PersistenceOptions, + ) -> Result { + let options = options.validate()?; + let (conference, messages) = Conference::create(policy)?; + Self::create_from_conference(path.as_ref(), conference, messages, options) + } + + /// Accepts a welcome into a newly created persistent state directory. + pub fn join(path: impl AsRef, joining: Joining, welcome: &[u8]) -> Result { + Self::join_with_options(path, joining, welcome, PersistenceOptions::default()) + } + + /// Accepts a welcome with explicit persisted limits. + pub fn join_with_options( + path: impl AsRef, + joining: Joining, + welcome: &[u8], + options: PersistenceOptions, + ) -> Result { + let options = options.validate()?; + let (conference, messages) = joining.accept(welcome)?; + Self::create_from_conference(path.as_ref(), conference, messages, options) + } + + fn create_from_conference( + path: &Path, + conference: Conference, + messages: Vec, + options: PersistenceOptions, + ) -> Result { + let state = LogicalState::new(conference, messages, options)?; + let payload = state.encode()?; + let store = Store::create(path, state.sequence, &payload)?; + Ok(Self { state, store }) + } + + /// Opens an existing state directory; absence never creates a new identity. + pub fn open(path: impl AsRef) -> Result { + let (store, recovery) = Store::open(path.as_ref())?; + let state = LogicalState::recover(recovery)?; + store.validate_runtime_limits(&state.options)?; + Ok(Self { state, store }) + } + + /// This participant's stable identity. + pub fn identity(&self) -> SigPublic { + self.state.conference.identity() + } + + /// The stable conference session identifier. + pub fn session_id(&self) -> SessionId { + self.state.conference.session_id() + } + + /// Current conference members. + pub fn members(&self) -> BTreeSet { + self.state.conference.members() + } + + /// Current key-version label. + pub fn version(&self) -> [u8; 8] { + self.state.conference.version() + } + + /// Monotonic committed persistence transaction sequence. + pub fn sequence(&self) -> u64 { + self.state.sequence + } + + /// Persisted resource and compaction settings. + pub fn options(&self) -> PersistenceOptions { + self.state.options + } + + /// Whether the current group key is locally derivable. + pub fn ready(&self) -> bool { + self.state.conference.ready() + } + + /// Whether current node-key repair is required. + pub fn needs_repair(&self) -> bool { + self.state.conference.needs_repair() + } + + /// Number of retained signed history operations. + pub fn history_len(&self) -> usize { + self.state.conference.history_len() + } + + /// Approximate retained protocol-state size. + pub fn state_bytes(&self) -> usize { + self.state.conference.state_bytes() + } + + /// Whether the active session should prepare signed reinitialization. + pub fn reinitialization_recommended(&self) -> bool { + self.state.conference.reinitialization_recommended() + } + + /// Prepares a non-mutating signed-session transition record. + pub fn prepare_checkpoint( + &self, + next_session: SessionId, + checkpoint_epoch: u64, + profile: ProtocolProfile, + ) -> Result { + Ok(self + .state + .conference + .prepare_checkpoint(next_session, checkpoint_epoch, profile)?) + } + + /// Signs a locally verified transition record without mutating conference state. + pub fn approve_checkpoint(&self, record: &ResumptionRecord) -> Result { + Ok(self.state.conference.approve_checkpoint(record)?) + } + + /// Durably queues a validated checkpoint offer. + pub fn offer_checkpoint( + &mut self, + certificate: &CheckpointCertificate, + ) -> Result> { + self.mutate_conference(|conference| { + let message = conference.offer_checkpoint(certificate)?; + Ok(((), vec![message])) + }) + .map(|((), deliveries)| deliveries) + } + + /// Returns a key-confirmation beacon without mutating state. + pub fn beacon(&self) -> [u8; cfr_core::BEACON_LEN] { + self.state.conference.beacon() + } + + /// Checks a peer key-confirmation beacon without mutating state. + pub fn check_beacon(&self, peer: &SigPublic, beacon: &[u8; cfr_core::BEACON_LEN]) -> Beacon { + self.state.conference.check_beacon(peer, beacon) + } + + /// Admits a newcomer and durably queues every resulting control message. + pub fn invite(&mut self, key_package: &KeyPackage) -> Result> { + self.mutate_conference(|conference| { + let messages = conference.invite(key_package)?; + Ok(((), messages)) + }) + .map(|((), deliveries)| deliveries) + } + + /// Evicts a participant and durably queues removal and rekey messages. + pub fn evict(&mut self, who: &SigPublic) -> Result> { + self.mutate_conference(|conference| { + let messages = conference.evict(who)?; + Ok(((), messages)) + }) + .map(|((), deliveries)| deliveries) + } + + /// Leaves and durably queues the signed removal message. + pub fn leave(&mut self) -> Result> { + self.mutate_conference(|conference| { + let message = conference.leave()?; + Ok(((), vec![message])) + }) + .map(|((), deliveries)| deliveries) + } + + /// Contributes fresh entropy and durably queues the resulting message. + pub fn rekey(&mut self) -> Result> { + self.mutate_conference(|conference| { + let messages = conference.rekey()?; + Ok(((), messages)) + }) + .map(|((), deliveries)| deliveries) + } + + /// Rotates prekeys, rekeys, and durably queues all resulting messages. + pub fn heal(&mut self) -> Result> { + self.mutate_conference(|conference| { + let messages = conference.heal()?; + Ok(((), messages)) + }) + .map(|((), deliveries)| deliveries) + } + + /// Durably advances the local prekey deadline clock. + pub fn tick(&mut self) -> Result<()> { + self.mutate_conference(|conference| { + conference.tick(); + Ok(((), Vec::new())) + }) + .map(|((), _)| ()) + } + + /// Atomically processes an inbound control payload and its transport ID. + pub fn handle_inbound(&mut self, id: InboundId, payload: &[u8]) -> Result { + let digest = state::inbound_digest(payload); + if let Some(committed) = self.state.inbound.get(&id) { + if committed == &digest { + return Ok(InboundResult { + duplicate: true, + events: Vec::new(), + deliveries: Vec::new(), + }); + } + return Err(Error::IdempotencyConflict { id }); + } + self.commit_candidate(|candidate| { + let (events, messages) = candidate.conference.handle(payload)?; + let deliveries = candidate.enqueue(messages)?; + candidate.record_inbound(id, digest); + Ok(InboundResult { + duplicate: false, + events, + deliveries, + }) + }) + } + + /// Durably queues anti-entropy and any required node-key repair requests. + pub fn resync(&mut self) -> Result> { + self.mutate_conference(|conference| Ok(((), conference.resync()))) + .map(|((), deliveries)| deliveries) + } + + /// Protects one media frame and commits its sender counter before return. + pub fn protect(&mut self, codec: Codec, frame: &[u8], keyframe: bool) -> Result> { + self.mutate_conference(|conference| { + Ok((conference.protect(codec, frame, keyframe)?, Vec::new())) + }) + .map(|(protected, _)| protected) + } + + /// Opens one media frame and commits ratchet/replay state before return. + pub fn open_media(&mut self, packet: &[u8]) -> Result<(SigPublic, Vec)> { + self.mutate_conference(|conference| Ok((conference.open(packet)?, Vec::new()))) + .map(|(opened, _)| opened) + } + + /// Reads media routing metadata without mutating state. + pub fn inspect(packet: &[u8]) -> Result { + Ok(Conference::inspect(packet)?) + } + + /// Returns all unacknowledged deliveries in monotonic ID order. + pub fn pending_deliveries(&self) -> Vec { + self.state.outbox.values().cloned().collect() + } + + /// Durably acknowledges one delivery; repeated acknowledgements return false. + pub fn acknowledge(&mut self, id: OutboundId) -> Result { + if !self.state.outbox.contains_key(&id) { + return Ok(false); + } + self.commit_candidate(|candidate| Ok(candidate.outbox.remove(&id).is_some())) + } + + /// Writes the current state as a snapshot and resets the WAL crash-safely. + pub fn checkpoint(&mut self) -> Result<()> { + let payload = self.state.encode()?; + self.store.checkpoint(self.state.sequence, &payload) + } + + fn mutate_conference( + &mut self, + operation: impl FnOnce(&mut Conference) -> crate::Result<(R, Vec)>, + ) -> Result<(R, Vec)> { + self.commit_candidate(|candidate| { + let (result, messages) = operation(&mut candidate.conference)?; + let deliveries = candidate.enqueue(messages)?; + Ok((result, deliveries)) + }) + } + + fn commit_candidate( + &mut self, + operation: impl FnOnce(&mut LogicalState) -> Result, + ) -> Result { + let current_payload = self.state.encode()?; + let mut candidate = LogicalState::decode(¤t_payload, self.state.sequence)?; + let result = operation(&mut candidate)?; + candidate.sequence = candidate + .sequence + .checked_add(1) + .ok_or(Error::LimitExceeded("transaction sequence exhausted"))?; + let candidate_payload = candidate.encode()?; + self.store.append( + self.state.sequence, + ¤t_payload, + candidate.sequence, + &candidate_payload, + candidate.options, + )?; + self.state = candidate; + Ok(result) + } +} + +#[cfg(test)] +mod tests { + use super::store::Fault; + use super::*; + use std::fs; + use std::path::PathBuf; + use std::sync::atomic::{AtomicU64, Ordering}; + + static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1); + + struct TestDirectory(PathBuf); + + impl TestDirectory { + fn new() -> Self { + let id = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + Self(std::env::temp_dir().join(format!( + "cfr-persistence-atomic-{}-{id}", + std::process::id() + ))) + } + } + + impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + #[test] + fn failed_sync_does_not_advance_live_or_recovered_state() { + let directory = TestDirectory::new(); + let mut conference = + PersistentConference::create(&directory.0, Policy::leaderless(2)).unwrap(); + let sequence = conference.sequence(); + let version = conference.version(); + conference.store.inject(Fault::BeforeSync); + assert!(matches!(conference.tick(), Err(Error::Io(_)))); + assert_eq!(conference.sequence(), sequence); + assert_eq!(conference.version(), version); + drop(conference); + + let reopened = PersistentConference::open(&directory.0).unwrap(); + assert_eq!(reopened.sequence(), sequence); + assert_eq!(reopened.version(), version); + } +} diff --git a/crates/cfr/src/persistence/state.rs b/crates/cfr/src/persistence/state.rs new file mode 100644 index 0000000..01e9401 --- /dev/null +++ b/crates/cfr/src/persistence/state.rs @@ -0,0 +1,401 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +use super::store::{Record, Recovery, SnapshotStatus}; +use super::{ + DeliveryKey, Error, InboundId, OutboundId, PendingDelivery, PersistenceOptions, Recipient, + Result, VersionKind, CURRENT_PERSISTENCE_SCHEMA_VERSION, HARD_MAX_STATE_BYTES, +}; +use crate::{Conference, Message}; +use cfr_core::codec::{Reader, Writer}; +use cfr_crypto::hash; +use std::collections::{BTreeMap, VecDeque}; + +type InboundDigests = BTreeMap; +type InboundOrder = VecDeque; + +pub(crate) struct LogicalState { + pub(crate) sequence: u64, + pub(crate) conference: Conference, + pub(crate) options: PersistenceOptions, + pub(crate) inbound: InboundDigests, + inbound_order: InboundOrder, + pub(crate) outbox: BTreeMap, + next_outbound_id: u64, +} + +pub(crate) fn inbound_digest(payload: &[u8]) -> [u8; 32] { + hash(b"cfr/persistence/inbound", &[payload]) +} + +fn delivery_key( + conference: &Conference, + id: OutboundId, + recipient: Recipient, + payload: &[u8], +) -> DeliveryKey { + let id_bytes = id.0.to_be_bytes(); + let everyone = [0u8]; + let peer = [1u8]; + let digest = match recipient { + Recipient::Everyone => hash( + b"cfr/persistence/delivery", + &[ + &conference.session_id(), + conference.identity().as_bytes(), + &id_bytes, + &everyone, + payload, + ], + ), + Recipient::Peer(identity) => hash( + b"cfr/persistence/delivery", + &[ + &conference.session_id(), + conference.identity().as_bytes(), + &id_bytes, + &peer, + identity.as_bytes(), + payload, + ], + ), + }; + DeliveryKey(digest) +} + +fn usize_to_u64(value: usize) -> Result { + u64::try_from(value).map_err(|_| Error::LimitExceeded("integer exceeds persisted width")) +} + +fn read_usize(reader: &mut Reader<'_>) -> Result { + usize::try_from(reader.u64().map_err(corrupt_codec)?) + .map_err(|_| Error::Corrupt("integer exceeds platform width")) +} + +fn corrupt_codec(_: cfr_core::Error) -> Error { + Error::Corrupt("logical state encoding is malformed") +} + +fn read_options(reader: &mut Reader<'_>) -> Result { + let options = PersistenceOptions { + inbound_window: read_usize(reader)?, + max_outbox_entries: read_usize(reader)?, + max_state_bytes: read_usize(reader)?, + max_record_bytes: read_usize(reader)?, + max_wal_bytes: reader.u64().map_err(corrupt_codec)?, + checkpoint_threshold: reader.u64().map_err(corrupt_codec)?, + }; + options + .validate() + .map_err(|_| Error::Corrupt("persisted resource limits are invalid")) +} + +fn write_options(writer: &mut Writer, options: PersistenceOptions) -> Result<()> { + writer + .u64(usize_to_u64(options.inbound_window)?) + .u64(usize_to_u64(options.max_outbox_entries)?) + .u64(usize_to_u64(options.max_state_bytes)?) + .u64(usize_to_u64(options.max_record_bytes)?) + .u64(options.max_wal_bytes) + .u64(options.checkpoint_threshold); + Ok(()) +} + +fn read_recipient(reader: &mut Reader<'_>) -> Result { + match reader.u32().map_err(corrupt_codec)? { + 0 => Ok(Recipient::Everyone), + 1 => Ok(Recipient::Peer(crate::SigPublic::from_bytes( + reader.array::<32>().map_err(corrupt_codec)?, + ))), + _ => Err(Error::Corrupt("outbox recipient tag is invalid")), + } +} + +fn write_recipient(writer: &mut Writer, recipient: Recipient) { + match recipient { + Recipient::Everyone => { + writer.u32(0); + } + Recipient::Peer(identity) => { + writer.u32(1).bytes(identity.as_bytes()); + } + } +} + +impl LogicalState { + pub(crate) fn new( + conference: Conference, + messages: Vec, + options: PersistenceOptions, + ) -> Result { + let mut state = Self { + sequence: 1, + conference, + options, + inbound: BTreeMap::new(), + inbound_order: VecDeque::new(), + outbox: BTreeMap::new(), + next_outbound_id: 1, + }; + state.enqueue(messages)?; + Ok(state) + } + + pub(crate) fn enqueue(&mut self, messages: Vec) -> Result> { + let available = self + .options + .max_outbox_entries + .checked_sub(self.outbox.len()) + .ok_or(Error::Corrupt("outbox exceeds its persisted limit"))?; + if messages.len() > available { + return Err(Error::LimitExceeded("durable outbox is full")); + } + let count = u64::try_from(messages.len()) + .map_err(|_| Error::LimitExceeded("too many outbound messages"))?; + self.next_outbound_id + .checked_add(count) + .ok_or(Error::LimitExceeded("outbound identifier exhausted"))?; + + let mut ids = Vec::with_capacity(messages.len()); + for message in messages { + let id = OutboundId(self.next_outbound_id); + self.next_outbound_id += 1; + let delivery = PendingDelivery { + id, + delivery_key: delivery_key(&self.conference, id, message.to, &message.payload), + recipient: message.to, + payload: message.payload, + }; + self.outbox.insert(id, delivery); + ids.push(id); + } + Ok(ids) + } + + pub(crate) fn record_inbound(&mut self, id: InboundId, digest: [u8; 32]) { + self.inbound.insert(id, digest); + self.inbound_order.push_back(id); + while self.inbound_order.len() > self.options.inbound_window { + if let Some(expired) = self.inbound_order.pop_front() { + self.inbound.remove(&expired); + } + } + } + + pub(crate) fn encode(&self) -> Result> { + let conference = self.conference.export_persistence_state()?; + let inbound: Vec<(InboundId, [u8; 32])> = self + .inbound_order + .iter() + .map(|id| { + self.inbound + .get(id) + .copied() + .map(|digest| (*id, digest)) + .ok_or(Error::Corrupt("inbound eviction order is inconsistent")) + }) + .collect::>()?; + if inbound.len() != self.inbound.len() { + return Err(Error::Corrupt("inbound eviction order is inconsistent")); + } + + let mut writer = Writer::new(); + writer + .u32(CURRENT_PERSISTENCE_SCHEMA_VERSION) + .u64(self.sequence); + write_options(&mut writer, self.options)?; + writer.bytes(&conference); + writer.list(&inbound, |writer, (id, digest)| { + writer.bytes(id.as_bytes()).bytes(digest); + }); + let outbox: Vec<_> = self.outbox.values().collect(); + writer.list(&outbox, |writer, delivery| { + writer + .u64(delivery.id.0) + .bytes(delivery.delivery_key.as_bytes()); + write_recipient(writer, delivery.recipient); + writer.bytes(&delivery.payload); + }); + writer.u64(self.next_outbound_id); + let bytes = writer.finish(); + if bytes.len() > self.options.max_state_bytes || bytes.len() > HARD_MAX_STATE_BYTES { + return Err(Error::LimitExceeded( + "logical state exceeds configured limit", + )); + } + if bytes.len() > self.options.max_record_bytes { + return Err(Error::LimitExceeded("WAL record exceeds configured limit")); + } + Ok(bytes) + } + + pub(crate) fn decode(bytes: &[u8], envelope_sequence: u64) -> Result { + if bytes.len() > HARD_MAX_STATE_BYTES { + return Err(Error::Corrupt("logical state exceeds hard limit")); + } + let mut reader = Reader::new(bytes); + let schema = reader.u32().map_err(corrupt_codec)?; + if schema != CURRENT_PERSISTENCE_SCHEMA_VERSION { + return Err(Error::UnsupportedVersion { + kind: VersionKind::PersistenceSchema, + found: schema, + }); + } + let sequence = reader.u64().map_err(corrupt_codec)?; + if sequence == 0 || sequence != envelope_sequence { + return Err(Error::Corrupt("logical and envelope sequences differ")); + } + let options = read_options(&mut reader)?; + if bytes.len() > options.max_state_bytes || bytes.len() > options.max_record_bytes { + return Err(Error::Corrupt("logical state exceeds persisted limits")); + } + let conference_bytes = reader.bytes().map_err(corrupt_codec)?; + let conference = Conference::import_persistence_state(conference_bytes) + .map_err(|_| Error::Corrupt("conference state failed validation"))?; + let (inbound, inbound_order) = Self::read_inbound(&mut reader, options)?; + let outbox = Self::read_outbox(&mut reader, &conference, options)?; + let next_outbound_id = reader.u64().map_err(corrupt_codec)?; + reader.finish().map_err(corrupt_codec)?; + if next_outbound_id == 0 + || outbox + .keys() + .next_back() + .is_some_and(|last| last.0 >= next_outbound_id) + { + return Err(Error::Corrupt("next outbound identifier is invalid")); + } + let state = Self { + sequence, + conference, + options, + inbound, + inbound_order, + outbox, + next_outbound_id, + }; + let canonical = state + .encode() + .map_err(|_| Error::Corrupt("logical state is not re-encodable"))?; + if canonical != bytes { + return Err(Error::Corrupt("logical state encoding is non-canonical")); + } + Ok(state) + } + + fn read_inbound( + reader: &mut Reader<'_>, + options: PersistenceOptions, + ) -> Result<(InboundDigests, InboundOrder)> { + let entries: Vec<(InboundId, [u8; 32])> = reader + .list(|reader| Ok((InboundId(reader.array::<32>()?), reader.array::<32>()?))) + .map_err(corrupt_codec)?; + if entries.len() > options.inbound_window { + return Err(Error::Corrupt("inbound window exceeds persisted limit")); + } + let mut inbound = BTreeMap::new(); + let mut order = VecDeque::with_capacity(entries.len()); + for (id, digest) in entries { + if inbound.insert(id, digest).is_some() { + return Err(Error::Corrupt("inbound window contains a duplicate ID")); + } + order.push_back(id); + } + Ok((inbound, order)) + } + + fn read_outbox( + reader: &mut Reader<'_>, + conference: &Conference, + options: PersistenceOptions, + ) -> Result> { + let entries: Vec = reader + .list(|reader| { + Ok(PendingDelivery { + id: OutboundId(reader.u64()?), + delivery_key: DeliveryKey(reader.array::<32>()?), + recipient: read_recipient(reader) + .map_err(|_| cfr_core::Error::Encoding("invalid persisted recipient"))?, + payload: reader.bytes()?.to_vec(), + }) + }) + .map_err(corrupt_codec)?; + if entries.len() > options.max_outbox_entries { + return Err(Error::Corrupt("outbox exceeds persisted limit")); + } + let mut outbox = BTreeMap::new(); + let mut previous = None; + for delivery in entries { + if delivery.id.0 == 0 + || previous.is_some_and(|id: OutboundId| id >= delivery.id) + || delivery.delivery_key + != delivery_key( + conference, + delivery.id, + delivery.recipient, + &delivery.payload, + ) + { + return Err(Error::Corrupt("outbox delivery binding is invalid")); + } + previous = Some(delivery.id); + outbox.insert(delivery.id, delivery); + } + Ok(outbox) + } + + pub(crate) fn recover(recovery: Recovery) -> Result { + let snapshot = match recovery.snapshot { + SnapshotStatus::Valid(record) => match Self::decode(&record.payload, record.sequence) { + Ok(state) => Some((state, record)), + Err(Error::UnsupportedVersion { kind, found }) => { + return Err(Error::UnsupportedVersion { kind, found }); + } + Err(_) => None, + }, + SnapshotStatus::Missing | SnapshotStatus::Corrupt => None, + }; + + let mut wal_states = Vec::with_capacity(recovery.wal.len()); + for record in recovery.wal { + wal_states.push((Self::decode(&record.payload, record.sequence)?, record)); + } + Self::select_recovered(snapshot, wal_states) + } + + fn select_recovered( + snapshot: Option<(Self, Record)>, + wal: Vec<(Self, Record)>, + ) -> Result { + let Some((mut selected, snapshot_record)) = snapshot else { + return wal + .into_iter() + .last() + .map(|(state, _)| state) + .ok_or(Error::Corrupt("no valid complete persisted state")); + }; + let mut selected_sequence = selected.sequence; + for (state, record) in wal { + if state.sequence < selected_sequence { + continue; + } + if state.sequence == selected_sequence { + if state.sequence == snapshot_record.sequence + && record.payload != snapshot_record.payload + { + return Err(Error::Corrupt("snapshot and WAL disagree at one sequence")); + } + continue; + } + if state.sequence != selected_sequence + 1 { + return Err(Error::Corrupt("WAL sequence has a gap after snapshot")); + } + selected = state; + selected_sequence = selected.sequence; + } + Ok(selected) + } +} diff --git a/crates/cfr/src/persistence/store.rs b/crates/cfr/src/persistence/store.rs new file mode 100644 index 0000000..3afc830 --- /dev/null +++ b/crates/cfr/src/persistence/store.rs @@ -0,0 +1,680 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +use super::{ + Error, PersistenceOptions, Result, VersionKind, HARD_MAX_STATE_BYTES, HARD_MAX_WAL_BYTES, +}; +use cfr_crypto::{ct_eq, hash}; +use fs2::FileExt; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +pub(crate) const STORE_FORMAT_VERSION: u32 = 1; + +const SNAPSHOT_FILE: &str = "snapshot"; +const SNAPSHOT_TEMP: &str = "snapshot.tmp"; +const WAL_FILE: &str = "wal"; +const WAL_TEMP: &str = "wal.tmp"; +const LOCK_FILE: &str = "lock"; + +const SNAPSHOT_MAGIC: [u8; 8] = *b"CFRSNAP\0"; +const WAL_MAGIC: [u8; 8] = *b"CFRWAL\0\0"; +const RECORD_MAGIC: [u8; 8] = *b"CFRREC\0\0"; +const COMMIT_MARKER: [u8; 8] = *b"CFRCMIT\0"; + +const WAL_HEADER_LEN: usize = 12; +const RECORD_PREFIX_LEN: usize = 8 + 8 + 8 + 32; +const RECORD_OVERHEAD: usize = RECORD_PREFIX_LEN + COMMIT_MARKER.len(); +const SNAPSHOT_PREFIX_LEN: usize = 8 + 4 + 8 + 8 + 32; +const SNAPSHOT_OVERHEAD: usize = SNAPSHOT_PREFIX_LEN + COMMIT_MARKER.len(); + +#[derive(Debug)] +pub(crate) struct Record { + pub(crate) sequence: u64, + pub(crate) payload: Vec, +} + +#[derive(Debug)] +pub(crate) enum SnapshotStatus { + Valid(Record), + Missing, + Corrupt, +} + +#[derive(Debug)] +pub(crate) struct Recovery { + pub(crate) snapshot: SnapshotStatus, + pub(crate) wal: Vec, +} + +pub(crate) struct Store { + directory: PathBuf, + _lock: File, + wal: File, + #[cfg(test)] + fault: Option, +} + +#[cfg(test)] +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum Fault { + BeforeWrite, + BeforeSync, +} + +fn snapshot_checksum(sequence: u64, payload: &[u8]) -> [u8; 32] { + hash( + b"cfr/persistence/snapshot-checksum", + &[&sequence.to_be_bytes(), payload], + ) +} + +fn record_checksum(sequence: u64, payload: &[u8]) -> [u8; 32] { + hash( + b"cfr/persistence/wal-checksum", + &[&sequence.to_be_bytes(), payload], + ) +} + +fn encode_snapshot(sequence: u64, payload: &[u8]) -> Result> { + let length = u64::try_from(payload.len()) + .map_err(|_| Error::LimitExceeded("snapshot payload length exceeds u64"))?; + let capacity = payload + .len() + .checked_add(SNAPSHOT_OVERHEAD) + .ok_or(Error::LimitExceeded("snapshot size overflow"))?; + let mut bytes = Vec::with_capacity(capacity); + bytes.extend_from_slice(&SNAPSHOT_MAGIC); + bytes.extend_from_slice(&STORE_FORMAT_VERSION.to_be_bytes()); + bytes.extend_from_slice(&sequence.to_be_bytes()); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.extend_from_slice(&snapshot_checksum(sequence, payload)); + bytes.extend_from_slice(payload); + bytes.extend_from_slice(&COMMIT_MARKER); + Ok(bytes) +} + +fn encode_record(sequence: u64, payload: &[u8]) -> Result> { + let length = u64::try_from(payload.len()) + .map_err(|_| Error::LimitExceeded("WAL payload length exceeds u64"))?; + let capacity = payload + .len() + .checked_add(RECORD_OVERHEAD) + .ok_or(Error::LimitExceeded("WAL record size overflow"))?; + let mut bytes = Vec::with_capacity(capacity); + bytes.extend_from_slice(&RECORD_MAGIC); + bytes.extend_from_slice(&sequence.to_be_bytes()); + bytes.extend_from_slice(&length.to_be_bytes()); + bytes.extend_from_slice(&record_checksum(sequence, payload)); + bytes.extend_from_slice(payload); + bytes.extend_from_slice(&COMMIT_MARKER); + Ok(bytes) +} + +fn wal_header() -> [u8; WAL_HEADER_LEN] { + let mut header = [0u8; WAL_HEADER_LEN]; + header[..8].copy_from_slice(&WAL_MAGIC); + header[8..].copy_from_slice(&STORE_FORMAT_VERSION.to_be_bytes()); + header +} + +fn be_u32(bytes: &[u8]) -> Option { + Some(u32::from_be_bytes(bytes.try_into().ok()?)) +} + +fn be_u64(bytes: &[u8]) -> Option { + Some(u64::from_be_bytes(bytes.try_into().ok()?)) +} + +fn parse_snapshot(bytes: &[u8]) -> Result { + if bytes.len() < SNAPSHOT_PREFIX_LEN || bytes.get(..8) != Some(&SNAPSHOT_MAGIC) { + return Ok(SnapshotStatus::Corrupt); + } + let version = be_u32(&bytes[8..12]).ok_or(Error::Corrupt("snapshot version is truncated"))?; + if version != STORE_FORMAT_VERSION { + return Err(Error::UnsupportedVersion { + kind: VersionKind::StoreEnvelope, + found: version, + }); + } + let Some(sequence) = be_u64(&bytes[12..20]) else { + return Ok(SnapshotStatus::Corrupt); + }; + let Some(length_u64) = be_u64(&bytes[20..28]) else { + return Ok(SnapshotStatus::Corrupt); + }; + let Ok(length) = usize::try_from(length_u64) else { + return Ok(SnapshotStatus::Corrupt); + }; + if sequence == 0 || length > HARD_MAX_STATE_BYTES { + return Ok(SnapshotStatus::Corrupt); + } + let Some(expected) = SNAPSHOT_OVERHEAD.checked_add(length) else { + return Ok(SnapshotStatus::Corrupt); + }; + if bytes.len() != expected || bytes[expected - 8..] != COMMIT_MARKER { + return Ok(SnapshotStatus::Corrupt); + } + let payload = &bytes[SNAPSHOT_PREFIX_LEN..SNAPSHOT_PREFIX_LEN + length]; + let checksum = &bytes[28..60]; + if !ct_eq(checksum, &snapshot_checksum(sequence, payload)) { + return Ok(SnapshotStatus::Corrupt); + } + Ok(SnapshotStatus::Valid(Record { + sequence, + payload: payload.to_vec(), + })) +} + +fn validate_wal_header(bytes: &[u8]) -> Result<()> { + if bytes.len() < WAL_HEADER_LEN || bytes.get(..8) != Some(&WAL_MAGIC) { + return Err(Error::Corrupt("WAL header is malformed")); + } + let version = be_u32(&bytes[8..12]).ok_or(Error::Corrupt("WAL version is truncated"))?; + if version != STORE_FORMAT_VERSION { + return Err(Error::UnsupportedVersion { + kind: VersionKind::StoreEnvelope, + found: version, + }); + } + Ok(()) +} + +fn parse_record(bytes: &[u8], start: usize) -> Result> { + let remaining = bytes.len() - start; + if remaining < RECORD_PREFIX_LEN { + return Ok(None); + } + if bytes[start..start + 8] != RECORD_MAGIC { + return Err(Error::Corrupt("WAL record magic is invalid")); + } + let sequence = + be_u64(&bytes[start + 8..start + 16]).ok_or(Error::Corrupt("WAL sequence is truncated"))?; + let length_u64 = + be_u64(&bytes[start + 16..start + 24]).ok_or(Error::Corrupt("WAL length is truncated"))?; + let length = usize::try_from(length_u64) + .map_err(|_| Error::Corrupt("WAL record length exceeds platform width"))?; + if sequence == 0 || length > HARD_MAX_STATE_BYTES { + return Err(Error::Corrupt("WAL record bounds are invalid")); + } + let total = RECORD_OVERHEAD + .checked_add(length) + .ok_or(Error::Corrupt("WAL record length overflows"))?; + if remaining < total { + return Ok(None); + } + let payload_start = start + RECORD_PREFIX_LEN; + let payload_end = payload_start + length; + if bytes[payload_end..payload_end + 8] != COMMIT_MARKER { + return Err(Error::Corrupt("WAL commit marker is invalid")); + } + let payload = &bytes[payload_start..payload_end]; + let checksum = &bytes[start + 24..start + 56]; + if !ct_eq(checksum, &record_checksum(sequence, payload)) { + return Err(Error::Corrupt("WAL record checksum is invalid")); + } + Ok(Some(( + Record { + sequence, + payload: payload.to_vec(), + }, + start + total, + ))) +} + +fn scan_wal(bytes: &[u8]) -> Result<(Vec, usize)> { + validate_wal_header(bytes)?; + let mut records = Vec::new(); + let mut position = WAL_HEADER_LEN; + let mut previous: Option = None; + while position < bytes.len() { + let Some((record, next)) = parse_record(bytes, position)? else { + break; + }; + if let Some(previous_sequence) = previous { + let expected = previous_sequence + .checked_add(1) + .ok_or(Error::Corrupt("WAL sequence overflowed"))?; + if record.sequence != expected { + return Err(Error::Corrupt("WAL sequence is not consecutive")); + } + } + previous = Some(record.sequence); + records.push(record); + position = next; + } + Ok((records, position)) +} + +#[cfg(unix)] +fn secure_directory_builder() -> fs::DirBuilder { + use std::os::unix::fs::DirBuilderExt; + let mut builder = fs::DirBuilder::new(); + builder.mode(0o700); + builder +} + +#[cfg(not(unix))] +fn secure_directory_builder() -> fs::DirBuilder { + fs::DirBuilder::new() +} + +fn secure_open_options() -> OpenOptions { + let mut options = OpenOptions::new(); + options.read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + options +} + +#[cfg(unix)] +fn validate_mode(path: &Path, directory: bool) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + let metadata = fs::symlink_metadata(path)?; + if metadata.file_type().is_symlink() + || (directory && !metadata.is_dir()) + || (!directory && !metadata.is_file()) + || metadata.permissions().mode() & 0o077 != 0 + { + return Err(Error::Corrupt("state path type or permissions are unsafe")); + } + Ok(()) +} + +#[cfg(not(unix))] +fn validate_mode(path: &Path, directory: bool) -> Result<()> { + let metadata = fs::metadata(path)?; + if (directory && !metadata.is_dir()) || (!directory && !metadata.is_file()) { + return Err(Error::Corrupt("state path has the wrong type")); + } + Ok(()) +} + +fn sync_directory(path: &Path) -> Result<()> { + #[cfg(unix)] + File::open(path)?.sync_all()?; + Ok(()) +} + +fn write_atomic(directory: &Path, temporary: &str, committed: &str, bytes: &[u8]) -> Result<()> { + let temporary_path = directory.join(temporary); + let committed_path = directory.join(committed); + let mut options = secure_open_options(); + options.create(true).truncate(true); + let mut file = options.open(&temporary_path)?; + file.write_all(bytes)?; + file.sync_all()?; + drop(file); + fs::rename(&temporary_path, &committed_path)?; + sync_directory(directory) +} + +fn read_bounded(path: &Path, limit: u64) -> Result> { + validate_mode(path, false)?; + let metadata = fs::metadata(path)?; + if metadata.len() > limit { + return Err(Error::Corrupt("persisted file exceeds hard limit")); + } + let capacity = usize::try_from(metadata.len()) + .map_err(|_| Error::Corrupt("persisted file length exceeds platform width"))?; + let mut bytes = Vec::with_capacity(capacity); + File::open(path)? + .take(limit.saturating_add(1)) + .read_to_end(&mut bytes)?; + if u64::try_from(bytes.len()) + .map_err(|_| Error::Corrupt("persisted file length exceeds u64"))? + > limit + { + return Err(Error::Corrupt("persisted file exceeds hard limit")); + } + Ok(bytes) +} + +fn acquire_lock(directory: &Path) -> Result { + let path = directory.join(LOCK_FILE); + let mut options = secure_open_options(); + options.create(true); + let file = options.open(&path)?; + validate_mode(&path, false)?; + match FileExt::try_lock_exclusive(&file) { + Ok(()) => Ok(file), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => Err(Error::Locked), + Err(error) => Err(Error::Io(error)), + } +} + +impl Store { + pub(crate) fn create(path: &Path, sequence: u64, payload: &[u8]) -> Result { + match fs::symlink_metadata(path) { + Ok(_) => return Err(Error::AlreadyExists), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(Error::Io(error)), + } + secure_directory_builder().create(path)?; + validate_mode(path, true)?; + if let Some(parent) = path.parent() { + sync_directory(if parent.as_os_str().is_empty() { + Path::new(".") + } else { + parent + })?; + } + let lock = acquire_lock(path)?; + let snapshot = encode_snapshot(sequence, payload)?; + write_atomic(path, SNAPSHOT_TEMP, SNAPSHOT_FILE, &snapshot)?; + write_atomic(path, WAL_TEMP, WAL_FILE, &wal_header())?; + let wal_path = path.join(WAL_FILE); + let wal = secure_open_options().open(&wal_path)?; + Ok(Self { + directory: path.to_path_buf(), + _lock: lock, + wal, + #[cfg(test)] + fault: None, + }) + } + + pub(crate) fn open(path: &Path) -> Result<(Self, Recovery)> { + match fs::symlink_metadata(path) { + Ok(_) => validate_mode(path, true)?, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(Error::NotFound); + } + Err(error) => return Err(Error::Io(error)), + } + let lock = acquire_lock(path)?; + let wal_path = path.join(WAL_FILE); + let wal_bytes = match read_bounded(&wal_path, HARD_MAX_WAL_BYTES) { + Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(Error::Corrupt("WAL file is missing")); + } + result => result?, + }; + let (records, valid_length) = scan_wal(&wal_bytes)?; + let wal = secure_open_options().open(&wal_path)?; + if valid_length != wal_bytes.len() { + wal.set_len( + u64::try_from(valid_length) + .map_err(|_| Error::Corrupt("valid WAL length exceeds u64"))?, + )?; + wal.sync_data()?; + } + let snapshot_path = path.join(SNAPSHOT_FILE); + let snapshot = match read_bounded( + &snapshot_path, + u64::try_from(HARD_MAX_STATE_BYTES + SNAPSHOT_OVERHEAD) + .map_err(|_| Error::Corrupt("snapshot hard limit exceeds u64"))?, + ) { + Ok(bytes) => parse_snapshot(&bytes)?, + Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => { + SnapshotStatus::Missing + } + Err(Error::Corrupt(_)) => SnapshotStatus::Corrupt, + Err(error) => return Err(error), + }; + Ok(( + Self { + directory: path.to_path_buf(), + _lock: lock, + wal, + #[cfg(test)] + fault: None, + }, + Recovery { + snapshot, + wal: records, + }, + )) + } + + pub(crate) fn validate_runtime_limits(&self, options: &PersistenceOptions) -> Result<()> { + if self.wal.metadata()?.len() > options.max_wal_bytes { + return Err(Error::Corrupt("WAL exceeds its persisted limit")); + } + Ok(()) + } + + pub(crate) fn append( + &mut self, + current_sequence: u64, + current_payload: &[u8], + candidate_sequence: u64, + candidate_payload: &[u8], + options: PersistenceOptions, + ) -> Result<()> { + if current_sequence.checked_add(1) != Some(candidate_sequence) { + return Err(Error::Corrupt("candidate sequence is not consecutive")); + } + if candidate_payload.len() > options.max_record_bytes + || candidate_payload.len() > HARD_MAX_STATE_BYTES + { + return Err(Error::LimitExceeded("WAL record exceeds configured limit")); + } + let record = encode_record(candidate_sequence, candidate_payload)?; + let mut wal_length = self.wal.metadata()?.len(); + let record_length = u64::try_from(record.len()) + .map_err(|_| Error::LimitExceeded("WAL record length exceeds u64"))?; + let projected = wal_length + .checked_add(record_length) + .ok_or(Error::LimitExceeded("WAL length overflow"))?; + if projected > options.checkpoint_threshold || projected > options.max_wal_bytes { + self.checkpoint(current_sequence, current_payload)?; + wal_length = u64::try_from(WAL_HEADER_LEN) + .map_err(|_| Error::LimitExceeded("WAL header length exceeds u64"))?; + } + let projected = wal_length + .checked_add(record_length) + .ok_or(Error::LimitExceeded("WAL length overflow"))?; + if projected > options.max_wal_bytes { + return Err(Error::LimitExceeded( + "one WAL record cannot fit configured limit", + )); + } + self.append_durable(wal_length, &record) + } + + pub(crate) fn checkpoint(&mut self, sequence: u64, payload: &[u8]) -> Result<()> { + if payload.len() > HARD_MAX_STATE_BYTES { + return Err(Error::LimitExceeded("snapshot exceeds hard limit")); + } + let snapshot = encode_snapshot(sequence, payload)?; + write_atomic(&self.directory, SNAPSHOT_TEMP, SNAPSHOT_FILE, &snapshot)?; + write_atomic(&self.directory, WAL_TEMP, WAL_FILE, &wal_header())?; + self.wal = secure_open_options().open(self.directory.join(WAL_FILE))?; + Ok(()) + } + + fn append_durable(&mut self, original_length: u64, record: &[u8]) -> Result<()> { + #[cfg(test)] + if self.take_fault(Fault::BeforeWrite) { + return Err(Error::Io(std::io::Error::other("injected write failure"))); + } + self.wal.seek(SeekFrom::End(0))?; + if let Err(error) = self.wal.write_all(record) { + return self.rollback_append(original_length, error); + } + #[cfg(test)] + if self.take_fault(Fault::BeforeSync) { + return self.rollback_append( + original_length, + std::io::Error::other("injected sync failure"), + ); + } + if let Err(error) = self.wal.sync_data() { + return self.rollback_append(original_length, error); + } + Ok(()) + } + + fn rollback_append(&mut self, original_length: u64, original: std::io::Error) -> Result<()> { + self.wal.set_len(original_length)?; + self.wal.sync_data()?; + self.wal.seek(SeekFrom::End(0))?; + Err(Error::Io(original)) + } + + #[cfg(test)] + pub(super) fn inject(&mut self, fault: Fault) { + self.fault = Some(fault); + } + + #[cfg(test)] + fn take_fault(&mut self, expected: Fault) -> bool { + if self.fault == Some(expected) { + self.fault = None; + true + } else { + false + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicU64, Ordering}; + + static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1); + + struct TestDirectory(PathBuf); + + impl TestDirectory { + fn new() -> Self { + let id = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + Self( + std::env::temp_dir() + .join(format!("cfr-persistence-store-{}-{id}", std::process::id())), + ) + } + } + + impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + fn options() -> PersistenceOptions { + PersistenceOptions::default() + } + + #[test] + fn snapshot_and_wal_recover_latest_record() { + let directory = TestDirectory::new(); + let mut store = Store::create(&directory.0, 1, b"one").unwrap(); + store.append(1, b"one", 2, b"two", options()).unwrap(); + drop(store); + + let (_store, recovery) = Store::open(&directory.0).unwrap(); + assert!(matches!(recovery.snapshot, SnapshotStatus::Valid(_))); + assert_eq!(recovery.wal.len(), 1); + assert_eq!(recovery.wal[0].sequence, 2); + assert_eq!(recovery.wal[0].payload, b"two"); + } + + #[test] + fn incomplete_final_tail_is_truncated() { + let directory = TestDirectory::new(); + let mut store = Store::create(&directory.0, 1, b"one").unwrap(); + store.append(1, b"one", 2, b"two", options()).unwrap(); + drop(store); + let wal_path = directory.0.join(WAL_FILE); + let valid_length = fs::metadata(&wal_path).unwrap().len(); + let mut wal = OpenOptions::new().append(true).open(&wal_path).unwrap(); + wal.write_all(&RECORD_MAGIC[..5]).unwrap(); + wal.sync_all().unwrap(); + drop(wal); + + let (_store, recovery) = Store::open(&directory.0).unwrap(); + assert_eq!(recovery.wal.len(), 1); + assert_eq!(fs::metadata(wal_path).unwrap().len(), valid_length); + } + + #[test] + fn complete_bad_checksum_fails_closed() { + let directory = TestDirectory::new(); + let mut store = Store::create(&directory.0, 1, b"one").unwrap(); + store.append(1, b"one", 2, b"two", options()).unwrap(); + drop(store); + let path = directory.0.join(WAL_FILE); + let mut bytes = fs::read(&path).unwrap(); + bytes[WAL_HEADER_LEN + 24] ^= 1; + fs::write(&path, bytes).unwrap(); + assert!(matches!(Store::open(&directory.0), Err(Error::Corrupt(_)))); + } + + #[test] + fn corrupt_snapshot_is_reported_alongside_valid_wal() { + let directory = TestDirectory::new(); + let mut store = Store::create(&directory.0, 1, b"one").unwrap(); + store.append(1, b"one", 2, b"two", options()).unwrap(); + drop(store); + let path = directory.0.join(SNAPSHOT_FILE); + let mut bytes = fs::read(&path).unwrap(); + bytes[28] ^= 1; + fs::write(path, bytes).unwrap(); + + let (_store, recovery) = Store::open(&directory.0).unwrap(); + assert!(matches!(recovery.snapshot, SnapshotStatus::Corrupt)); + assert_eq!(recovery.wal.len(), 1); + } + + #[test] + fn lock_is_exclusive_until_store_drop() { + let directory = TestDirectory::new(); + let first = Store::create(&directory.0, 1, b"one").unwrap(); + assert!(matches!(Store::open(&directory.0), Err(Error::Locked))); + drop(first); + assert!(Store::open(&directory.0).is_ok()); + } + + #[test] + fn injected_write_and_sync_failures_leave_previous_state() { + for fault in [Fault::BeforeWrite, Fault::BeforeSync] { + let directory = TestDirectory::new(); + let mut store = Store::create(&directory.0, 1, b"one").unwrap(); + store.inject(fault); + assert!(matches!( + store.append(1, b"one", 2, b"two", options()), + Err(Error::Io(_)) + )); + drop(store); + let (_store, recovery) = Store::open(&directory.0).unwrap(); + assert!(recovery.wal.is_empty()); + assert!(matches!( + recovery.snapshot, + SnapshotStatus::Valid(Record { sequence: 1, .. }) + )); + } + } + + #[cfg(unix)] + #[test] + fn created_paths_are_owner_only() { + use std::os::unix::fs::PermissionsExt; + let directory = TestDirectory::new(); + let _store = Store::create(&directory.0, 1, b"one").unwrap(); + assert_eq!( + fs::metadata(&directory.0).unwrap().permissions().mode() & 0o777, + 0o700 + ); + for name in [LOCK_FILE, SNAPSHOT_FILE, WAL_FILE] { + assert_eq!( + fs::metadata(directory.0.join(name)) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o600 + ); + } + } +} diff --git a/crates/cfr/tests/persistence.rs b/crates/cfr/tests/persistence.rs new file mode 100644 index 0000000..eeff6b7 --- /dev/null +++ b/crates/cfr/tests/persistence.rs @@ -0,0 +1,343 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +//! Real-filesystem restart tests for the public persistence boundary. + +#![allow(missing_docs)] + +use cfr::persistence::{ + Error, InboundId, PendingDelivery, PersistenceOptions, PersistentConference, +}; +use cfr::{Codec, Joining, Policy, Recipient, SigPublic}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1); + +struct TestDirectory(PathBuf); + +impl TestDirectory { + fn new(label: &str) -> Self { + let id = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + Self(std::env::temp_dir().join(format!( + "cfr-persistence-{label}-{}-{id}", + std::process::id() + ))) + } + + fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn policy() -> Policy { + Policy::leaderless(2) +} + +fn inbound_id(delivery: &PendingDelivery) -> InboundId { + InboundId::from_bytes(*delivery.delivery_key.as_bytes()) +} + +fn acknowledge_all(conference: &mut PersistentConference) { + let deliveries = conference.pending_deliveries(); + for delivery in deliveries { + assert!(conference.acknowledge(delivery.id).unwrap()); + } +} + +fn deliver(source: &mut PersistentConference, target: &mut PersistentConference) -> usize { + let source_identity = source.identity(); + let target_identity = target.identity(); + let deliveries = source.pending_deliveries(); + let mut delivered = 0; + for delivery in deliveries { + let addressed = match delivery.recipient { + Recipient::Everyone => source_identity != target_identity, + Recipient::Peer(peer) => peer == target_identity, + }; + if !addressed { + continue; + } + target + .handle_inbound(inbound_id(&delivery), &delivery.payload) + .unwrap(); + assert!(source.acknowledge(delivery.id).unwrap()); + delivered += 1; + } + delivered +} + +fn settle(alice: &mut PersistentConference, bob: &mut PersistentConference) { + for _ in 0..256 { + let delivered = deliver(alice, bob) + deliver(bob, alice); + if delivered == 0 { + return; + } + } + panic!("persistent protocol flow did not settle"); +} + +fn pair(alice_path: &Path, bob_path: &Path) -> (PersistentConference, PersistentConference) { + pair_with_bob_options(alice_path, bob_path, PersistenceOptions::default()) +} + +fn pair_with_bob_options( + alice_path: &Path, + bob_path: &Path, + bob_options: PersistenceOptions, +) -> (PersistentConference, PersistentConference) { + let mut alice = PersistentConference::create(alice_path, policy()).unwrap(); + acknowledge_all(&mut alice); + let joining = Joining::new(policy()).unwrap(); + let bob_identity = joining.identity(); + alice.invite(&joining.key_package()).unwrap(); + let welcome = alice + .pending_deliveries() + .into_iter() + .find(|delivery| delivery.recipient == Recipient::Peer(bob_identity)) + .expect("invite transaction must durably queue a welcome"); + let mut bob = + PersistentConference::join_with_options(bob_path, joining, &welcome.payload, bob_options) + .unwrap(); + assert!(alice.acknowledge(welcome.id).unwrap()); + settle(&mut alice, &mut bob); + bob.rekey().unwrap(); + settle(&mut alice, &mut bob); + assert_eq!(alice.version(), bob.version()); + (alice, bob) +} + +#[test] +fn create_shutdown_open_preserves_identity_session_and_state() { + let directory = TestDirectory::new("create-open"); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + let identity = conference.identity(); + let session = conference.session_id(); + let members = conference.members(); + let version = conference.version(); + let sequence = conference.sequence(); + drop(conference); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.identity(), identity); + assert_eq!(reopened.session_id(), session); + assert_eq!(reopened.members(), members); + assert_eq!(reopened.version(), version); + assert_eq!(reopened.sequence(), sequence); +} + +#[test] +fn unacknowledged_outbox_is_stable_across_restart_and_ack_is_durable() { + let directory = TestDirectory::new("outbox"); + let mut conference = PersistentConference::create(directory.path(), policy()).unwrap(); + acknowledge_all(&mut conference); + conference.rekey().unwrap(); + let pending = conference.pending_deliveries(); + assert!(!pending.is_empty()); + drop(conference); + + let mut reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.pending_deliveries(), pending); + let acknowledged = pending[0].id; + assert!(reopened.acknowledge(acknowledged).unwrap()); + assert!(!reopened.acknowledge(acknowledged).unwrap()); + drop(reopened); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert!(reopened + .pending_deliveries() + .iter() + .all(|delivery| delivery.id != acknowledged)); +} + +#[test] +fn inbound_dedup_and_conflict_survive_restart_without_new_output() { + let alice_directory = TestDirectory::new("dedup-alice"); + let bob_directory = TestDirectory::new("dedup-bob"); + let (mut alice, mut bob) = pair(alice_directory.path(), bob_directory.path()); + alice.rekey().unwrap(); + let delivery = alice + .pending_deliveries() + .into_iter() + .find(|delivery| delivery.recipient == Recipient::Everyone) + .unwrap(); + let id = inbound_id(&delivery); + let first = bob.handle_inbound(id, &delivery.payload).unwrap(); + assert!(!first.duplicate); + let sequence = bob.sequence(); + let outbox = bob.pending_deliveries(); + drop(bob); + + let mut bob = PersistentConference::open(bob_directory.path()).unwrap(); + let duplicate = bob.handle_inbound(id, &delivery.payload).unwrap(); + assert!(duplicate.duplicate); + assert!(duplicate.events.is_empty()); + assert!(duplicate.deliveries.is_empty()); + assert_eq!(bob.sequence(), sequence); + assert_eq!(bob.pending_deliveries(), outbox); + + let mut conflicting = delivery.payload.clone(); + conflicting.push(0); + assert!(matches!( + bob.handle_inbound(id, &conflicting), + Err(Error::IdempotencyConflict { id: conflict }) if conflict == id + )); + assert_eq!(bob.sequence(), sequence); + assert_eq!(bob.pending_deliveries(), outbox); +} + +#[test] +fn control_flow_and_media_ratchets_continue_across_restart() { + let alice_directory = TestDirectory::new("flow-alice"); + let bob_directory = TestDirectory::new("flow-bob"); + let (mut alice, mut bob) = pair(alice_directory.path(), bob_directory.path()); + + let first = alice.protect(Codec::Generic, b"frame zero", false).unwrap(); + assert_eq!(PersistentConference::inspect(&first).unwrap().counter, 0); + let opened = bob.open_media(&first).unwrap(); + assert_eq!(opened.0, alice.identity()); + assert_eq!(opened.1, b"frame zero"); + drop(alice); + drop(bob); + + let mut alice = PersistentConference::open(alice_directory.path()).unwrap(); + let mut bob = PersistentConference::open(bob_directory.path()).unwrap(); + assert!( + bob.open_media(&first).is_err(), + "media replay must survive restart" + ); + let second = alice.protect(Codec::Generic, b"frame one", false).unwrap(); + assert_eq!(PersistentConference::inspect(&second).unwrap().counter, 1); + assert_eq!(bob.open_media(&second).unwrap().1, b"frame one"); + + bob.rekey().unwrap(); + settle(&mut alice, &mut bob); + drop(alice); + let mut alice = PersistentConference::open(alice_directory.path()).unwrap(); + alice.rekey().unwrap(); + settle(&mut alice, &mut bob); + assert_eq!(alice.version(), bob.version()); + let post_restart = alice + .protect(Codec::Generic, b"after restart", false) + .unwrap(); + assert_eq!(bob.open_media(&post_restart).unwrap().1, b"after restart"); +} + +#[test] +fn one_directory_has_one_writer() { + let directory = TestDirectory::new("lock"); + let first = PersistentConference::create(directory.path(), policy()).unwrap(); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::Locked) + )); + drop(first); + assert!(PersistentConference::open(directory.path()).is_ok()); +} + +#[test] +fn open_never_silently_creates_missing_state() { + let directory = TestDirectory::new("missing"); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::NotFound) + )); + assert!(!directory.path().exists()); +} + +#[test] +fn identities_remain_distinct_across_persistent_join() { + let alice_directory = TestDirectory::new("identity-alice"); + let bob_directory = TestDirectory::new("identity-bob"); + let (alice, bob) = pair(alice_directory.path(), bob_directory.path()); + let identities: std::collections::BTreeSet = + [alice.identity(), bob.identity()].into_iter().collect(); + assert_eq!(identities.len(), 2); + assert_eq!(alice.session_id(), bob.session_id()); +} + +#[test] +fn outbox_exhaustion_fails_before_protocol_or_sequence_commit() { + let directory = TestDirectory::new("outbox-limit"); + let options = PersistenceOptions { + max_outbox_entries: 3, + ..PersistenceOptions::default() + }; + let mut conference = + PersistentConference::create_with_options(directory.path(), policy(), options).unwrap(); + assert_eq!(conference.pending_deliveries().len(), 3); + acknowledge_all(&mut conference); + for _ in 0..3 { + conference.rekey().unwrap(); + } + let sequence = conference.sequence(); + let version = conference.version(); + let pending = conference.pending_deliveries(); + assert!(matches!( + conference.rekey(), + Err(Error::LimitExceeded("durable outbox is full")) + )); + assert_eq!(conference.sequence(), sequence); + assert_eq!(conference.version(), version); + assert_eq!(conference.pending_deliveries(), pending); + drop(conference); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.sequence(), sequence); + assert_eq!(reopened.version(), version); + assert_eq!(reopened.pending_deliveries(), pending); +} + +#[test] +fn inbound_idempotency_eviction_is_bounded_and_durable() { + let alice_directory = TestDirectory::new("window-alice"); + let bob_directory = TestDirectory::new("window-bob"); + let options = PersistenceOptions { + inbound_window: 2, + ..PersistenceOptions::default() + }; + let (mut alice, mut bob) = + pair_with_bob_options(alice_directory.path(), bob_directory.path(), options); + let mut first = None; + for index in 0..3 { + alice.rekey().unwrap(); + let delivery = alice + .pending_deliveries() + .into_iter() + .find(|delivery| delivery.recipient == Recipient::Everyone) + .unwrap(); + let result = bob + .handle_inbound(inbound_id(&delivery), &delivery.payload) + .unwrap(); + assert!(!result.duplicate); + assert!(alice.acknowledge(delivery.id).unwrap()); + if index == 0 { + first = Some(delivery); + } + } + let first = first.unwrap(); + let before = bob.sequence(); + let replay_after_eviction = bob + .handle_inbound(inbound_id(&first), &first.payload) + .unwrap(); + assert!(!replay_after_eviction.duplicate); + assert_eq!(bob.sequence(), before + 1); + drop(bob); + + let mut bob = PersistentConference::open(bob_directory.path()).unwrap(); + let durable_duplicate = bob + .handle_inbound(inbound_id(&first), &first.payload) + .unwrap(); + assert!(durable_duplicate.duplicate); +} diff --git a/crates/cfr/tests/persistence_snapshot.rs b/crates/cfr/tests/persistence_snapshot.rs new file mode 100644 index 0000000..e7fd4e5 --- /dev/null +++ b/crates/cfr/tests/persistence_snapshot.rs @@ -0,0 +1,325 @@ +// Copyright Nixort 2026. +// +// License: GNU General Public License v3.0 only. +// You can find the license file in the project root. +// +// Causal Frontier Ratchet (CFR). + +//! Filesystem corruption, version, and WAL-compaction acceptance tests. + +#![allow(missing_docs)] + +use cfr::layers::crypto::hash; +use cfr::persistence::{ + Error, PersistenceOptions, PersistentConference, VersionKind, + CURRENT_PERSISTENCE_SCHEMA_VERSION, +}; +use cfr::Policy; +use std::fs::{self, OpenOptions}; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +const SNAPSHOT_PREFIX: usize = 60; +const WAL_HEADER: usize = 12; +const RECORD_PREFIX: usize = 56; + +static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(1); + +struct TestDirectory(PathBuf); + +impl TestDirectory { + fn new(label: &str) -> Self { + let id = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + Self(std::env::temp_dir().join(format!( + "cfr-persistence-snapshot-{label}-{}-{id}", + std::process::id() + ))) + } + + fn path(&self) -> &Path { + &self.0 + } + + fn snapshot(&self) -> PathBuf { + self.0.join("snapshot") + } + + fn wal(&self) -> PathBuf { + self.0.join("wal") + } +} + +impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn policy() -> Policy { + Policy::leaderless(2) +} + +fn read_u64(bytes: &[u8]) -> u64 { + u64::from_be_bytes(bytes.try_into().unwrap()) +} + +fn write_synced(path: &Path, bytes: &[u8]) { + let mut file = OpenOptions::new() + .write(true) + .truncate(true) + .open(path) + .unwrap(); + file.write_all(bytes).unwrap(); + file.sync_all().unwrap(); +} + +fn rewrite_snapshot_checksum(bytes: &mut [u8]) { + let sequence_bytes: [u8; 8] = bytes[12..20].try_into().unwrap(); + let length = usize::try_from(read_u64(&bytes[20..28])).unwrap(); + let payload = &bytes[SNAPSHOT_PREFIX..SNAPSHOT_PREFIX + length]; + let checksum = hash( + b"cfr/persistence/snapshot-checksum", + &[&sequence_bytes, payload], + ); + bytes[28..60].copy_from_slice(&checksum); +} + +#[test] +fn valid_wal_newer_than_snapshot_restores_latest_state() { + let directory = TestDirectory::new("newer-wal"); + let mut conference = PersistentConference::create(directory.path(), policy()).unwrap(); + let identity = conference.identity(); + conference.tick().unwrap(); + conference.rekey().unwrap(); + let latest_sequence = conference.sequence(); + let latest_version = conference.version(); + drop(conference); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.identity(), identity); + assert_eq!(reopened.sequence(), latest_sequence); + assert_eq!(reopened.version(), latest_version); +} + +#[test] +fn incomplete_final_wal_record_is_removed_without_losing_valid_state() { + let directory = TestDirectory::new("tail"); + let mut conference = PersistentConference::create(directory.path(), policy()).unwrap(); + conference.tick().unwrap(); + let sequence = conference.sequence(); + let identity = conference.identity(); + drop(conference); + + let wal_path = directory.wal(); + let valid_length = fs::metadata(&wal_path).unwrap().len(); + let mut wal = OpenOptions::new().append(true).open(&wal_path).unwrap(); + wal.write_all(b"CFRREC\0\0").unwrap(); + wal.write_all(&(sequence + 1).to_be_bytes()).unwrap(); + wal.write_all(&100u64.to_be_bytes()).unwrap(); + wal.write_all(&[0xA5; 32]).unwrap(); + wal.write_all(b"partial").unwrap(); + wal.sync_all().unwrap(); + drop(wal); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.sequence(), sequence); + assert_eq!(reopened.identity(), identity); + assert_eq!(fs::metadata(wal_path).unwrap().len(), valid_length); +} + +#[test] +fn complete_wal_checksum_or_marker_corruption_fails_closed() { + for corrupt_marker in [false, true] { + let directory = TestDirectory::new(if corrupt_marker { "marker" } else { "checksum" }); + let mut conference = PersistentConference::create(directory.path(), policy()).unwrap(); + conference.tick().unwrap(); + drop(conference); + + let wal_path = directory.wal(); + let mut bytes = fs::read(&wal_path).unwrap(); + let payload_length = + usize::try_from(read_u64(&bytes[WAL_HEADER + 16..WAL_HEADER + 24])).unwrap(); + if corrupt_marker { + bytes[WAL_HEADER + RECORD_PREFIX + payload_length] ^= 1; + } else { + bytes[WAL_HEADER + 24] ^= 1; + } + write_synced(&wal_path, &bytes); + let before = fs::read(&wal_path).unwrap(); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::Corrupt(_)) + )); + assert_eq!(fs::read(wal_path).unwrap(), before); + } +} + +#[test] +fn corrupt_snapshot_falls_back_to_newer_full_state_wal() { + let directory = TestDirectory::new("snapshot-fallback"); + let mut conference = PersistentConference::create(directory.path(), policy()).unwrap(); + let identity = conference.identity(); + conference.tick().unwrap(); + let sequence = conference.sequence(); + drop(conference); + + let snapshot_path = directory.snapshot(); + let mut snapshot = fs::read(&snapshot_path).unwrap(); + snapshot[28] ^= 1; + write_synced(&snapshot_path, &snapshot); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.identity(), identity); + assert_eq!(reopened.sequence(), sequence); +} + +#[test] +fn corrupt_snapshot_without_wal_state_fails_closed() { + let directory = TestDirectory::new("snapshot-only-corrupt"); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + drop(conference); + let snapshot_path = directory.snapshot(); + let mut snapshot = fs::read(&snapshot_path).unwrap(); + snapshot[28] ^= 1; + write_synced(&snapshot_path, &snapshot); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::Corrupt(_)) + )); +} + +#[test] +fn unknown_snapshot_and_wal_envelope_versions_are_explicit() { + let snapshot_directory = TestDirectory::new("snapshot-version"); + let conference = PersistentConference::create(snapshot_directory.path(), policy()).unwrap(); + drop(conference); + let path = snapshot_directory.snapshot(); + let mut snapshot = fs::read(&path).unwrap(); + snapshot[8..12].copy_from_slice(&99u32.to_be_bytes()); + write_synced(&path, &snapshot); + assert!(matches!( + PersistentConference::open(snapshot_directory.path()), + Err(Error::UnsupportedVersion { + kind: VersionKind::StoreEnvelope, + found: 99 + }) + )); + + let wal_directory = TestDirectory::new("wal-version"); + let conference = PersistentConference::create(wal_directory.path(), policy()).unwrap(); + drop(conference); + let path = wal_directory.wal(); + let mut wal = fs::read(&path).unwrap(); + wal[8..12].copy_from_slice(&77u32.to_be_bytes()); + write_synced(&path, &wal); + assert!(matches!( + PersistentConference::open(wal_directory.path()), + Err(Error::UnsupportedVersion { + kind: VersionKind::StoreEnvelope, + found: 77 + }) + )); +} + +#[test] +fn unknown_logical_schema_is_explicit_and_not_autodetected() { + assert_eq!(CURRENT_PERSISTENCE_SCHEMA_VERSION, 1); + let directory = TestDirectory::new("schema-version"); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + drop(conference); + let path = directory.snapshot(); + let mut snapshot = fs::read(&path).unwrap(); + assert_eq!(snapshot[SNAPSHOT_PREFIX], 2, "schema is a TLV integer"); + snapshot[SNAPSHOT_PREFIX + 1..SNAPSHOT_PREFIX + 9].copy_from_slice(&99u64.to_be_bytes()); + rewrite_snapshot_checksum(&mut snapshot); + write_synced(&path, &snapshot); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::UnsupportedVersion { + kind: VersionKind::PersistenceSchema, + found: 99 + }) + )); +} + +#[test] +fn small_checkpoint_threshold_compacts_before_each_crossing() { + let directory = TestDirectory::new("wal-limit"); + let options = PersistenceOptions { + checkpoint_threshold: 1, + max_wal_bytes: 256 * 1024, + ..PersistenceOptions::default() + }; + let mut conference = + PersistentConference::create_with_options(directory.path(), policy(), options).unwrap(); + conference.tick().unwrap(); + conference.tick().unwrap(); + conference.rekey().unwrap(); + let sequence = conference.sequence(); + let identity = conference.identity(); + assert!(fs::metadata(directory.wal()).unwrap().len() <= options.max_wal_bytes); + drop(conference); + + let reopened = PersistentConference::open(directory.path()).unwrap(); + assert_eq!(reopened.sequence(), sequence); + assert_eq!(reopened.identity(), identity); + assert!(fs::metadata(directory.wal()).unwrap().len() <= options.max_wal_bytes); +} + +#[test] +fn snapshot_with_trailing_bytes_is_not_accepted() { + let directory = TestDirectory::new("snapshot-trailing"); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + drop(conference); + let path = directory.snapshot(); + let mut file = OpenOptions::new().append(true).open(&path).unwrap(); + file.write_all(b"trailing").unwrap(); + file.sync_all().unwrap(); + drop(file); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::Corrupt(_)) + )); +} + +#[test] +fn state_directory_cannot_be_recreated_over_existing_data() { + let directory = TestDirectory::new("already-exists"); + fs::create_dir(directory.path()).unwrap(); + assert!(matches!( + PersistentConference::create(directory.path(), policy()), + Err(Error::AlreadyExists) + )); + assert!(fs::metadata(directory.path()).unwrap().is_dir()); +} + +#[test] +fn missing_wal_in_existing_state_is_corruption_not_absence() { + let directory = TestDirectory::new("missing-wal"); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + drop(conference); + fs::remove_file(directory.wal()).unwrap(); + assert!(matches!( + PersistentConference::open(directory.path()), + Err(Error::Corrupt("WAL file is missing")) + )); +} + +#[test] +fn single_component_relative_state_path_is_supported() { + let id = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + let directory = TestDirectory(PathBuf::from(format!( + "cfr-relative-persistence-{}-{id}", + std::process::id() + ))); + let conference = PersistentConference::create(directory.path(), policy()).unwrap(); + let identity = conference.identity(); + drop(conference); + assert_eq!( + PersistentConference::open(directory.path()) + .unwrap() + .identity(), + identity + ); +} diff --git a/docs/integration.md b/docs/integration.md index 7e77579..400cd84 100644 --- a/docs/integration.md +++ b/docs/integration.md @@ -126,7 +126,96 @@ missed operations and missing node keys. An inviter that cannot currently derive the key will refuse to admit — it cannot hand over material it does not have. Resync, then retry. -## 7. Post-compromise +## 7. Durable process restart + +With the default `std` feature, `cfr::persistence::PersistentConference` owns a +`Conference`, a bounded inbound idempotency window and a durable control-message +outbox. It never exposes `&mut Conference`; every state-changing protocol and +media operation crosses the same filesystem transaction boundary. + +```rust +use cfr::persistence::{InboundId, PersistentConference}; + +let mut conference = PersistentConference::create("call-state", policy)?; + +// The transport supplies a stable ID. It is not generated by CFR. +let result = conference.handle_inbound(InboundId::from_bytes(transport_id), &payload)?; + +for delivery in conference.pending_deliveries() { + transport.send( + delivery.delivery_key.as_bytes(), + delivery.recipient, + &delivery.payload, + )?; + // Acknowledge only after the transport accepts responsibility for delivery. + conference.acknowledge(delivery.id)?; +} + +drop(conference); // process shutdown +let conference = PersistentConference::open("call-state")?; +``` + +`create` refuses an existing path and `open` returns `NotFound` for an absent +state directory. There is deliberately no `open_or_create`: a missing state can +never silently replace the identity or session. A newcomer uses +`PersistentConference::join` with its `Joining` value and welcome payload. + +An inbound ID has three outcomes: + +| condition | result | +|---|---| +| new ID | protocol mutation, inbound digest and resulting outbox rows commit atomically | +| same ID and bytes | `duplicate = true`; no events, outbox rows or transaction | +| same ID, different bytes | `IdempotencyConflict`; no state change | + +Outbox IDs are monotonic and delivery keys are deterministic over the session, +local identity, ID, recipient and exact payload. Unacknowledged rows survive +restart in ID order. A repeated acknowledgement returns `false` without writing +a transaction. This is an at-least-once queue: the transport must use the +delivery key to suppress duplicate sends around its own crash boundary. + +For every mutation, CFR first imports an isolated candidate, applies the +operation there, encodes and validates the complete candidate, appends a +full-state WAL record and waits for `sync_data`. Only then does it replace the +live state or return media plaintext/ciphertext, events or outbound IDs. This +includes `protect` and `open_media`, so a restart cannot roll back a sender +counter or an authenticated replay window. + +The logical state schema and snapshot/WAL envelope have independent internal +version tags, both currently `1`. Unknown tags fail with `UnsupportedVersion`; +there are no guessed legacy formats or synthetic migrations. Recovery verifies +every complete WAL record. It truncates only an incomplete final record, fails +closed on a complete bad checksum/marker/state, and can use a newer full-state +WAL record when the snapshot is corrupt. Snapshot replacement and WAL reset use +synced temporary files, atomic rename and directory sync. + +Default persisted limits are: + +| bound | default | +|---|---:| +| inbound idempotency IDs | 4,096 | +| unacknowledged outbox rows | 1,024 | +| logical state / WAL-record payload | 4 MiB | +| WAL file | 64 MiB | +| checkpoint threshold | 32 MiB | + +`PersistenceOptions` can lower these values for a deployment or test. The +options themselves are persisted and validated on open. Before an append would +cross the threshold or WAL limit, the current committed state is checkpointed +and the WAL is reset; if one candidate still cannot fit, the operation fails +before changing live state. `checkpoint()` also exposes explicit compaction. + +Only one writable handle can own a state directory. CFR holds an OS advisory +lock on an open file descriptor for the handle lifetime; a diagnostic PID file +is not used for ownership. On Unix, new directories are `0700` and state files +are `0600`. + +The store contains secret key material in plaintext. Permissions are not +encryption, and checksums are corruption detection rather than authentication. +See the persisted-state boundary in [`security.md`](security.md) before deciding +where the directory and its backups may live. + +## 8. Post-compromise After any suspicion that a device was compromised: @@ -138,7 +227,7 @@ This rotates the prekey and contributes. Both are needed. Rotation alone leaves the attacker holding the current key; a contribution alone leaves it able to read the channels. -## 8. Choosing a policy +## 9. Choosing a policy `Policy::leaderless(quorum)` is the configuration the analysis assumes. A quorum of two means two distinct participants must agree to evict, which stops a single @@ -148,7 +237,7 @@ Naming administrators is supported and weakens the leaderless property: a named identity can evict unilaterally. Use it only when the deployment already has an authority worth trusting with that. -## 9. Post-quantum +## 10. Post-quantum Enable the `pq` feature to use X25519 + ML-KEM-768 hybrid key encapsulation. Every reduction in the analysis goes through unchanged; the assumption becomes diff --git a/docs/security.md b/docs/security.md index 9e8b6df..5b0ad96 100644 --- a/docs/security.md +++ b/docs/security.md @@ -102,6 +102,24 @@ control flow in this library has not been audited for timing behaviour. copy was left by an optimiser, an allocator, a swap file or a hypervisor. Forward secrecy is a statement about the whole system, not about one type. +**Persisted state at rest.** `PersistentConference` snapshots and WAL records +contain the identity signing seed, conference seed, retained node keys, +unretired prekeys, channel ratchets, media counters and replay windows. The +store does not encrypt them. Unix `0700`/`0600` modes limit accidental access; +they do not protect against a compromised account, disk image, backup operator +or filesystem administrator. Use an encrypted volume or an application-owned +encryption layer when that threat is in scope. + +Store checksums detect torn writes and accidental corruption. They are unkeyed +and do not authenticate state against an attacker who can rewrite files and +recompute checksums. The single-writer lock is advisory, so a process that +deliberately ignores OS advisory locks is outside the integrity boundary. + +Old filesystem snapshots, copy-on-write extents and backups can retain key +material after CFR has checkpointed or erased its live copy. Keeping them +extends the practical compromise horizon and must be governed by the +deployment's retention and secure-deletion policy. + **Formal verification of this code.** The construction has a separate written analysis with reductions and an exhaustive symbolic check. That analysis is about the construction. This crate is the construction *implemented*, and the bridge