clawft_kernel/
mesh_swarm_revoke.rs

1//! Revocation across the mesh (ADR-099 section 6, card mesh-placement-25).
2//!
3//! A revocation of a package id, signer key or artifact hash has to stop
4//! seeding and empty caches on every node, not only where the operator
5//! typed it. [`RevocationExchange`] carries it as a signed notice on the
6//! `mesh.artifact.revoke` control topic:
7//!
8//! - **Signed.** The notice is signed with a pinned trust-anchor key whose
9//!   origin is `Operator` or `Weftos`; a Cognitum release key, an unpinned
10//!   key or a bad signature is ignored. The domain tag keeps it from being
11//!   replayed as anything else.
12//! - **Applied.** A valid notice is recorded in the local
13//!   [`RevocationList`] and applied at once: grants dropped, bytes evicted,
14//!   `artifact.revoke` and `artifact.evict` chained
15//!   ([`ArtifactExchange::apply_revocations`]), the revocation itself chained
16//!   as `workload.revoke` (revoked by `mesh:<signer>`), and the
17//!   [`RevocationExchange::set_on_applied`] hook run so placement can stop
18//!   what is already running from it.
19//! - **Limited.** A signer key that is itself revoked can no longer issue
20//!   revocations and signatures are checked strictly. Cheap checks (size,
21//!   pinned key, revoked signer, already applied) come first and are free; a
22//!   notice that passes them spends a token from its own connection's bucket
23//!   before the signature verify, so junk on one connection cannot starve
24//!   notices from another, and the operator's own `issue` is exempt.
25//! - **Flooded.** A notice that was new here is forwarded to every other
26//!   peer; one that was already known is not, so a notice crosses a mesh
27//!   once and cannot loop. Revocations are only ever added by notices, so a
28//!   replayed old notice changes nothing.
29
30use std::sync::Arc;
31
32use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
33use serde::{Deserialize, Serialize};
34
35use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
36use crate::mesh_artifact::ArtifactExchange;
37use crate::mesh_delivery::PeerCtx;
38use crate::mesh_runtime::{MeshRuntime, PeerControlSink, REVOKE_TOPIC};
39use crate::revocation::{RevocationKind, RevocationList};
40use crate::workload_pkg::{KeyOrigin, TrustAnchors};
41
42/// Domain tag signed before a revocation payload.
43pub const REVOKE_DOMAIN: &[u8] = b"weftos.artifact_revocation.v1\0";
44/// Largest notice payload accepted, in bytes.
45pub const MAX_NOTICE_BYTES: usize = 4096;
46
47/// What is revoked and why.
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct RevocationNotice {
50    /// Package id, signer key or artifact hash.
51    pub kind: RevocationKind,
52    /// Canonical id for that kind.
53    pub id: String,
54    /// Human-readable reason.
55    pub reason: String,
56    /// Issue time, unix seconds.
57    pub issued_at: u64,
58}
59
60/// A notice as it travels: the exact signed JSON, the key and the signature.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub struct SignedRevocation {
63    /// JSON of the [`RevocationNotice`], exactly as signed.
64    pub payload: String,
65    /// Ed25519 public key (32 bytes).
66    pub public_key: Vec<u8>,
67    /// Ed25519 signature (64 bytes) over [`REVOKE_DOMAIN`] || payload.
68    pub signature: Vec<u8>,
69}
70
71/// Why a notice was refused.
72#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
73pub enum NoticeError {
74    /// Malformed, oversize or invalid id.
75    #[error("malformed revocation notice: {0}")]
76    Malformed(String),
77    /// Signature does not verify.
78    #[error("revocation signature does not verify")]
79    BadSignature,
80    /// Signer is not a pinned operator or WeftOS key.
81    #[error("revocation signer is not a pinned operator key")]
82    UnauthorizedSigner,
83    /// The signing key has itself been revoked.
84    #[error("revocation signer key is revoked")]
85    SignerRevoked,
86    /// Too many notices arrived too fast.
87    #[error("revocation notices are arriving too fast")]
88    RateLimited,
89}
90
91fn signed_bytes(payload: &str) -> Vec<u8> {
92    let mut v = REVOKE_DOMAIN.to_vec();
93    v.extend_from_slice(payload.as_bytes());
94    v
95}
96
97/// Sign a revocation notice.
98pub fn sign_revocation(
99    kind: RevocationKind,
100    id: &str,
101    reason: &str,
102    now: u64,
103    key: &SigningKey,
104) -> Result<SignedRevocation, NoticeError> {
105    let id = kind
106        .normalize(id)
107        .map_err(|e| NoticeError::Malformed(e.to_string()))?;
108    let payload = serde_json::to_string(&RevocationNotice {
109        kind,
110        id,
111        reason: reason.chars().take(256).collect(),
112        issued_at: now,
113    })
114    .map_err(|e| NoticeError::Malformed(e.to_string()))?;
115    let sig = key.sign(&signed_bytes(&payload));
116    Ok(SignedRevocation {
117        payload,
118        public_key: key.verifying_key().to_bytes().to_vec(),
119        signature: sig.to_bytes().to_vec(),
120    })
121}
122
123/// Verify `signed` against `anchors` and return the notice.
124pub fn verify_revocation(
125    signed: &SignedRevocation,
126    anchors: &TrustAnchors,
127) -> Result<RevocationNotice, NoticeError> {
128    if signed.payload.len() > MAX_NOTICE_BYTES {
129        return Err(NoticeError::Malformed("payload too large".into()));
130    }
131    let pk: [u8; 32] = signed
132        .public_key
133        .as_slice()
134        .try_into()
135        .map_err(|_| NoticeError::Malformed("public key length".into()))?;
136    let sig: [u8; 64] = signed
137        .signature
138        .as_slice()
139        .try_into()
140        .map_err(|_| NoticeError::Malformed("signature length".into()))?;
141    let vk = VerifyingKey::from_bytes(&pk).map_err(|_| NoticeError::BadSignature)?;
142    vk.verify_strict(&signed_bytes(&signed.payload), &Signature::from_bytes(&sig))
143        .map_err(|_| NoticeError::BadSignature)?;
144    let pinned = anchors.signer(&pk).ok_or(NoticeError::UnauthorizedSigner)?;
145    if !matches!(pinned.origin, KeyOrigin::Operator | KeyOrigin::Weftos) {
146        return Err(NoticeError::UnauthorizedSigner);
147    }
148    let notice: RevocationNotice = serde_json::from_str(&signed.payload)
149        .map_err(|e| NoticeError::Malformed(e.to_string()))?;
150    let canonical = notice
151        .kind
152        .normalize(&notice.id)
153        .map_err(|e| NoticeError::Malformed(e.to_string()))?;
154    if canonical != notice.id {
155        return Err(NoticeError::Malformed("id is not canonical".into()));
156    }
157    Ok(notice)
158}
159
160/// Receives, applies and forwards revocation notices for one node.
161pub struct RevocationExchange {
162    ex: Arc<ArtifactExchange>,
163    list: Arc<RevocationList>,
164    anchors: TrustAnchors,
165    runtime: Arc<MeshRuntime>,
166    me: std::sync::Weak<Self>,
167    /// Token buckets for notices that reach signature verification, one per
168    /// connection (0 = callers with no connection): `(tokens, last refill)`.
169    buckets: dashmap::DashMap<u64, (f64, std::time::Instant)>,
170    /// Notices already applied (hash of payload, key, signature), oldest first.
171    seen: std::sync::Mutex<Seen>,
172    /// Run after a notice that was new here has been applied.
173    on_applied: std::sync::OnceLock<AppliedHook>,
174    /// Verified notices kept for replay to peers that rejoin.
175    log: std::sync::Mutex<log::NoticeLog>,
176    /// Replay bookkeeping per peer (in flight, cooldown, cancel).
177    replays: std::sync::Mutex<log::Replays>,
178    replays_started: std::sync::atomic::AtomicU64,
179}
180
181/// Called with each notice that was new here, after it was recorded and the
182/// held artifacts were swept. Runs on the receiving task: it must not block
183/// (spawn what needs time).
184pub type AppliedHook = Arc<dyn Fn(&RevocationNotice) + Send + Sync>;
185
186/// Applied-notice keys: a set for lookup, a queue for oldest-first eviction.
187type Seen = (std::collections::HashSet<[u8; 32]>, std::collections::VecDeque<[u8; 32]>);
188
189/// Connections whose buckets are remembered at once.
190const MAX_TRACKED_CONNS: usize = 1024;
191/// Applied notices remembered for de-duplication.
192const MAX_SEEN: usize = 4096;
193
194/// Notices accepted per second (burst [`NOTICE_BURST`]).
195pub const NOTICES_PER_SEC: f64 = 2.0;
196/// Burst of notices accepted at once.
197pub const NOTICE_BURST: f64 = 10.0;
198
199impl RevocationExchange {
200    /// Exchange applying notices to `ex`, recording them in `list` and
201    /// trusting `anchors`. Installs itself on `runtime` and gives `ex` the
202    /// list.
203    pub fn start(
204        ex: Arc<ArtifactExchange>,
205        list: Arc<RevocationList>,
206        anchors: TrustAnchors,
207        runtime: Arc<MeshRuntime>,
208    ) -> Arc<Self> {
209        ex.set_revocations(list.clone());
210        let me = Arc::new_cyclic(|w| Self {
211            ex,
212            list,
213            anchors,
214            runtime: runtime.clone(),
215            me: w.clone(),
216            buckets: dashmap::DashMap::new(),
217            seen: Default::default(),
218            on_applied: std::sync::OnceLock::new(),
219            log: Default::default(),
220            replays: Default::default(),
221            replays_started: Default::default(),
222        });
223        runtime.set_control_sink(REVOKE_TOPIC, me.clone());
224        Self::spawn_rejoin_replay(&me);
225        me
226    }
227
228    /// Run `hook` after every notice that was new here is applied (first
229    /// call wins; returns whether this one did). Covers a notice from a peer
230    /// and the operator's own [`Self::issue`].
231    pub fn set_on_applied(&self, hook: AppliedHook) -> bool {
232        self.on_applied.set(hook).is_ok()
233    }
234
235    /// Spend one token from `conn`'s bucket.
236    fn take_token(&self, conn: u64) -> bool {
237        if self.buckets.len() >= MAX_TRACKED_CONNS && !self.buckets.contains_key(&conn) {
238            let oldest = self.buckets.iter().min_by_key(|b| b.1).map(|b| *b.key());
239            if let Some(k) = oldest {
240                self.buckets.remove(&k);
241            }
242        }
243        let now = std::time::Instant::now();
244        let mut b = self.buckets.entry(conn).or_insert((NOTICE_BURST, now));
245        let dt = now.duration_since(b.1).as_secs_f64();
246        b.1 = now;
247        b.0 = (b.0 + dt * NOTICES_PER_SEC).min(NOTICE_BURST);
248        if b.0 < 1.0 {
249            return false;
250        }
251        b.0 -= 1.0;
252        true
253    }
254
255    fn notice_key(signed: &SignedRevocation) -> [u8; 32] {
256        let mut h = blake3::Hasher::new();
257        h.update(signed.payload.as_bytes());
258        h.update(&signed.public_key);
259        h.update(&signed.signature);
260        *h.finalize().as_bytes()
261    }
262
263    /// Record and apply a notice from a caller with no connection (budget 0).
264    /// Returns whether it was new here; a notice already applied is not
265    /// swept again.
266    pub fn accept(&self, signed: &SignedRevocation) -> Result<bool, NoticeError> {
267        self.accept_from(signed, Some(0))
268    }
269
270    /// As [`Self::accept`] for a notice from connection `conn`. The cheap
271    /// checks come first and cost no budget: size, a pinned operator/WeftOS
272    /// key, a signer that is not itself revoked, a notice not already
273    /// applied. Only then is a token spent (from `conn`'s own bucket, so
274    /// junk on one connection cannot starve notices on another), before the
275    /// signature verify. `None` skips the budget (the operator's own `issue`).
276    fn accept_from(
277        &self,
278        signed: &SignedRevocation,
279        conn: Option<u64>,
280    ) -> Result<bool, NoticeError> {
281        if signed.payload.len() > MAX_NOTICE_BYTES {
282            return Err(NoticeError::Malformed("payload too large".into()));
283        }
284        let pk: [u8; 32] = signed
285            .public_key
286            .as_slice()
287            .try_into()
288            .map_err(|_| NoticeError::Malformed("public key length".into()))?;
289        match self.anchors.signer(&pk) {
290            Some(k) if matches!(k.origin, KeyOrigin::Operator | KeyOrigin::Weftos) => {}
291            _ => return Err(NoticeError::UnauthorizedSigner),
292        }
293        if self
294            .list
295            .is_subject_revoked(RevocationKind::SignerKey, &crate::workload_pkg::codec::hex_encode(&pk))
296        {
297            return Err(NoticeError::SignerRevoked);
298        }
299        let key = Self::notice_key(signed);
300        if self.seen.lock().unwrap_or_else(|p| p.into_inner()).0.contains(&key) {
301            return Ok(false);
302        }
303        if let Some(conn) = conn
304            && !self.take_token(conn)
305        {
306            return Err(NoticeError::RateLimited);
307        }
308        let n = verify_revocation(signed, &self.anchors)?;
309        let by = format!(
310            "mesh:{}",
311            crate::workload_pkg::codec::hex_encode(&pk[..4])
312        );
313        let chain = self.ex.chain.clone();
314        let sink = chain.as_ref().map(|cm| {
315            let cm = cm.clone();
316            move |k: &str, p: serde_json::Value| {
317                cm.append(crate::workload_governance::gate::CHAIN_SOURCE, k, Some(p));
318            }
319        });
320        let new = match self.list.revoke_audited(
321            n.kind,
322            &n.id,
323            &n.reason,
324            &by,
325            sink.as_ref().map(|f| f as &dyn Fn(&str, serde_json::Value)),
326        ) {
327            Ok(new) => new,
328            // The entry is held in memory (fail-closed) and was chained: it
329            // is in force, so apply and forward it; the disk write is the
330            // only thing that failed.
331            Err(crate::revocation::RevocationError::Persist(e)) => {
332                tracing::error!(error = %e, "revocation applied but not persisted");
333                true
334            }
335            Err(e) => return Err(NoticeError::Malformed(e.to_string())),
336        };
337        {
338            let mut g = self.seen.lock().unwrap_or_else(|p| p.into_inner());
339            if g.0.insert(key) {
340                g.1.push_back(key);
341                if g.1.len() > MAX_SEEN
342                    && let Some(old) = g.1.pop_front()
343                {
344                    g.0.remove(&old);
345                }
346            }
347        }
348        self.log_notice(signed);
349        if new {
350            self.ex.apply_revocations();
351            if let Some(hook) = self.on_applied.get() {
352                hook(&n);
353            }
354        }
355        Ok(new)
356    }
357
358    /// Revoke here (the operator's own node) and send the notice to every
359    /// peer. `signed` must verify against this node's anchors. Not rate
360    /// limited: peers' traffic cannot starve the operator.
361    pub async fn issue(&self, signed: SignedRevocation) -> Result<bool, NoticeError> {
362        let new = self.accept_from(&signed, None)?;
363        self.flood(&signed, None).await;
364        Ok(new)
365    }
366
367    async fn flood(&self, signed: &SignedRevocation, except: Option<&str>) {
368        let Ok(value) = serde_json::to_value(signed) else {
369            return;
370        };
371        for peer in self.runtime.peer_ids() {
372            if Some(peer.as_str()) == except {
373                continue;
374            }
375            let msg = KernelMessage::new(
376                0,
377                MessageTarget::Topic(REVOKE_TOPIC.to_string()),
378                MessagePayload::Json(value.clone()),
379            );
380            if let Err(e) = self.runtime.route_to_remote(&peer, msg).await {
381                tracing::debug!(peer, error = %e, "revocation forward failed");
382            }
383        }
384    }
385}
386
387impl PeerControlSink for RevocationExchange {
388    fn on_peer_control(
389        &self,
390        ctx: &PeerCtx,
391        conn: u64,
392        payload: &serde_json::Value,
393    ) -> Vec<serde_json::Value> {
394        let signed: SignedRevocation = match serde_json::from_value(payload.clone()) {
395            Ok(s) => s,
396            Err(e) => {
397                tracing::warn!(peer = %ctx.peer_id, error = %e, "malformed revocation message");
398                return Vec::new();
399            }
400        };
401        match self.accept_from(&signed, Some(conn)) {
402            Ok(true) => {
403                // New here: pass it on. Done on a task: the sink is sync and
404                // the runtime is mid-dispatch.
405                if let Some(me) = self.me.upgrade() {
406                    let from = ctx.peer_id.clone();
407                    tokio::spawn(async move { me.flood(&signed, Some(&from)).await });
408                }
409            }
410            Ok(false) => {}
411            Err(e) => {
412                tracing::warn!(peer = %ctx.peer_id, verified = ctx.node_verified, error = %e,
413                    "revocation notice refused");
414            }
415        }
416        Vec::new()
417    }
418}
419
420#[path = "mesh_swarm_revoke_log.rs"]
421mod log;
422pub use log::{MAX_LOGGED, REPLAY_COOLDOWN};
423
424#[cfg(test)]
425#[path = "mesh_swarm_revoke_tests.rs"]
426mod tests;
clawft_kernel/mesh_swarm_revoke.rs

clawft_kernel/
mesh_swarm_revoke.rs
