clawft_kernel/licence/
exchange.rs

1//! Floods of the licence records across the mesh (ADR-106 phase 1b).
2//!
3//! [`LicenceExchange`] carries the Seed binding on `mesh.cog.binding` and
4//! checkout grants plus operator approvals on `mesh.cog.grant`, with the same
5//! discipline as [`crate::mesh_swarm_revoke::RevocationExchange`]:
6//!
7//! - **Cheap first, and free.** Size, a signer that is a pinned operator key
8//!   (binding, approval) or the bound grant key (grant), the record naming
9//!   this node's mesh, and a `seq` (or content key) the store does not
10//!   already hold.
11//! - **Budgeted before the verify.** A record that passes spends a token
12//!   from its own connection's bucket for that record kind, so junk on one
13//!   connection starves neither another connection, another kind, nor the
14//!   revocation floods (those live in a different exchange with its own
15//!   buckets). The operator's own `issue_*` is exempt.
16//! - **Verified by the stores.** [`CheckoutGrantStore`] and
17//!   [`ApprovalStore`] do the signature and rule checks; this module adds
18//!   none of its own.
19//! - **Forwarded once.** A record that was new here goes to every other
20//!   peer; the seen-set is written only after a record verified.
21//! - **Deferred, not refused.** A grant ahead of the local clock
22//!   (`NotYetValid`) is neither forwarded nor remembered: catch-up sync
23//!   brings it again later.
24//!
25//! The exchange is inert while the local mesh id is unset.
26
27use std::sync::{Arc, Mutex, Weak};
28use std::time::Instant;
29
30use dashmap::DashMap;
31use ed25519_dalek::SigningKey;
32use serde::Deserialize;
33
34use super::binding::require_operator;
35pub use super::exchange_types::{
36    CtxAdmission, ExchangeError, GrantMsg, LicenceExchangeConfig, LicenceExchangeParts,
37    PeerAdmission, PostureFn, Receipt, Spend,
38};
39use super::exchange_types::{Buckets, MAX_SEEN, RECORDS_PER_SEC, RECORD_BURST};
40pub(super) use super::exchange_types::Budget;
41use super::{
42    Approval, ApprovalStore, BindState, BindingRecord, CheckoutGrant,
43    CheckoutGrantStore, LicenceError, LicenceEventSink, NoExtraChecks, Outcome, SignedApproval,
44    SignedBinding, SignedEnvelope, SignedGrant, envelope_key, sign_binding,
45};
46use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
47use crate::mesh_delivery::PeerCtx;
48use super::links::LicenceLinks;
49use crate::mesh_runtime::{COG_BINDING_TOPIC, COG_GRANT_TOPIC, COG_SYNC_TOPIC, PeerControlSink};
50use crate::workload_pkg::TrustAnchors;
51
52type Seen = (std::collections::HashSet<[u8; 32]>, std::collections::VecDeque<[u8; 32]>);
53
54/// Receives, applies, forwards and syncs the licence records of one node.
55pub struct LicenceExchange {
56    pub(super) store: Arc<CheckoutGrantStore>,
57    pub(super) approvals: Arc<ApprovalStore>,
58    pub(super) anchors: Arc<TrustAnchors>,
59    pub(super) runtime: Arc<dyn LicenceLinks>,
60    pub(super) posture: PostureFn,
61    pub(super) admission: Arc<dyn PeerAdmission>,
62    pub(super) sink: Arc<dyn LicenceEventSink>,
63    pub(super) config: LicenceExchangeConfig,
64    pub(super) me: Weak<Self>,
65    flood_buckets: Buckets,
66    pub(super) sync_buckets: Buckets,
67    seen: Mutex<Seen>,
68    /// Peers with a sync request in flight.
69    pub(super) pending: DashMap<String, super::exchange_sync::Pending>,
70    /// When a sync was last answered for a peer.
71    pub(super) served: DashMap<String, Instant>,
72    /// Peers banned from sync, and since when.
73    pub(super) banned: DashMap<String, Instant>,
74    /// Unanswered sync requests re-sent to a peer in a row.
75    pub(super) retries: DashMap<String, u8>,
76    /// Open sync sessions served to a peer.
77    pub(super) sessions: DashMap<String, super::exchange_sync::ServeSession>,
78    /// Flood messages sent (a forward or an issue to one peer counts one).
79    pub(super) flooded: std::sync::atomic::AtomicU64,
80}
81
82impl LicenceExchange {
83    /// Start the exchange: install the control sinks on the mesh (the
84    /// kernel runtime or the service links) and begin the periodic and
85    /// on-connect catch-up sync (when inside a tokio runtime).
86    pub fn start(p: LicenceExchangeParts) -> Arc<Self> {
87        let rt = p.runtime.clone();
88        let me = Arc::new_cyclic(|w| Self {
89            store: p.store,
90            approvals: p.approvals,
91            anchors: p.anchors,
92            runtime: p.runtime,
93            posture: p.posture,
94            admission: p.admission,
95            sink: p.sink,
96            config: p.config,
97            me: w.clone(),
98            flood_buckets: Buckets::new(RECORDS_PER_SEC, RECORD_BURST),
99            sync_buckets: Buckets::new(super::exchange_sync::SYNC_ENTRIES_PER_SEC, super::exchange_sync::SYNC_BURST),
100            seen: Default::default(),
101            pending: DashMap::new(),
102            served: DashMap::new(),
103            banned: DashMap::new(),
104            retries: DashMap::new(),
105            sessions: DashMap::new(),
106            flooded: Default::default(),
107        });
108        rt.set_control_sink(COG_BINDING_TOPIC, Arc::new(BindingSink(me.clone())));
109        rt.set_control_sink(COG_GRANT_TOPIC, Arc::new(GrantSink(me.clone())));
110        rt.set_control_sink(COG_SYNC_TOPIC, Arc::new(super::exchange_sync::SyncSink(me.clone())));
111        me.spawn_sync_tasks();
112        me
113    }
114
115    /// The node's approval store (the run gate and `weaver cog checkout
116    /// status` read it).
117    pub fn approvals(&self) -> &Arc<ApprovalStore> {
118        &self.approvals
119    }
120
121    /// The local mesh id, or [`LicenceError::NoLocalMesh`] (the exchange is
122    /// inert while it is unset).
123    pub(super) fn local(&self) -> Result<super::MeshId, LicenceError> {
124        self.store.local_mesh_id().get().ok_or(LicenceError::NoLocalMesh)
125    }
126
127    fn record_key(tag: u8, env: &SignedEnvelope) -> [u8; 32] {
128        let mut h = blake3::Hasher::new();
129        h.update(&[tag]);
130        h.update(env.payload.as_bytes());
131        h.update(env.public_key.as_bytes());
132        h.update(env.signature.as_bytes());
133        *h.finalize().as_bytes()
134    }
135
136    pub(super) fn already_seen(&self, key: &[u8; 32]) -> bool {
137        self.seen.lock().unwrap_or_else(|p| p.into_inner()).0.contains(key)
138    }
139
140    /// Records remembered for de-duplication.
141    #[cfg(test)]
142    pub(super) fn seen_len(&self) -> usize {
143        self.seen.lock().unwrap_or_else(|p| p.into_inner()).0.len()
144    }
145
146    /// A conflict is final for this exact record: remember it so it is not
147    /// verified and chained again.
148    fn remember_conflict(&self, key: [u8; 32], e: &LicenceError) {
149        if matches!(e, LicenceError::Conflict(_)) {
150            self.remember(key);
151        }
152    }
153
154    pub(super) fn remember(&self, key: [u8; 32]) {
155        let mut g = self.seen.lock().unwrap_or_else(|p| p.into_inner());
156        if g.0.insert(key) {
157            g.1.push_back(key);
158            if g.1.len() > MAX_SEEN
159                && let Some(old) = g.1.pop_front()
160            {
161                g.0.remove(&old);
162            }
163        }
164    }
165
166    /// Spend a token before a signature verify (`conn` `None` is exempt).
167    fn budget(&self, kind: Budget, spend: Spend) -> Result<(), ExchangeError> {
168        let ok = match spend {
169            Spend::Exempt => true,
170            Spend::Flood(c) => self.flood_buckets.take(kind, c, 1.0),
171            Spend::Sync(c) => self.sync_buckets.take(Budget::Sync, c, 1.0),
172        };
173        if ok { Ok(()) } else { Err(ExchangeError::RateLimited) }
174    }
175
176    fn receipt(o: Outcome) -> Receipt {
177        match o {
178            Outcome::Applied | Outcome::AppliedUnsaved => Receipt::New,
179            Outcome::Duplicate | Outcome::Ignored => Receipt::Known,
180        }
181    }
182
183    /// Cheap pre-verify filter and apply for a binding. `spend` picks
184    /// the bucket (the operator's own issue); see [`Spend`].
185    pub fn accept_binding(
186        &self,
187        env: &SignedBinding,
188        spend: Spend,
189    ) -> Result<Receipt, ExchangeError> {
190        let local = self.local()?;
191        let pk = envelope_key(env)?;
192        require_operator(&self.anchors, &pk)?;
193        let rec: BindingRecord = serde_json::from_str(&env.payload)
194            .map_err(|e| LicenceError::Malformed(e.to_string()))?;
195        if rec.mesh_id != local.to_hex() {
196            return Err(LicenceError::WrongMesh.into());
197        }
198        if let Some((seq, held)) = self.store.held_signed_binding()
199            && (rec.seq < seq || (rec.seq == seq && held.payload == env.payload))
200        {
201            return Ok(Receipt::Known);
202        }
203        let key = Self::record_key(1, env);
204        if self.already_seen(&key) {
205            return Ok(Receipt::Known);
206        }
207        self.budget(Budget::Binding, spend)?;
208        let out = self
209            .store
210            .accept_binding(env, (self.posture)(), &NoExtraChecks)
211            .inspect_err(|e| self.remember_conflict(key, e))?;
212        self.remember(key);
213        Ok(Self::receipt(out))
214    }
215
216    /// Cheap pre-verify filter and apply for a grant.
217    pub fn accept_grant(
218        &self,
219        env: &SignedGrant,
220        spend: Spend,
221    ) -> Result<Receipt, ExchangeError> {
222        self.local()?;
223        let pk = envelope_key(env)?;
224        let bound = self.store.bound_grant_key().ok_or(LicenceError::NoBinding)?;
225        if super::hex_encode(&pk) != bound {
226            return Err(LicenceError::UntrustedKey.into());
227        }
228        let g: CheckoutGrant = serde_json::from_str(&env.payload)
229            .map_err(|e| LicenceError::Malformed(e.to_string()))?;
230        if let Some((seq, held)) = self.store.held_grant(&g.cog_id, &g.version)
231            && (g.seq < seq || (g.seq == seq && held.payload == env.payload))
232        {
233            return Ok(Receipt::Known);
234        }
235        let key = Self::record_key(2, env);
236        if self.already_seen(&key) {
237            return Ok(Receipt::Known);
238        }
239        self.budget(Budget::Grant, spend)?;
240        match self.store.accept_grant(env) {
241            Ok(o) => {
242                self.remember(key);
243                Ok(Self::receipt(o))
244            }
245            Err(LicenceError::NotYetValid) => Ok(Receipt::Deferred),
246            Err(e) => {
247                self.remember_conflict(key, &e);
248                Err(e.into())
249            }
250        }
251    }
252
253    /// Cheap pre-verify filter and apply for an approval.
254    pub fn accept_approval(
255        &self,
256        env: &SignedApproval,
257        spend: Spend,
258    ) -> Result<Receipt, ExchangeError> {
259        let local = self.local()?;
260        let pk = envelope_key(env)?;
261        require_operator(&self.anchors, &pk)?;
262        let a: Approval = serde_json::from_str(&env.payload)
263            .map_err(|e| LicenceError::Malformed(e.to_string()))?;
264        if a.mesh_id != local.to_hex() {
265            return Err(LicenceError::WrongMesh.into());
266        }
267        if self.approvals.holds(&a.content_key()) {
268            return Ok(Receipt::Known);
269        }
270        let key = Self::record_key(3, env);
271        if self.already_seen(&key) {
272            return Ok(Receipt::Known);
273        }
274        self.budget(Budget::Approval, spend)?;
275        let out = self.approvals.accept(env)?;
276        self.remember(key);
277        Ok(Self::receipt(out))
278    }
279
280    /// Accept a binding here (the operator's own node, not rate limited) and
281    /// send it to every peer. Used for a bind and for an unbind: any node
282    /// with an operator key may issue one, so a steward cannot block it.
283    pub async fn issue_binding(&self, env: SignedBinding) -> Result<Receipt, ExchangeError> {
284        let r = self.accept_binding(&env, Spend::Exempt)?;
285        self.flood_binding(&env, None).await;
286        Ok(r)
287    }
288
289    /// As [`Self::issue_binding`] for a grant.
290    pub async fn issue_grant(&self, env: SignedGrant) -> Result<Receipt, ExchangeError> {
291        let r = self.accept_grant(&env, Spend::Exempt)?;
292        self.flood_grant(&GrantMsg::Grant(env), None).await;
293        Ok(r)
294    }
295
296    /// As [`Self::issue_binding`] for an approval.
297    pub async fn issue_approval(&self, env: SignedApproval) -> Result<Receipt, ExchangeError> {
298        let r = self.accept_approval(&env, Spend::Exempt)?;
299        self.flood_grant(&GrantMsg::Approval(env), None).await;
300        Ok(r)
301    }
302
303    pub(super) async fn flood_binding(&self, env: &SignedBinding, except: Option<&str>) {
304        if let Ok(v) = serde_json::to_value(env) {
305            self.send_all(COG_BINDING_TOPIC, v, except).await;
306        }
307    }
308
309    pub(super) async fn flood_grant(&self, msg: &GrantMsg, except: Option<&str>) {
310        if let Ok(v) = serde_json::to_value(msg) {
311            self.send_all(COG_GRANT_TOPIC, v, except).await;
312        }
313    }
314
315    async fn send_all(&self, topic: &str, value: serde_json::Value, except: Option<&str>) {
316        for peer in self.runtime.peer_ids() {
317            if Some(peer.as_str()) == except || !self.admission.peer_admitted(self.runtime.as_ref(), &peer) {
318                continue;
319            }
320            self.flooded.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
321            self.send_to(&peer, topic, value.clone()).await;
322        }
323    }
324
325    pub(super) async fn send_to(&self, peer: &str, topic: &str, value: serde_json::Value) {
326        let msg = KernelMessage::new(
327            0,
328            MessageTarget::Topic(topic.to_string()),
329            MessagePayload::Json(value),
330        );
331        if let Err(e) = self.runtime.route_to_remote(peer, msg).await {
332            tracing::debug!(peer, topic, error = %e, "licence send failed");
333        }
334    }
335
336    /// Forward a record that was new here, on a task (the sink is sync and
337    /// the runtime is mid-dispatch).
338    pub(super) fn forward(&self, from: &str, what: Forward) {
339        let Some(me) = self.me.upgrade() else { return };
340        let from = from.to_owned();
341        tokio::spawn(async move {
342            match what {
343                Forward::Binding(b) => me.flood_binding(&b, Some(&from)).await,
344                Forward::Grant(m) => me.flood_grant(&m, Some(&from)).await,
345            }
346        });
347    }
348}
349
350/// A record to pass on.
351pub(super) enum Forward {
352    Binding(SignedBinding),
353    Grant(GrantMsg),
354}
355
356/// Sign the unbind of `prev`: same Seed and mesh, `state: unbound`, `seq`
357/// one higher. Everything else, `bound_at` included, is copied from `prev`,
358/// so two operators unbinding the same record at once produce byte-identical
359/// records (Ed25519 is deterministic) instead of an equal-`seq` conflict.
360/// Any node holding an operator key can do this and
361/// [`LicenceExchange::issue_binding`] it.
362pub fn sign_unbind(prev: &BindingRecord, operator: &SigningKey) -> Result<SignedBinding, LicenceError> {
363    let rec = BindingRecord {
364        state: BindState::Unbound,
365        seq: prev.seq.saturating_add(1),
366        ..prev.clone()
367    };
368    sign_binding(&rec, operator)
369}
370
371/// The steward relay floods the grants it obtains through the exchange
372/// (ADR-106 phase 1b over the 1c relay): same topic, budgets and admission
373/// as any other grant flood.
374#[async_trait::async_trait]
375impl super::GrantFlood for LicenceExchange {
376    async fn flood(&self, grant: &SignedGrant) {
377        self.flood_grant(&GrantMsg::Grant(grant.clone()), None).await;
378    }
379}
380
381struct BindingSink(Arc<LicenceExchange>);
382struct GrantSink(Arc<LicenceExchange>);
383
384fn refused(kind: &str, ctx: &PeerCtx, e: &ExchangeError) {
385    if matches!(e, ExchangeError::RateLimited) {
386        tracing::debug!(peer = %ctx.peer_id, "licence {kind} rate limited");
387        return;
388    }
389    tracing::warn!(peer = %ctx.peer_id, verified = ctx.node_verified, error = %e,
390        "licence {kind} refused");
391}
392
393impl PeerControlSink for BindingSink {
394    fn on_peer_control(&self, ctx: &PeerCtx, conn: u64, payload: &serde_json::Value) -> Vec<serde_json::Value> {
395        let Ok(env) = SignedBinding::deserialize(payload) else {
396            tracing::warn!(peer = %ctx.peer_id, "malformed licence binding message");
397            return Vec::new();
398        };
399        match self.0.accept_binding(&env, Spend::Flood(conn)) {
400            Ok(Receipt::New) => self.0.forward(&ctx.peer_id, Forward::Binding(env)),
401            Ok(_) => {}
402            Err(ExchangeError::Licence(LicenceError::NoLocalMesh)) => {}
403            Err(e) => refused("binding", ctx, &e),
404        }
405        Vec::new()
406    }
407}
408
409impl PeerControlSink for GrantSink {
410    fn on_peer_control(&self, ctx: &PeerCtx, conn: u64, payload: &serde_json::Value) -> Vec<serde_json::Value> {
411        let Ok(msg) = GrantMsg::deserialize(payload) else {
412            tracing::warn!(peer = %ctx.peer_id, "malformed licence grant message");
413            return Vec::new();
414        };
415        let r = match &msg {
416            GrantMsg::Grant(g) => self.0.accept_grant(g, Spend::Flood(conn)),
417            GrantMsg::Approval(a) => self.0.accept_approval(a, Spend::Flood(conn)),
418        };
419        match r {
420            Ok(Receipt::New) => self.0.forward(&ctx.peer_id, Forward::Grant(msg)),
421            Ok(_) => {}
422            Err(ExchangeError::Licence(LicenceError::NoLocalMesh)) => {}
423            Err(e) => refused("grant", ctx, &e),
424        }
425        Vec::new()
426    }
427}
clawft_kernel/licence/exchange.rs

clawft_kernel/licence/
exchange.rs
