clawft_kernel/licence/
relay.rs

1//! The steward's checkout relay (ADR-106 section 4, steps 2, 3 and 5).
2//!
3//! A `mesh.cog.checkout` request is accepted only from the steward's own
4//! kernel or a verified `node` peer (the service-stamped `AdmittedPeer`),
5//! after the governance gate permits `cog.checkout`. Concurrent requests for
6//! one (cog, version, arch) share a single relay call. The steward forwards
7//! to `weft-licence`, verifies the returned grant against the bound key and
8//! the mesh id, checks the bytes against the signed sha256 and BLAKE3, seeds
9//! them, registers the grant (which makes the bytes shareable through
10//! [`crate::mesh_artifact::ArtifactExchange::grant_checkout`]), chains the
11//! outcome and floods the grant through a [`GrantFlood`].
12
13use std::collections::HashMap;
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::{Arc, Mutex};
16use std::time::{Duration, Instant};
17
18use async_trait::async_trait;
19use tokio::sync::{OnceCell, Semaphore};
20
21use super::client::{CheckoutWire, LicenceClient, LicenceClientError};
22use super::{
23    CheckoutGrant, CheckoutGrantStore, LicenceError, Outcome, SignedGrant, sha256_hex, verify_grant,
24};
25use crate::chain::ChainManager;
26use crate::gate::{GateBackend, GateDecision};
27use crate::mesh_admit::PeerClass;
28use crate::mesh_artifact::ArtifactExchange;
29use crate::mesh_artifact_types::ArtifactKey;
30use crate::mesh_delivery::PeerCtx;
31use crate::workload_pkg::codec::{hex_decode_exact, hex_encode};
32
33/// Gate action asked for every checkout request.
34pub const GATE_ACTION: &str = "cog.checkout";
35/// Chain event: a grant was obtained and seeded.
36pub const EVENT_KIND_CHECKOUT_GRANTED: &str = "cog.checkout.granted";
37/// Chain event: a checkout request was refused or failed.
38pub const EVENT_KIND_CHECKOUT_REFUSED: &str = "cog.checkout.refused";
39
40/// Who is asking.
41#[derive(Debug, Clone, Copy)]
42pub enum CheckoutCaller<'a> {
43    /// This node's own kernel (internal callers; not rate limited).
44    Kernel,
45    /// An RPC caller on this node, named by its principal (a project, a
46    /// token, or the local operator). Charged to its own budget and to the
47    /// node-wide local budget; the gate is asked as this principal.
48    Local(&'a str),
49    /// A delivery from a mesh peer; the context is built from the service
50    /// origin, so `node_verified` means `AdmittedPeer`.
51    Peer(&'a PeerCtx),
52}
53
54/// Why a checkout was refused or failed. [`CheckoutRefusal::code`] is the
55/// stable code a requester sees.
56#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
57pub enum CheckoutRefusal {
58    /// The caller is not the steward's kernel or a verified node peer
59    /// (`LocalTenant`, `Unadmitted`, a leaf, a legacy peer).
60    #[error("caller is not an admitted node")]
61    NotAdmitted,
62    /// The governance gate did not permit `cog.checkout`.
63    #[error("governance gate refused: {0}")]
64    GateDenied(String),
65    /// The request is not well formed.
66    #[error("bad request: {0}")]
67    BadRequest(String),
68    /// `weft-licence` refused with a stable code.
69    #[error("licence refused: {0}")]
70    Licence(String),
71    /// The link to `weft-licence` failed.
72    #[error("licence link failed: {0}")]
73    LicenceLink(String),
74    /// The returned grant did not verify.
75    #[error("returned grant did not verify: {0}")]
76    BadGrant(String),
77    /// The Seed's clock is ahead of this node's by more than the skew allowance.
78    #[error("the Seed's clock is ahead of this node's")]
79    SeedClockSkew,
80    /// The returned bytes do not match the signed sizes or hashes.
81    #[error("artifact does not match the grant: {0}")]
82    ArtifactMismatch(String),
83    /// This member asked too often (the steward's per-member budget).
84    #[error("too many checkout requests from this member")]
85    RateLimited,
86    /// The steward has too many checkouts in flight.
87    #[error("the steward is busy")]
88    Busy,
89    /// The node could not store the bytes.
90    #[error("could not store the artifact: {0}")]
91    Storage(String),
92}
93
94impl CheckoutRefusal {
95    /// The stable wire code.
96    pub fn code(&self) -> String {
97        match self {
98            Self::NotAdmitted => "not_admitted".into(),
99            Self::GateDenied(_) => "gate_denied".into(),
100            Self::BadRequest(_) => "bad_request".into(),
101            Self::Licence(c) => c.clone(),
102            Self::LicenceLink(_) => "licence_unreachable".into(),
103            Self::BadGrant(_) => "bad_grant".into(),
104            Self::SeedClockSkew => "seed_clock_skew".into(),
105            Self::ArtifactMismatch(_) => "artifact_mismatch".into(),
106            Self::Storage(_) => "storage_failed".into(),
107            Self::RateLimited => "rate_limited".into(),
108            Self::Busy => "busy".into(),
109        }
110    }
111}
112
113impl From<LicenceClientError> for CheckoutRefusal {
114    fn from(e: LicenceClientError) -> Self {
115        match e {
116            LicenceClientError::Refused { code, .. } => Self::Licence(code),
117            LicenceClientError::Transport(w) => Self::LicenceLink(w),
118            LicenceClientError::BadResponse(w) => Self::BadGrant(w),
119        }
120    }
121}
122
123/// Spreads a new grant to the mesh. Phase 1b replaces the implementation with
124/// the `mesh.cog.grant` flood and sync; the relay only calls this.
125#[async_trait]
126pub trait GrantFlood: Send + Sync + 'static {
127    /// Flood `grant` to the admitted peers. Best effort: the requester also
128    /// gets the grant in its reply, and sync catches the rest up.
129    async fn flood(&self, grant: &SignedGrant);
130}
131
132/// Floods nothing.
133#[derive(Debug, Default)]
134pub struct NoFlood;
135
136#[async_trait]
137impl GrantFlood for NoFlood {
138    async fn flood(&self, _: &SignedGrant) {}
139}
140
141/// Register `signed` in `store` and, when accepted, make the bytes of every
142/// currently valid grant for the cog version shareable in `exchange`. This is
143/// the member side of a grant arriving (reply, flood or sync).
144pub fn install_grant(
145    store: &CheckoutGrantStore,
146    exchange: &ArtifactExchange,
147    signed: &SignedGrant,
148) -> Result<Outcome, LicenceError> {
149    let out = store.accept_grant(signed)?;
150    // Accepted, so the payload verified and is canonical JSON.
151    let g: CheckoutGrant = serde_json::from_str(&signed.payload)
152        .map_err(|e| LicenceError::Malformed(e.to_string()))?;
153    for v in store.verified_grants() {
154        if v.grant().cog_id == g.cog_id && v.grant().version == g.version {
155            exchange.grant_checkout(&v);
156        }
157    }
158    Ok(out)
159}
160
161/// Limits on what members can cost the steward (ADR-106 section 6): a request
162/// budget per member, and a bound on checkouts running at once. Over-limit
163/// refusals are not chained, so they cannot grow the chain either.
164#[derive(Debug, Clone)]
165pub struct RelayLimits {
166    /// Requests per member per window.
167    pub per_peer: u32,
168    /// The window.
169    pub window: Duration,
170    /// Checkouts in flight at once.
171    pub concurrent: usize,
172    /// Members remembered at once; the oldest windows are pruned.
173    pub max_peers: usize,
174    /// Requests per local principal per window.
175    pub per_local: u32,
176    /// Requests from all local principals together per window.
177    pub local_total: u32,
178}
179
180impl Default for RelayLimits {
181    fn default() -> Self {
182        Self {
183            per_peer: 5,
184            window: Duration::from_secs(60),
185            concurrent: 8,
186            max_peers: 1024,
187            per_local: 5,
188            local_total: 20,
189        }
190    }
191}
192
193type Key = (String, String, String);
194type Shared = Arc<OnceCell<Result<SignedGrant, CheckoutRefusal>>>;
195
196/// The relay.
197pub struct CheckoutRelay {
198    store: Arc<CheckoutGrantStore>,
199    exchange: Arc<ArtifactExchange>,
200    client: Arc<dyn LicenceClient>,
201    gate: Arc<dyn GateBackend>,
202    flood: Arc<dyn GrantFlood>,
203    chain: Option<Arc<ChainManager>>,
204    inflight: Mutex<HashMap<Key, Shared>>,
205    limits: RelayLimits,
206    buckets: Mutex<HashMap<String, (Instant, u32)>>,
207    local_total: Mutex<(Instant, u32)>,
208    slots: Semaphore,
209}
210
211impl CheckoutRelay {
212    /// A relay. The gate is required: with no gate nothing is permitted.
213    pub fn new(
214        store: Arc<CheckoutGrantStore>,
215        exchange: Arc<ArtifactExchange>,
216        client: Arc<dyn LicenceClient>,
217        gate: Arc<dyn GateBackend>,
218        flood: Arc<dyn GrantFlood>,
219        chain: Option<Arc<ChainManager>>,
220    ) -> Self {
221        let limits = RelayLimits::default();
222        Self {
223            store,
224            exchange,
225            client,
226            gate,
227            flood,
228            chain,
229            inflight: Mutex::default(),
230            slots: Semaphore::new(limits.concurrent),
231            buckets: Mutex::default(),
232            local_total: Mutex::new((Instant::now(), 0)),
233            limits,
234        }
235    }
236
237    /// Replace the limits (before the relay is shared).
238    pub fn with_limits(mut self, limits: RelayLimits) -> Self {
239        self.slots = Semaphore::new(limits.concurrent);
240        self.limits = limits;
241        self
242    }
243
244    /// Charge one request to the node-wide local budget.
245    fn allow_local_total(&self) -> bool {
246        let now = Instant::now();
247        let mut t = self.local_total.lock().unwrap_or_else(|p| p.into_inner());
248        if now.duration_since(t.0) >= self.limits.window {
249            *t = (now, 0);
250        }
251        if t.1 >= self.limits.local_total {
252            return false;
253        }
254        t.1 += 1;
255        true
256    }
257
258    /// Charge one request to `peer`; false when over its budget.
259    fn allow_peer(&self, peer: &str) -> bool {
260        self.allow_key(peer, self.limits.per_peer)
261    }
262
263    fn allow_key(&self, peer: &str, budget: u32) -> bool {
264        let now = Instant::now();
265        let mut m = self.buckets.lock().unwrap_or_else(|p| p.into_inner());
266        if m.len() >= self.limits.max_peers && !m.contains_key(peer) {
267            let w = self.limits.window;
268            m.retain(|_, (start, _)| now.duration_since(*start) < w);
269            if m.len() >= self.limits.max_peers {
270                return false;
271            }
272        }
273        let e = m.entry(peer.to_owned()).or_insert((now, 0));
274        if now.duration_since(e.0) >= self.limits.window {
275            *e = (now, 0);
276        }
277        if e.1 >= budget {
278            return false;
279        }
280        e.1 += 1;
281        true
282    }
283
284    /// Handle one checkout request. Admission, then the gate, then the merged
285    /// relay call.
286    pub async fn handle(
287        &self,
288        caller: CheckoutCaller<'_>,
289        req: &CheckoutWire,
290    ) -> Result<SignedGrant, CheckoutRefusal> {
291        let who = match caller {
292            CheckoutCaller::Kernel => "kernel".to_owned(),
293            CheckoutCaller::Local(p) => format!("local:{p}"),
294            CheckoutCaller::Peer(p) if p.node_verified && p.class == PeerClass::Node => {
295                p.peer_id.clone()
296            }
297            CheckoutCaller::Peer(_) => {
298                let e = CheckoutRefusal::NotAdmitted;
299                self.chain_refused(req, "unadmitted", &e);
300                return Err(e);
301            }
302        };
303        // Limits come before the gate and the chain: a flood of requests
304        // costs a counter, not a chain event.
305        let allowed = match caller {
306            CheckoutCaller::Kernel => true,
307            CheckoutCaller::Peer(_) => self.allow_peer(&who),
308            // Its own budget first, so one principal cannot use up the node's.
309            CheckoutCaller::Local(_) => self.allow_key(&who, self.limits.per_local) && self.allow_local_total(),
310        };
311        if !allowed {
312            return Err(CheckoutRefusal::RateLimited);
313        }
314        let Ok(_slot) = self.slots.try_acquire() else {
315            return Err(CheckoutRefusal::Busy);
316        };
317        let r = self.checked(&who, req).await;
318        if let Err(e) = &r {
319            self.chain_refused(req, &who, e);
320        }
321        r
322    }
323
324    async fn checked(&self, who: &str, req: &CheckoutWire) -> Result<SignedGrant, CheckoutRefusal> {
325        req.validate().map_err(CheckoutRefusal::BadRequest)?;
326        let ctx = serde_json::json!({
327            "cog_id": req.cog_id, "version": req.version, "arch": req.arch,
328        });
329        match self.gate.check(who, GATE_ACTION, &ctx) {
330            GateDecision::Permit { .. } => {}
331            GateDecision::Defer { reason } | GateDecision::Deny { reason, .. } => {
332                return Err(CheckoutRefusal::GateDenied(reason));
333            }
334        }
335        let key = (req.cog_id.clone(), req.version.clone(), req.arch.clone());
336        let cell: Shared = {
337            let mut m = self.inflight.lock().unwrap_or_else(|p| p.into_inner());
338            m.entry(key.clone()).or_default().clone()
339        };
340        let ran = AtomicBool::new(false);
341        let out = cell
342            .get_or_init(|| async {
343                ran.store(true, Ordering::Relaxed);
344                let r = self.relay(req).await;
345                // Later requests start a fresh relay call; waiters on this
346                // cell still read the stored result.
347                let mut m = self.inflight.lock().unwrap_or_else(|p| p.into_inner());
348                if m.get(&key).is_some_and(|c| Arc::ptr_eq(c, &cell)) {
349                    m.remove(&key);
350                }
351                r
352            })
353            .await;
354        // Every caller is chained as a grantee, the merged ones too.
355        if let Ok(signed) = out
356            && let Ok(g) = serde_json::from_str::<CheckoutGrant>(&signed.payload)
357        {
358            let b3 = g.artifact(&req.arch).map(|a| a.blake3.clone()).unwrap_or_default();
359            self.chain_granted(who, req, &g, &b3, !ran.load(Ordering::Relaxed));
360        }
361        out.clone()
362    }
363
364    async fn relay(&self, req: &CheckoutWire) -> Result<SignedGrant, CheckoutRefusal> {
365        let signed = self.client.checkout(req).await?;
366        let binding = self
367            .store
368            .active_binding()
369            .ok_or_else(|| CheckoutRefusal::BadGrant("no binding".into()))?;
370        let pk = hex_decode_exact::<32>(&binding.grant_pubkey)
371            .ok_or_else(|| CheckoutRefusal::BadGrant("bound grant key".into()))?;
372        let mesh = self
373            .store
374            .local_mesh_id()
375            .get()
376            .ok_or_else(|| CheckoutRefusal::BadGrant("no local mesh id".into()))?;
377        let g = verify_grant(&signed, &pk, &mesh).map_err(|e| CheckoutRefusal::BadGrant(e.to_string()))?;
378        if g.cog_id != req.cog_id || (req.version != "latest" && g.version != req.version) {
379            return Err(CheckoutRefusal::BadGrant("grant is for another cog or version".into()));
380        }
381        let art = g
382            .artifact(&req.arch)
383            .ok_or_else(|| CheckoutRefusal::BadGrant(format!("grant has no {} artifact", req.arch)))?
384            .clone();
385        let hash = hex_decode_exact::<32>(&art.blake3)
386            .ok_or_else(|| CheckoutRefusal::BadGrant("artifact hash".into()))?;
387        match self.exchange.resolve(&ArtifactKey::Content(hash)) {
388            // Held already: the BLAKE3 was checked on the way in, but the grant
389            // also commits to the registry sha256 of these bytes.
390            Some(d) => self.check_held_sha256(&d, &art.sha256)?,
391            None => self.fetch_and_seed(&art.blake3, &art.sha256, art.size).await?,
392        }
393        install_grant(&self.store, &self.exchange, &signed).map_err(|e| match e {
394            LicenceError::NotYetValid => CheckoutRefusal::SeedClockSkew,
395            e => CheckoutRefusal::BadGrant(e.to_string()),
396        })?;
397        self.flood.flood(&signed).await;
398        Ok(signed)
399    }
400
401    fn check_held_sha256(
402        &self,
403        d: &crate::mesh_artifact_types::ArtifactDescriptor,
404        sha256: &str,
405    ) -> Result<(), CheckoutRefusal> {
406        use sha2::{Digest, Sha256};
407        let mut h = Sha256::new();
408        self.exchange
409            .read_to(&d.id(), &mut |b| {
410                h.update(b);
411                Ok(())
412            })
413            .map_err(|e| CheckoutRefusal::Storage(e.to_string()))?;
414        if hex_encode(&h.finalize()) != sha256 {
415            return Err(CheckoutRefusal::ArtifactMismatch("sha256 of the held bytes".into()));
416        }
417        Ok(())
418    }
419
420    async fn fetch_and_seed(&self, blake3: &str, sha256: &str, size: u64) -> Result<(), CheckoutRefusal> {
421        let limit = self.exchange.config().max_artifact_bytes;
422        if size > limit {
423            return Err(CheckoutRefusal::ArtifactMismatch(format!("{size} bytes is over the limit")));
424        }
425        let bytes = self.client.artifact(blake3, size).await?;
426        if bytes.len() as u64 != size {
427            return Err(CheckoutRefusal::ArtifactMismatch("size".into()));
428        }
429        if sha256_hex(&bytes) != sha256 {
430            return Err(CheckoutRefusal::ArtifactMismatch("sha256".into()));
431        }
432        if hex_encode(blake3::hash(&bytes).as_bytes()) != blake3 {
433            return Err(CheckoutRefusal::ArtifactMismatch("blake3".into()));
434        }
435        self.exchange.seed_bytes(&bytes).map_err(|e| CheckoutRefusal::Storage(e.to_string()))?;
436        Ok(())
437    }
438
439    fn chain_granted(&self, who: &str, req: &CheckoutWire, g: &CheckoutGrant, blake3: &str, merged: bool) {
440        if let Some(cm) = &self.chain {
441            cm.append(
442                "licence",
443                EVENT_KIND_CHECKOUT_GRANTED,
444                Some(serde_json::json!({
445                    "cog_id": g.cog_id, "version": g.version, "arch": req.arch,
446                    "grant_id": g.grant_id, "seq": g.seq, "blake3": blake3, "requester": who,
447                    "merged": merged,
448                })),
449            );
450        }
451    }
452
453    fn chain_refused(&self, req: &CheckoutWire, who: &str, e: &CheckoutRefusal) {
454        if let Some(cm) = &self.chain {
455            cm.append(
456                "licence",
457                EVENT_KIND_CHECKOUT_REFUSED,
458                Some(serde_json::json!({
459                    "cog_id": req.cog_id, "version": req.version, "arch": req.arch,
460                    "code": e.code(), "requester": who,
461                })),
462            );
463        }
464    }
465}
clawft_kernel/licence/relay.rs

clawft_kernel/licence/
relay.rs
