diff --git a/jcp/SPEC.md b/jcp/SPEC.md index af0661243..7f6a93373 100644 --- a/jcp/SPEC.md +++ b/jcp/SPEC.md @@ -1265,6 +1265,14 @@ action: the grant's holder is the enforcement point that ran it, as a steward is in Mesh, rather than the agent the grant governs reporting its own consumption. The endpoint behind the enforcement point is invoked only on Allow. +**What the endpoint holds survives it.** An endpoint that can restart is bound +by §4.11: it MUST record each request it admits before deciding it, and each +revocation it honours before applying it, and on restart MUST replay them +through the gate, each request at the time it was decided, before deciding +anything new. That re-establishes the spend, the revocations, and the memory of +requests still inside the window. A request it cannot record is denied +`unrecorded`. + ## 12. Non-goals JCP does not define a model or a tool registry. It does diff --git a/jcp/jcp-rs/crates/jcp-schema/src/lib.rs b/jcp/jcp-rs/crates/jcp-schema/src/lib.rs index 72812a0da..90514e094 100644 --- a/jcp/jcp-rs/crates/jcp-schema/src/lib.rs +++ b/jcp/jcp-rs/crates/jcp-schema/src/lib.rs @@ -609,6 +609,10 @@ pub enum DenyReason { /// A wire request the enforcement point has already decided, presented /// again (SPEC §11). It is decided once, and debited once. Replayed, + /// The enforcement point could not record the request before deciding it + /// (SPEC §4.11), so it does not decide it: a decision it could not record + /// would be lost, with its debit, on a restart. + Unrecorded, } impl DenyReason { @@ -632,6 +636,7 @@ impl DenyReason { DenyReason::TaintBudgetExceeded => "taint-budget-exceeded", DenyReason::StaleRequest => "stale-request", DenyReason::Replayed => "replayed", + DenyReason::Unrecorded => "unrecorded", } } } diff --git a/jcp/jcp-rs/crates/jcp-wire/src/journal.rs b/jcp/jcp-rs/crates/jcp-wire/src/journal.rs new file mode 100644 index 000000000..c4ae85831 --- /dev/null +++ b/jcp/jcp-rs/crates/jcp-wire/src/journal.rs @@ -0,0 +1,108 @@ +//! The endpoint's journal (SPEC §4.11): every request it decided and every +//! revocation it honoured, recorded before it acts, and replayed through the +//! gate when it restarts, so a restart gives no budget back, unrevokes +//! nothing, and forgets no request it may still be shown again. + +use std::fs::{File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::path::Path; + +use jcp_schema::{canonical_cbor, from_canonical_cbor, Grant, SignedRevocation}; +use serde::{Deserialize, Serialize}; + +use crate::WireRequest; + +/// One record in an endpoint's journal. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum JournalRecord { + /// A request the endpoint admitted, and the time it decided it at. + /// Written before the decision is made, so a decision the endpoint was + /// interrupted in is still on record and is debited on replay. + Decided { at: u64, request: WireRequest }, + /// A revocation the endpoint honoured, with the grant it revokes. + Revoked { revocation: SignedRevocation, grant: Grant }, +} + +/// Where an endpoint records what it decided and revoked. +pub trait Journal { + /// Record one record durably. An error refuses what it was for: a request + /// that could not be recorded is denied `unrecorded`, and a revocation + /// that could not be recorded is not applied. + fn append(&mut self, record: &JournalRecord) -> io::Result<()>; +} + +/// A journal in memory, and the default. It is enough for an endpoint that +/// never restarts; one that can restart MUST use a durable journal such as +/// [`FileJournal`] (SPEC §4.11). +#[derive(Clone, Debug, Default)] +pub struct MemoryJournal { + records: Vec, +} + +impl MemoryJournal { + /// The records so far, in order. + pub fn records(&self) -> &[JournalRecord] { + &self.records + } +} + +impl Journal for MemoryJournal { + fn append(&mut self, record: &JournalRecord) -> io::Result<()> { + self.records.push(record.clone()); + Ok(()) + } +} + +/// A journal in an append-only file: records framed as on the wire, a 4-byte +/// big-endian length then canonical det-CBOR, each synced to disk before +/// [`append`](Journal::append) returns. +pub struct FileJournal { + file: File, +} + +impl FileJournal { + /// Open the journal at `path`, creating it if absent. Returns it with the + /// records it already holds, for [`Endpoint::with_journal`](crate::Endpoint::with_journal) + /// to replay. + /// + /// A final record cut short is an append the endpoint was interrupted in. + /// It is dropped and cut from the file, which is safe for both kinds: a + /// request never recorded was never decided, and a revocation never + /// recorded was never applied. A complete record that does not decode is an + /// error, because a journal that cannot be read cannot re-establish state. + pub fn open(path: impl AsRef) -> io::Result<(Self, Vec)> { + let mut file = OpenOptions::new().read(true).append(true).create(true).open(path)?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes)?; + let mut records = Vec::new(); + let mut at = 0usize; + while bytes.len() - at >= 4 { + let len = u32::from_be_bytes([bytes[at], bytes[at + 1], bytes[at + 2], bytes[at + 3]]) as usize; + if bytes.len() - at - 4 < len { + break; + } + let body = &bytes[at + 4..at + 4 + len]; + let record = from_canonical_cbor(body).map_err(|e| { + io::Error::new(io::ErrorKind::InvalidData, format!("journal record {}: {e}", records.len())) + })?; + records.push(record); + at += 4 + len; + } + if at < bytes.len() { + file.set_len(at as u64)?; + file.sync_data()?; + } + Ok((FileJournal { file }, records)) + } +} + +impl Journal for FileJournal { + fn append(&mut self, record: &JournalRecord) -> io::Result<()> { + let body = canonical_cbor(record).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + let mut framed = Vec::with_capacity(4 + body.len()); + framed.extend_from_slice(&(body.len() as u32).to_be_bytes()); + framed.extend_from_slice(&body); + self.file.write_all(&framed)?; + self.file.sync_data() + } +} diff --git a/jcp/jcp-rs/crates/jcp-wire/src/lib.rs b/jcp/jcp-rs/crates/jcp-wire/src/lib.rs index e21e0b7eb..53dc9f8a5 100644 --- a/jcp/jcp-rs/crates/jcp-wire/src/lib.rs +++ b/jcp/jcp-rs/crates/jcp-wire/src/lib.rs @@ -25,6 +25,15 @@ //! requester is the grant's holder and the party that metered the action //! (SPEC §11): an endpoint takes no figure from a party it does not hold to //! account for it. +//! +//! An endpoint that can restart keeps a [`Journal`] (SPEC §4.11): each admitted +//! request is recorded before it is decided, each revocation before it is +//! applied, and [`Endpoint::with_journal`] replays them through the gate at the +//! times they were decided, so a restart gives no budget back, unrevokes +//! nothing, and forgets no request that may still be shown again. + +mod journal; +pub use journal::{FileJournal, Journal, JournalRecord, MemoryJournal}; use std::collections::HashMap; use std::io::{Read, Write}; @@ -53,6 +62,21 @@ pub enum WireError { Closed, } +/// Errors from an endpoint's journal: an append it could not make, a +/// revocation it refused, or a journal that does not replay. +#[derive(Debug, thiserror::Error)] +pub enum EndpointError { + #[error("journal: {0}")] + Journal(#[from] std::io::Error), + #[error("revocation: {0}")] + Revocation(#[from] jcp_grant::GrantError), + /// A replayed record the gate would not have produced: a request whose + /// caller signature fails, or that was decided twice, or a revocation that + /// does not verify. + #[error("journal record {record} is inconsistent: {reason}")] + Inconsistent { record: usize, reason: String }, +} + /// One action over the wire: decide `frame` against `chain` (root first, leaf /// last) for `caller` at metered `cost`. A single un-delegated grant is a /// one-link chain. @@ -201,6 +225,8 @@ pub struct Endpoint { skew_secs: u64, /// The requests decided within the window, by content id, with their stamp. decided: HashMap, + /// Where each admitted request and each revocation is recorded first. + journal: Box, } impl Endpoint { @@ -214,9 +240,54 @@ impl Endpoint { clock: system_clock, skew_secs: DEFAULT_SKEW_SECS, decided: HashMap::new(), + journal: Box::new(MemoryJournal::default()), } } + /// Record in `journal`, first replaying `recorded`, the records it already + /// holds, through the gate (SPEC §4.11). Each request is decided again at + /// the time it was decided, which re-establishes the spend and the memory + /// of requests still inside the window; each revocation is applied again. + /// Configure the runtime (trust roots, taint floors) as it was before + /// calling this, since the replay decides under it. + pub fn with_journal(mut self, journal: impl Journal + Send + 'static, recorded: &[JournalRecord]) -> Result { + let inconsistent = |record: usize, reason: &str| EndpointError::Inconsistent { record, reason: reason.into() }; + for (i, record) in recorded.iter().enumerate() { + match record { + JournalRecord::Decided { at, request } => { + if !request.verify_caller() { + return Err(inconsistent(i, "the caller's signature does not verify")); + } + let id = ContentId::of(request).map_err(|e| inconsistent(i, &e.to_string()))?; + if self.decided.insert(id, request.now).is_some() { + return Err(inconsistent(i, "the request was decided twice")); + } + self.runtime.decide_chained(&request.frame, &request.chain, request.caller, request.cost, *at); + } + JournalRecord::Revoked { revocation, grant } => { + self.runtime + .apply_revocation(revocation, grant) + .map_err(|e| inconsistent(i, &e.to_string()))?; + } + } + } + let now = (self.clock)(); + let skew = self.skew_secs; + self.decided.retain(|_, stamp| stamp.saturating_add(skew) >= now); + self.journal = Box::new(journal); + Ok(self) + } + + /// Honour a revocation: verify it, record it, then apply it, so it + /// survives a restart. Applying one to `runtime` directly works until the + /// endpoint restarts. + pub fn revoke(&mut self, revocation: &jcp_schema::SignedRevocation, grant: &jcp_schema::Grant) -> Result<(), EndpointError> { + jcp_grant::verify_revocation(revocation, grant)?; + self.journal.append(&JournalRecord::Revoked { revocation: revocation.clone(), grant: grant.clone() })?; + self.runtime.apply_revocation(revocation, grant)?; + Ok(()) + } + /// Decide at `clock` (unix seconds) instead of the system clock. pub fn with_clock(mut self, clock: fn() -> u64) -> Self { self.clock = clock; @@ -251,7 +322,15 @@ impl Endpoint { /// other decision — the refused impersonation is evidence too. pub fn handle(&mut self, req: &WireRequest) -> WireResponse { let now = (self.clock)(); - let refused = if req.verify_caller() { self.admit(req, now).err() } else { Some(jcp_schema::DenyReason::BadSignature) }; + let mut refused = if req.verify_caller() { self.admit(req, now).err() } else { Some(jcp_schema::DenyReason::BadSignature) }; + // Recorded before it is decided: a decision that is lost on a restart + // would take its debit with it (SPEC §4.11). + if refused.is_none() && self.journal.append(&JournalRecord::Decided { at: now, request: req.clone() }).is_err() { + if let Ok(id) = ContentId::of(req) { + self.decided.remove(&id); + } + refused = Some(jcp_schema::DenyReason::Unrecorded); + } let decision = match refused { None => self.runtime.decide_chained(&req.frame, &req.chain, req.caller, req.cost, now), Some(reason) => Decision { @@ -460,6 +539,118 @@ mod tests { assert!(ep.decided.is_empty(), "nothing outside the window is remembered"); } + /// A journal the test can read after the endpoint is gone. + #[derive(Clone, Default)] + struct Shared(std::sync::Arc>>); + impl Journal for Shared { + fn append(&mut self, record: &JournalRecord) -> std::io::Result<()> { + self.0.lock().unwrap().push(record.clone()); + Ok(()) + } + } + + /// A journal that cannot be written. + struct Broken; + impl Journal for Broken { + fn append(&mut self, _: &JournalRecord) -> std::io::Result<()> { + Err(std::io::Error::other("disk full")) + } + } + + fn endpoint(clock: fn() -> u64) -> Endpoint { + Endpoint::new(Runtime::new(), sk(7), "did:example:steward").with_clock(clock) + } + + /// A restart is not a fresh budget: the endpoint replays what it decided, + /// so the leaf has what it had left. + #[test] + fn a_restart_keeps_the_spend() { + let (chain, _, r) = fixture(); + let journal = Shared::default(); + let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap(); + assert!(before.handle(&stamped(&chain, r, 40_000, 10)).decision.is_allow()); + drop(before); + let recorded = journal.0.lock().unwrap().clone(); + let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap(); + assert_eq!(after.runtime.spent(&chain[1].grant.id()), 40_000, "the restart gave nothing back"); + let d = after.handle(&stamped(&chain, r, 20_000, 9)); + assert_eq!(d.decision.verdict, Verdict::Deny(DenyReason::BudgetExceeded)); + } + + /// A revocation honoured through the endpoint is recorded, and still + /// stands after a restart. + #[test] + fn a_revocation_survives_a_restart() { + let (chain, _, r) = fixture(); + let journal = Shared::default(); + let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap(); + let of_root = revoke(&sk(1), &chain[0].grant, [9; 16]).unwrap(); + assert!(before.revoke(&of_root, &chain[1].grant).is_err(), "a revocation of another grant"); + before.revoke(&of_root, &chain[0].grant).unwrap(); + drop(before); + let recorded = journal.0.lock().unwrap().clone(); + assert_eq!(recorded.len(), 1, "a refused revocation is not recorded"); + let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap(); + assert_eq!(after.handle(&stamped(&chain, r, 1, 10)).decision.verdict, Verdict::Deny(DenyReason::Revoked)); + } + + /// The endpoint remembers what it decided across a restart, so a request + /// still inside the window is refused, not decided a second time. + #[test] + fn a_request_decided_before_a_restart_is_still_refused() { + let (chain, _, r) = fixture(); + let journal = Shared::default(); + let req = stamped(&chain, r, 10_000, 10); + let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap(); + assert!(before.handle(&req).decision.is_allow()); + drop(before); + let recorded = journal.0.lock().unwrap().clone(); + let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap(); + assert_eq!(after.handle(&req).decision.verdict, Verdict::Deny(DenyReason::Replayed)); + assert_eq!(after.runtime.spent(&chain[1].grant.id()), 10_000); + + let twice = [recorded[0].clone(), recorded[0].clone()]; + assert!(matches!(endpoint(at_10).with_journal(MemoryJournal::default(), &twice), Err(EndpointError::Inconsistent { record: 1, .. }))); + } + + /// A request that cannot be recorded is not decided, and debits nothing. + #[test] + fn an_unrecordable_request_is_not_decided() { + let (chain, _, r) = fixture(); + let mut ep = endpoint(at_10).with_journal(Broken, &[]).unwrap(); + let req = stamped(&chain, r, 10_000, 10); + assert_eq!(ep.handle(&req).decision.verdict, Verdict::Deny(DenyReason::Unrecorded)); + assert_eq!(ep.runtime.spent(&chain[1].grant.id()), 0); + assert!(ep.decided.is_empty(), "an undecided request is not remembered as decided"); + } + + /// The file journal round-trips, and a record a crash cut short is dropped + /// and cut from the file rather than failing the restart. + #[test] + fn the_file_journal_drops_a_torn_record() { + let dir = std::env::temp_dir().join(format!("jcp-wire-journal-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("journal.cbor"); + let _ = std::fs::remove_file(&path); + let (chain, _, r) = fixture(); + { + let (journal, recorded) = FileJournal::open(&path).unwrap(); + assert!(recorded.is_empty()); + let mut ep = endpoint(at_10).with_journal(journal, &recorded).unwrap(); + assert!(ep.handle(&stamped(&chain, r, 30_000, 10)).decision.is_allow()); + } + let whole = std::fs::metadata(&path).unwrap().len(); + let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap(); + f.write_all(&[0, 0, 1, 0, 7, 7]).unwrap(); + drop(f); + let (journal, recorded) = FileJournal::open(&path).unwrap(); + assert_eq!(recorded.len(), 1); + assert_eq!(std::fs::metadata(&path).unwrap().len(), whole, "the torn record was cut"); + let after = endpoint(at_10).with_journal(journal, &recorded).unwrap(); + assert_eq!(after.runtime.spent(&chain[1].grant.id()), 30_000); + std::fs::remove_dir_all(&dir).unwrap(); + } + // the fixture's caller is sk(3) — sign as them fn request(chain: &[SignedGrant], _caller: Did, r: ContentId, trust: Trust, cost: u64) -> WireRequest { let frame = Frame {