clawft_kernel/
mesh_admit_gate.rs

1//! Verdict sources and the cryptographic admission gate (P3-K1).
2//!
3//! Split from [`mesh_admit`](crate::mesh_admit), which re-exports
4//! everything here.
5
6use std::collections::{HashMap, VecDeque};
7use std::sync::{Arc, Mutex};
8use std::time::{Duration, Instant};
9
10use async_trait::async_trait;
11use clawft_types::config::MeshAdmissionMode;
12
13use crate::mesh_admit::{
14    Admission, AdmissionGate, AdmitContext, Grant, HelloFailure, PeerClass, PeerLimits, Refusal,
15    VerifiedHello,
16};
17use crate::revocation::RevocationList;
18
19/// Default verdict cache lifetime.
20pub const VERDICT_TTL: Duration = Duration::from_secs(300);
21/// How long a denial is cached (short: an owner who changes their mind
22/// should not wait five minutes).
23pub const DENY_TTL: Duration = Duration::from_secs(5);
24const MAX_CACHE: usize = 1024;
25const WARN_EVERY: Duration = Duration::from_secs(60);
26const MAX_RECORDS: usize = 256;
27
28/// Request to the cluster-owner for a join verdict.
29#[derive(Debug, Clone)]
30pub struct VerdictRequest {
31    /// Verified node id.
32    pub node_id: String,
33    /// Verified key.
34    pub pubkey: [u8; 32],
35    /// Peer class.
36    pub class: PeerClass,
37    /// Platform string.
38    pub platform: String,
39}
40
41/// Verdict on a join request.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub enum Verdict {
44    /// Peer may join as its declared class.
45    Permit,
46    /// Peer may join, served as this class (overrides the self-declared
47    /// class under `enforce`).
48    PermitAs(PeerClass),
49    /// Peer may not join (deferred decisions count as denials).
50    Deny(String),
51    /// No decision could be made (source not ready). Refused like a denial
52    /// but never cached.
53    Unavailable(String),
54}
55
56/// Source of `peer.admit` verdicts. In service mode this is the
57/// cluster-owner daemon; in collapsed mode [`GateVerdictSource`].
58#[async_trait]
59pub trait VerdictSource: Send + Sync + 'static {
60    /// Decide on `req`.
61    async fn verdict(&self, req: &VerdictRequest) -> Verdict;
62
63    /// True when this source permits everybody (no real membership
64    /// decision). `enforce` over an open source needs an explicit opt-in.
65    fn is_open(&self) -> bool {
66        false
67    }
68}
69
70/// Permits every request (no governance configured).
71#[derive(Debug, Default, Clone, Copy)]
72pub struct OpenVerdicts;
73
74#[async_trait]
75impl VerdictSource for OpenVerdicts {
76    async fn verdict(&self, _: &VerdictRequest) -> Verdict {
77        Verdict::Permit
78    }
79    fn is_open(&self) -> bool {
80        true
81    }
82}
83
84#[cfg(feature = "exochain")]
85enum GateState {
86    Gate(Arc<dyn crate::gate::GateBackend>),
87    Open,
88    Closed,
89}
90
91/// Collapsed-mode verdict source: the local governance gate, asked the
92/// `cluster.join` action, so collapsed and service mode share one path.
93/// The gate is bound late because the mesh listener starts before the
94/// governance gate is built at boot; until bound, requests are
95/// [`Verdict::Unavailable`].
96#[cfg(feature = "exochain")]
97#[derive(Default)]
98pub struct GateVerdictSource {
99    gate: std::sync::OnceLock<GateState>,
100}
101
102#[cfg(feature = "exochain")]
103impl GateVerdictSource {
104    /// Unbound source.
105    pub fn late() -> Self {
106        Self::default()
107    }
108    /// Bind the governance gate.
109    pub fn bind(&self, gate: Arc<dyn crate::gate::GateBackend>) {
110        let _ = self.gate.set(GateState::Gate(gate));
111    }
112    /// Bind "no governance available": permit everything.
113    pub fn bind_open(&self) {
114        let _ = self.gate.set(GateState::Open);
115    }
116    /// Bind "no governance available and open membership not allowed":
117    /// refuse everything.
118    pub fn bind_closed(&self) {
119        let _ = self.gate.set(GateState::Closed);
120    }
121}
122
123#[cfg(feature = "exochain")]
124#[async_trait]
125impl VerdictSource for GateVerdictSource {
126    async fn verdict(&self, req: &VerdictRequest) -> Verdict {
127        use crate::gate::GateDecision;
128        match self.gate.get() {
129            None => Verdict::Unavailable("governance gate not ready".into()),
130            Some(GateState::Open) => Verdict::Permit,
131            Some(GateState::Closed) => Verdict::Unavailable(
132                "no governance gate; set kernel.mesh.admission_open_membership to allow".into(),
133            ),
134            Some(GateState::Gate(g)) => {
135                let ctx = serde_json::json!({
136                    "node_id": req.node_id,
137                    "platform": req.platform,
138                    "class": format!("{:?}", req.class),
139                });
140                match g.check(&req.node_id, "cluster.join", &ctx) {
141                    GateDecision::Permit { .. } => Verdict::Permit,
142                    GateDecision::Defer { reason } => Verdict::Deny(format!("deferred: {reason}")),
143                    GateDecision::Deny { reason, .. } => Verdict::Deny(reason),
144                }
145            }
146        }
147    }
148    fn is_open(&self) -> bool {
149        matches!(self.gate.get(), Some(GateState::Open))
150    }
151}
152
153/// One admission outcome, kept for `doctor` / journaling by the owner.
154#[derive(Debug, Clone)]
155pub struct AdmissionRecord {
156    /// Verified node id, when a hello verified.
157    pub node_id: Option<String>,
158    /// Refusal / would-refuse code.
159    pub code: &'static str,
160    /// Detail text.
161    pub detail: String,
162    /// Mode in force.
163    pub mode: MeshAdmissionMode,
164    /// True when the peer was served anyway (observe).
165    pub admitted: bool,
166}
167
168type CacheKey = (String, [u8; 32], PeerClass, String);
169
170/// Cryptographic admission: genesis pin, revocation, verdict, mode.
171///
172/// `genesis_hash` is a cluster *label*, not a credential: anyone who knows
173/// it can present it. Membership is the verdict source's decision.
174pub struct CryptoGate {
175    genesis: [u8; 32],
176    revocations: Arc<RevocationList>,
177    verdicts: Arc<dyn VerdictSource>,
178    mode: MeshAdmissionMode,
179    ttl: Duration,
180    // Keyed by everything the verdict depends on (id, key, declared class,
181    // platform): a cached permit never applies to a different key or class.
182    cache: Mutex<HashMap<CacheKey, (Instant, Verdict)>>,
183    records: Mutex<VecDeque<AdmissionRecord>>,
184    last_warn: Mutex<HashMap<String, Instant>>,
185}
186
187impl CryptoGate {
188    /// Build a gate pinned to `genesis`.
189    pub fn new(
190        genesis: [u8; 32],
191        revocations: Arc<RevocationList>,
192        verdicts: Arc<dyn VerdictSource>,
193        mode: MeshAdmissionMode,
194    ) -> Self {
195        Self {
196            genesis,
197            revocations,
198            verdicts,
199            mode,
200            ttl: VERDICT_TTL,
201            cache: Mutex::new(HashMap::new()),
202            records: Mutex::new(VecDeque::new()),
203            last_warn: Mutex::new(HashMap::new()),
204        }
205    }
206
207    /// Override the verdict cache lifetime.
208    pub fn with_ttl(mut self, ttl: Duration) -> Self {
209        self.ttl = ttl;
210        self
211    }
212
213    /// Recent observe/refuse records, oldest first.
214    pub fn records(&self) -> Vec<AdmissionRecord> {
215        self.records.lock().unwrap().iter().cloned().collect()
216    }
217
218    /// Number of cached verdicts (for tests and diagnostics).
219    pub fn cache_len(&self) -> usize {
220        self.cache.lock().unwrap().len()
221    }
222
223    fn warn_due(&self, key: &str) -> bool {
224        let mut w = self.last_warn.lock().unwrap();
225        if w.len() >= MAX_CACHE {
226            w.retain(|_, t| t.elapsed() < WARN_EVERY);
227        }
228        match w.get(key) {
229            Some(t) if t.elapsed() < WARN_EVERY => false,
230            _ => {
231                w.insert(key.to_owned(), Instant::now());
232                true
233            }
234        }
235    }
236
237    fn finish(&self, node_id: Option<&str>, grant: Grant, refusal: Refusal) -> Admission {
238        let admitted = self.mode != MeshAdmissionMode::Enforce;
239        if self.warn_due(&format!("{}/{}", node_id.unwrap_or("-"), refusal.code)) {
240            tracing::warn!(
241                node = node_id.unwrap_or("-"), code = refusal.code, detail = %refusal.detail,
242                mode = ?self.mode, admitted, "mesh admission failure"
243            );
244        }
245        let mut rec = self.records.lock().unwrap();
246        if rec.len() >= MAX_RECORDS {
247            rec.pop_front();
248        }
249        rec.push_back(AdmissionRecord {
250            node_id: node_id.map(str::to_owned),
251            code: refusal.code,
252            detail: refusal.detail.clone(),
253            mode: self.mode,
254            admitted,
255        });
256        if admitted {
257            Admission::Admit(Grant {
258                observed: Some(refusal),
259                trust_scope: false,
260                admitted: false,
261                ..grant
262            })
263        } else {
264            Admission::Refuse(refusal)
265        }
266    }
267
268    async fn cached_verdict(&self, req: &VerdictRequest) -> Verdict {
269        let key: CacheKey = (req.node_id.clone(), req.pubkey, req.class, req.platform.clone());
270        if let Some((at, v)) = self.cache.lock().unwrap().get(&key) {
271            let ttl = if matches!(v, Verdict::Deny(_)) { DENY_TTL.min(self.ttl) } else { self.ttl };
272            if at.elapsed() < ttl {
273                return v.clone();
274            }
275        }
276        let v = self.verdicts.verdict(req).await;
277        if !matches!(v, Verdict::Unavailable(_)) {
278            let mut c = self.cache.lock().unwrap();
279            if c.len() >= MAX_CACHE {
280                let ttl = self.ttl;
281                c.retain(|_, (t, _)| t.elapsed() < ttl);
282                // Still full: evict the oldest entries, not everything.
283                while c.len() >= MAX_CACHE {
284                    let Some(oldest) = c.iter().min_by_key(|(_, (t, _))| *t).map(|(k, _)| k.clone())
285                    else {
286                        break;
287                    };
288                    c.remove(&oldest);
289                }
290            }
291            c.insert(key, (Instant::now(), v.clone()));
292        }
293        v
294    }
295}
296
297#[async_trait]
298impl AdmissionGate for CryptoGate {
299    /// Only `enforce` bounds connections: observe must stay a no-op for
300    /// legacy peers (quiet leaves, large NAT'd fleets).
301    fn strict(&self) -> bool {
302        self.mode == MeshAdmissionMode::Enforce
303    }
304
305    /// Under `enforce`, a revocation applies to live connections too; under
306    /// `observe` it is recorded at the next admission only, as before.
307    fn is_revoked(&self, node_id: &str) -> bool {
308        self.mode == MeshAdmissionMode::Enforce && self.revocations.is_revoked(node_id)
309    }
310
311    async fn admit(&self, hello: &VerifiedHello, ctx: &AdmitContext) -> Admission {
312        if self.mode == MeshAdmissionMode::Off {
313            return Admission::Admit(Grant::open(ctx.class));
314        }
315        let enforce = self.mode == MeshAdmissionMode::Enforce;
316        let id = hello.node_id.as_str();
317        if !nested_peer_allowed(NESTED_PEERS.get().map(Vec::as_slice), id) {
318            return Admission::Refuse(Refusal { code: "nested_peer_not_granted", detail: "peer is outside the master grant".into() });
319        }
320        let base = Grant::open(ctx.class);
321        if hello.genesis_hash != self.genesis {
322            return self.finish(Some(id), base, Refusal {
323                code: "wrong_genesis",
324                detail: "hello genesis differs from the pinned cluster genesis".into(),
325            });
326        }
327        if self.revocations.is_revoked(id) {
328            return self.finish(Some(id), base, Refusal {
329                code: "revoked",
330                detail: "node id is on the revocation list".into(),
331            });
332        }
333        let req = VerdictRequest {
334            node_id: id.to_owned(),
335            pubkey: hello.pubkey,
336            class: ctx.class,
337            platform: hello.platform.clone(),
338        };
339        let class = match self.cached_verdict(&req).await {
340            Verdict::Permit => ctx.class,
341            Verdict::PermitAs(c) if enforce => c,
342            Verdict::PermitAs(_) => ctx.class,
343            Verdict::Deny(why) => {
344                return self.finish(Some(id), base, Refusal { code: "verdict_denied", detail: why });
345            }
346            Verdict::Unavailable(why) => {
347                return self.finish(Some(id), base, Refusal { code: "verdict_unavailable", detail: why });
348            }
349        };
350        Admission::Admit(Grant {
351            limits: if enforce && class == PeerClass::Leaf { PeerLimits::Leaf } else { PeerLimits::None },
352            class,
353            admitted: enforce,
354            trust_scope: enforce && class == PeerClass::Node,
355            observed: None,
356        })
357    }
358
359    async fn admit_unverified(&self, failure: &HelloFailure, ctx: &AdmitContext) -> Admission {
360        if self.mode == MeshAdmissionMode::Off {
361            return Admission::Admit(Grant::open(ctx.class));
362        }
363        let refusal = Refusal { code: failure.code(), detail: format!("{failure:?}") };
364        self.finish(None, Grant::open(ctx.class), refusal)
365    }
366}
367
368/// Additional ceiling for a master-registered inner user; admission still runs
369/// the ordinary Noise/genesis/revocation/governance checks after this filter.
370static NESTED_PEERS: std::sync::OnceLock<Vec<String>> = std::sync::OnceLock::new();
371pub fn install_nested_peer_ceiling(peers: Vec<String>) -> Result<(), String> {
372    NESTED_PEERS.set(peers).map_err(|_| "nested peer ceiling already installed".into())
373}
374
375fn nested_peer_allowed(ceiling: Option<&[String]>, id: &str) -> bool {
376    ceiling.is_none_or(|peers| peers.iter().any(|peer| peer == id))
377}
378
379
380/// Nested instances must never take the legacy unverified outbound-seed path.
381pub fn nested_peer_ceiling_active() -> bool { NESTED_PEERS.get().is_some() }
382
383#[cfg(test)]
384mod nested_peer_tests {
385    use super::nested_peer_allowed;
386    #[test]
387    fn granted_peer_ceiling_never_becomes_open_membership() {
388        let peers = vec!["granted".to_owned()];
389        assert!(nested_peer_allowed(None, "ordinary-node"));
390        assert!(nested_peer_allowed(Some(&peers), "granted"));
391        assert!(!nested_peer_allowed(Some(&peers), "unregistered"));
392        assert!(!nested_peer_allowed(Some(&[]), "unregistered"));
393    }
394}
clawft_kernel/mesh_admit_gate.rs

clawft_kernel/
mesh_admit_gate.rs
