clawft_kernel/
mesh_leaf.rs

1//! Parent-side ADR-103 leaf admission and replay floor.
2//!
3//! The machine mesh service owns this state beside its journal. In a
4//! collapsed kernel the same component lives under its isolated runtime
5//! root. The enrolled public certificate is the explicit admission grant;
6//! removing it revokes future frames. No leaf seed is stored here.
7
8use std::fs::{self, File, OpenOptions};
9use std::io::{Read, Write};
10use std::path::{Path, PathBuf};
11
12use ed25519_dalek::SigningKey;
13use clawft_types::project::cert::{key_id, ProjectCert};
14use serde::{Deserialize, Serialize};
15use weftos_leaf_types::link::{LeafCertificate, ParentAdvertisement, PublishAck, SignedPublish};
16
17pub const MAX_LEAF_FRAME: usize = 32 * 1024;
18pub const MAX_LEAF_PAYLOAD: usize = 16 * 1024;
19pub const DISCOVERY_MAGIC: &[u8; 4] = b"WLD1";
20const MAX_ENROLLMENT_FILE: usize = 4096;
21const MAX_FLOOR_FILE: usize = 512;
22
23fn private_dir(path: &Path) -> Result<(), LeafError> {
24    match fs::symlink_metadata(path) {
25        Ok(meta) if !meta.is_dir() => return Err(LeafError::Artifact(format!("{} is not a directory", path.display()))),
26        Ok(_) => {},
27        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
28            let mut builder = fs::DirBuilder::new();
29            #[cfg(unix)] {
30                use std::os::unix::fs::DirBuilderExt;
31                builder.mode(0o700);
32            }
33            builder.create(path)?;
34        },
35        Err(e) => return Err(e.into()),
36    }
37    let meta = fs::symlink_metadata(path)?;
38    #[cfg(unix)] {
39        use std::os::unix::fs::{MetadataExt, PermissionsExt};
40        if meta.uid() != unsafe { libc::geteuid() } || meta.permissions().mode() & 0o077 != 0 {
41            return Err(LeafError::Artifact(format!("{} must be owned by the service and mode 0700", path.display())));
42        }
43    }
44    Ok(())
45}
46
47fn read_bounded(path: &Path, max: usize) -> Result<Option<Vec<u8>>, LeafError> {
48    let mut opts = OpenOptions::new(); opts.read(true);
49    #[cfg(unix)] {
50        use std::os::unix::fs::OpenOptionsExt;
51        opts.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
52    }
53    let f = match opts.open(path) {
54        Ok(f) => f,
55        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
56        Err(e) => return Err(e.into()),
57    };
58    let meta = f.metadata()?;
59    if !meta.is_file() || meta.len() > max as u64 {
60        return Err(LeafError::Artifact(format!("{} is not a bounded regular file", path.display())));
61    }
62    #[cfg(unix)] {
63        use std::os::unix::fs::MetadataExt;
64        if meta.uid() != unsafe { libc::geteuid() } {
65            return Err(LeafError::Artifact(format!("{} is not owned by the service", path.display())));
66        }
67    }
68    let mut bytes = Vec::new();
69    f.take(max as u64 + 1).read_to_end(&mut bytes)?;
70    if bytes.len() > max { return Err(LeafError::Artifact(format!("{} exceeds {} bytes", path.display(), max))); }
71    Ok(Some(bytes))
72}
73
74#[derive(Debug, thiserror::Error)]
75pub enum LeafError {
76    #[error("leaf io: {0}")]
77    Io(#[from] std::io::Error),
78    #[error("leaf artifact: {0}")]
79    Artifact(String),
80    #[error("leaf certificate is not enrolled or has been revoked")]
81    NotEnrolled,
82    #[error("leaf certificate/signature: {0:?}")]
83    Signature(weftos_leaf_types::link::LinkError),
84    #[error("leaf sequence gap or regression: expected {expected}, got {got}")]
85    Sequence { expected: u64, got: u64 },
86    #[error("leaf sequence reused with different content")]
87    Equivocation,
88    #[error("leaf payload exceeds {MAX_LEAF_PAYLOAD} bytes")]
89    TooLarge,
90    #[error("leaf parent does not own the certified tenant scope: {0}")]
91    WrongScope(String),
92}
93
94#[derive(Deserialize)]
95struct Enrollment {
96    node_id: String,
97    cert: LeafCertificate,
98    #[serde(default)]
99    project_cert: Option<ProjectCert>,
100}
101
102fn hex_key(s: &str) -> Option<[u8; 32]> {
103    let raw = hex::decode(s).ok()?;
104    raw.try_into().ok()
105}
106
107/// Prove the leaf issuer owns its requested tenant. A user issuer must be
108/// that user's key. A project issuer needs the user's signed project cert,
109/// with both key IDs recomputed by `ProjectCert::verify`.
110pub fn verify_scope_chain(cert: &LeafCertificate, project_cert: Option<&ProjectCert>, now: u64) -> Result<[u8; 32], LeafError> {
111    let (user_id, project_id) = weftos_leaf_types::link::parse_parent_scope(&cert.parent_scope)
112        .map_err(|e| LeafError::WrongScope(format!("scope syntax: {e:?}")))?;
113    match project_id {
114        None => {
115            if project_cert.is_some() || key_id(&cert.parent_pubkey) != user_id {
116                return Err(LeafError::WrongScope("user issuer key ID does not match user scope".into()));
117            }
118            Ok(cert.parent_pubkey)
119        }
120        Some(project_id) => {
121            let pc = project_cert.ok_or_else(|| LeafError::WrongScope("project certificate missing".into()))?;
122            let user_key = hex_key(&pc.user_pubkey).ok_or_else(|| LeafError::WrongScope("project user key malformed".into()))?;
123            let at = i64::try_from(now).ok().and_then(|s| chrono::DateTime::<chrono::Utc>::from_timestamp(s, 0))
124                .ok_or_else(|| LeafError::WrongScope("invalid verification time".into()))?;
125            pc.verify(&user_key, at).map_err(|e| LeafError::WrongScope(format!("project certificate: {e}")))?;
126            if pc.project_id != project_id || pc.project_key_id != key_id(&cert.parent_pubkey)
127                || pc.user_key_id != user_id || hex_key(&pc.project_pubkey) != Some(cert.parent_pubkey) {
128                return Err(LeafError::WrongScope("project certificate does not bind issuer and tenant".into()));
129            }
130            Ok(user_key)
131        }
132    }
133}
134
135#[derive(Serialize, Deserialize)]
136struct Floor {
137    seq: u64,
138    digest: [u8; 32],
139}
140
141#[derive(Debug, Clone, Copy, PartialEq, Eq)]
142pub enum Prepare { New, Duplicate }
143
144type ScopeAuthority = std::sync::Arc<dyn Fn(&str, &[u8; 32], Option<&ProjectCert>) -> bool + Send + Sync>;
145
146pub struct LeafIngress {
147    registry: PathBuf,
148    floors: PathBuf,
149    machine_key: SigningKey,
150    scope_authority: Option<ScopeAuthority>,
151    /// Serialize a leaf's check, delivery and floor commit. Held across the
152    /// runtime call; callers use Tokio's async mutex guard.
153    pub lock: tokio::sync::Mutex<()>,
154}
155
156impl LeafIngress {
157    /// `root/registry/<id>.json` holds enrollment artifacts; `root/floors`
158    /// is service-owned and must never be writable by a leaf process.
159    pub fn open(root: &Path, machine_key: SigningKey) -> Result<Self, LeafError> {
160        let registry = root.join("registry");
161        let floors = root.join("floors");
162        private_dir(root)?;
163        private_dir(&registry)?;
164        private_dir(&floors)?;
165        Ok(Self { registry, floors, machine_key, scope_authority: None, lock: tokio::sync::Mutex::new(()) })
166    }
167
168    /// Machine-service live binding check. The callback must prove the
169    /// registered tenant currently holds the user key named by the chain.
170    pub fn with_scope_authority(mut self, authority: ScopeAuthority) -> Self {
171        self.scope_authority = Some(authority);
172        self
173    }
174
175    fn authorized(&self, enrolled: &Enrollment, cert: &LeafCertificate, now: u64) -> bool {
176        if enrolled.node_id != cert.leaf_id() || enrolled.cert != *cert
177            || cert.mesh_pubkey != self.machine_key.verifying_key().to_bytes()
178            || cert.verify(&cert.parent_pubkey, now).is_err() { return false; }
179        let Ok(user_key) = verify_scope_chain(cert, enrolled.project_cert.as_ref(), now) else { return false };
180        let (user_id, _) = weftos_leaf_types::link::parse_parent_scope(&cert.parent_scope).expect("verified scope");
181        self.scope_authority.as_ref().is_none_or(|authority| authority(user_id, &user_key, enrolled.project_cert.as_ref()))
182    }
183
184    fn floor_path(&self, id: &str) -> PathBuf { self.floors.join(format!("{id}.json")) }
185
186    fn read_floor(&self, id: &str) -> Result<Option<Floor>, LeafError> {
187        read_bounded(&self.floor_path(id), MAX_FLOOR_FILE)?
188            .map(|bytes| serde_json::from_slice(&bytes).map_err(|e| LeafError::Artifact(format!("invalid replay floor: {e}"))))
189            .transpose()
190    }
191
192    /// Recheck an idle subscribed connection. Enrollment removal, exact
193    /// certificate replacement and expiry all terminate its push route.
194    pub fn still_enrolled(&self, cert: &LeafCertificate, now: u64) -> bool {
195        let id = cert.leaf_id();
196        let Ok(Some(bytes)) = read_bounded(&self.registry.join(format!("{id}.json")), MAX_ENROLLMENT_FILE) else { return false };
197        let Ok(enrolled) = serde_json::from_slice::<Enrollment>(&bytes) else { return false };
198        self.authorized(&enrolled, cert, now)
199    }
200
201    /// Validate against the installed certificate and durable replay floor.
202    /// `Duplicate` is safe to ACK but must never be delivered again.
203    pub fn prepare(&self, frame: &SignedPublish, now: u64) -> Result<Prepare, LeafError> {
204        if frame.payload.len() > MAX_LEAF_PAYLOAD { return Err(LeafError::TooLarge); }
205        let id = frame.cert.leaf_id();
206        let bytes = read_bounded(&self.registry.join(format!("{id}.json")), MAX_ENROLLMENT_FILE)?.ok_or(LeafError::NotEnrolled)?;
207        let enrolled: Enrollment = serde_json::from_slice(&bytes).map_err(|e| LeafError::Artifact(format!("invalid enrollment: {e}")))?;
208        if !self.authorized(&enrolled, &frame.cert, now) {
209            return Err(LeafError::NotEnrolled);
210        }
211        frame.verify(&enrolled.cert.parent_pubkey, now).map_err(LeafError::Signature)?;
212        match self.read_floor(&id)? {
213            None if frame.seq == 1 => Ok(Prepare::New),
214            None => Err(LeafError::Sequence { expected: 1, got: frame.seq }),
215            Some(floor) if frame.seq == floor.seq => {
216                if frame.digest() == floor.digest { Ok(Prepare::Duplicate) } else { Err(LeafError::Equivocation) }
217            }
218            Some(floor) if frame.seq == floor.seq.saturating_add(1) => Ok(Prepare::New),
219            Some(floor) => Err(LeafError::Sequence { expected: floor.seq.saturating_add(1), got: frame.seq }),
220        }
221    }
222
223    /// Call only after delivery returned success. Sync file and directory
224    /// before sending an ACK. A crash between delivery and commit can replay
225    /// once; downstream operations must be idempotent by `(leaf_id, seq)`.
226    pub fn commit(&self, frame: &SignedPublish) -> Result<(), LeafError> {
227        let id = frame.cert.leaf_id();
228        let path = self.floor_path(&id);
229        let tmp = self.floors.join(format!(".{id}.{:016x}.tmp", rand::random::<u64>()));
230        let bytes = serde_json::to_vec(&Floor { seq: frame.seq, digest: frame.digest() })
231            .map_err(|e| LeafError::Artifact(e.to_string()))?;
232        let mut opts = OpenOptions::new(); opts.write(true).create_new(true);
233        #[cfg(unix)] {
234            use std::os::unix::fs::OpenOptionsExt;
235            opts.mode(0o600);
236        }
237        let mut f = opts.open(&tmp)?;
238        let result = (|| -> Result<(), LeafError> {
239            f.write_all(&bytes)?;
240            f.sync_all()?;
241            fs::rename(&tmp, &path)?;
242            File::open(&self.floors)?.sync_all()?;
243            Ok(())
244        })();
245        if result.is_err() { let _ = fs::remove_file(&tmp); }
246        result
247    }
248
249    pub fn ack(&self, frame: &SignedPublish) -> PublishAck { PublishAck::sign(&self.machine_key, frame) }
250
251    fn advertisement(&self, id: &str, endpoint: String, now: u64, nonce: [u8; 16]) -> Option<ParentAdvertisement> {
252        if id.len() != 32 || !id.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) { return None; }
253        let bytes = read_bounded(&self.registry.join(format!("{id}.json")), MAX_ENROLLMENT_FILE).ok()??;
254        let enrolled: Enrollment = serde_json::from_slice(&bytes).ok()?;
255        if !self.authorized(&enrolled, &enrolled.cert, now) { return None; }
256        Some(ParentAdvertisement::sign(&self.machine_key, enrolled.cert.parent_scope, endpoint, now.saturating_add(30), nonce))
257    }
258
259    /// UDP discovery on the mesh listener's address and port + 1. Requests
260    /// carry `WLD1 | leaf id (32 ASCII hex) | nonce (16)`. The response is a
261    /// machine-signed CBOR advertisement bound to that nonce and enrollment.
262    pub async fn serve_discovery(&self, mesh_addr: std::net::SocketAddr, leaf_addr: std::net::SocketAddr) -> Result<(), LeafError> {
263        let port = mesh_addr.port().checked_add(1).ok_or_else(|| LeafError::Artifact("mesh port has no discovery successor".into()))?;
264        let bind = std::net::SocketAddr::new(mesh_addr.ip(), port);
265        let sock = tokio::net::UdpSocket::bind(bind).await?;
266        let mut request = [0u8; 64];
267        loop {
268            let (n, peer) = sock.recv_from(&mut request).await?;
269            if n != 52 || &request[..4] != DISCOVERY_MAGIC { continue; }
270            let Ok(id) = std::str::from_utf8(&request[4..36]) else { continue };
271            let mut nonce = [0u8; 16]; nonce.copy_from_slice(&request[36..52]);
272            let ip = if mesh_addr.ip().is_unspecified() {
273                let probe = std::net::UdpSocket::bind(std::net::SocketAddr::new(mesh_addr.ip(), 0))?;
274                if probe.connect(peer).is_err() { continue; }
275                match probe.local_addr() { Ok(a) => a.ip(), Err(_) => continue }
276            } else { mesh_addr.ip() };
277            let endpoint = std::net::SocketAddr::new(ip, leaf_addr.port()).to_string();
278            let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
279            if let Some(ad) = self.advertisement(id, endpoint, now, nonce)
280                && let Ok(bytes) = weftos_leaf_types::encode(&ad)
281                && bytes.len() <= 1024 {
282                    let _ = sock.send_to(&bytes, peer).await;
283                }
284        }
285    }
286}
287
288#[cfg(test)]
289mod tests {
290    use super::*;
291    use clawft_types::project::cert::{CertRequest, ProjectCert};
292    use ed25519_dalek::SigningKey;
293    use weftos_leaf_types::link::node_id;
294
295    #[test]
296    fn scope_chain_rejects_another_users_scope_and_requires_project_issuer_proof() {
297        let user_a = SigningKey::from_bytes(&[51; 32]);
298        let user_b = SigningKey::from_bytes(&[52; 32]);
299        let project = SigningKey::from_bytes(&[53; 32]);
300        let leaf = SigningKey::from_bytes(&[54; 32]);
301        let machine = SigningKey::from_bytes(&[55; 32]);
302        let id_a = key_id(&user_a.verifying_key().to_bytes());
303        let id_b = key_id(&user_b.verifying_key().to_bytes());
304        let now = chrono::Utc::now();
305        let secs = now.timestamp() as u64;
306        let wrong_user = LeafCertificate::issue(&user_a, format!("user:{id_b}"), machine.verifying_key().to_bytes(),
307            leaf.verifying_key().to_bytes(), 1, secs - 1, secs + 60, vec!["input.publish".into()]).unwrap();
308        assert!(verify_scope_chain(&wrong_user, None, secs).is_err());
309
310        let project_id = "01HZY7Y9M3MZQ5RSY3MTEP6F3C";
311        let chain = ProjectCert::sign(&user_a, &CertRequest {
312            project_id: project_id.into(), project_pubkey: project.verifying_key().to_bytes(),
313            serial: 1, issued_at: now, expires_at: None,
314        });
315        let good = LeafCertificate::issue(&project, format!("project:{id_a}:{project_id}"), machine.verifying_key().to_bytes(),
316            leaf.verifying_key().to_bytes(), 2, secs - 1, secs + 60, vec!["input.publish".into()]).unwrap();
317        assert_eq!(verify_scope_chain(&good, Some(&chain), secs).unwrap(), user_a.verifying_key().to_bytes());
318        assert!(verify_scope_chain(&good, None, secs).is_err());
319        let forged_scope = LeafCertificate::issue(&project, format!("project:{id_b}:{project_id}"), machine.verifying_key().to_bytes(),
320            leaf.verifying_key().to_bytes(), 3, secs - 1, secs + 60, vec!["input.publish".into()]).unwrap();
321        assert!(verify_scope_chain(&forged_scope, Some(&chain), secs).is_err());
322        let other_project = SigningKey::from_bytes(&[56; 32]);
323        let wrong_issuer = LeafCertificate::issue(&other_project, format!("project:{id_a}:{project_id}"), machine.verifying_key().to_bytes(),
324            leaf.verifying_key().to_bytes(), 4, secs - 1, secs + 60, vec!["input.publish".into()]).unwrap();
325        assert!(verify_scope_chain(&wrong_issuer, Some(&chain), secs).is_err());
326
327        let dir = tempfile::tempdir().unwrap();
328        #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).unwrap(); }
329        let expected_user = id_a.clone();
330        let expected_key = user_a.verifying_key().to_bytes();
331        let ingress = LeafIngress::open(dir.path(), machine).unwrap().with_scope_authority(std::sync::Arc::new(move |id, key, _| {
332            id == expected_user && *key == expected_key
333        }));
334        let leaf_id = good.leaf_id();
335        fs::write(ingress.registry.join(format!("{leaf_id}.json")),
336            serde_json::json!({"node_id": leaf_id, "cert": good.clone(), "project_cert": chain.clone()}).to_string()).unwrap();
337        let frame = SignedPublish::sign(good.clone(), &leaf, 1, format!("mesh.leaf.{leaf_id}.input"), vec![1]).unwrap();
338        assert_eq!(ingress.prepare(&frame, secs).unwrap(), Prepare::New);
339        let denied = LeafIngress::open(dir.path(), ingress.machine_key.clone()).unwrap()
340            .with_scope_authority(std::sync::Arc::new(|_, _, _| false));
341        assert!(matches!(denied.prepare(&frame, secs), Err(LeafError::NotEnrolled)));
342    }
343
344    #[test]
345    fn enrolled_frame_replays_exactly_once_across_restart() {
346        let dir = tempfile::tempdir().unwrap();
347        #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).unwrap(); }
348        let parent = SigningKey::from_bytes(&[3;32]);
349        let user_id = clawft_types::project::cert::key_id(&parent.verifying_key().to_bytes());
350        let machine = SigningKey::from_bytes(&[4;32]);
351        let leaf = SigningKey::from_bytes(&[5;32]);
352        let cert = LeafCertificate::issue(&parent, format!("user:{user_id}"), machine.verifying_key().to_bytes(), leaf.verifying_key().to_bytes(), 1, 1, 1000, vec!["input.publish".into()]).unwrap();
353        let id = node_id(&cert.leaf_pubkey);
354        let frame = SignedPublish::sign(cert.clone(), &leaf, 1, format!("mesh.leaf.{id}.input"), vec![1]).unwrap();
355        let ingress = LeafIngress::open(dir.path(), machine.clone()).unwrap();
356        fs::write(ingress.registry.join(format!("{id}.json")), serde_json::json!({"node_id":id,"cert":cert}).to_string()).unwrap();
357        assert_eq!(ingress.prepare(&frame, 10).unwrap(), Prepare::New);
358        ingress.commit(&frame).unwrap();
359        let ingress = LeafIngress::open(dir.path(), machine).unwrap();
360        assert_eq!(ingress.prepare(&frame, 10).unwrap(), Prepare::Duplicate);
361        assert!(ingress.still_enrolled(&frame.cert, 10));
362        assert!(!ingress.still_enrolled(&frame.cert, 1000));
363        ingress.ack(&frame).verify(&frame.cert.mesh_pubkey, &frame).unwrap();
364        let mut altered = frame.clone(); altered.payload.push(2);
365        assert!(matches!(ingress.prepare(&altered, 10), Err(LeafError::Signature(_))));
366        fs::remove_file(ingress.registry.join(format!("{id}.json"))).unwrap();
367        assert!(!ingress.still_enrolled(&frame.cert, 10));
368        assert!(matches!(ingress.prepare(&frame, 10), Err(LeafError::NotEnrolled)));
369    }
370
371    #[test]
372    fn discovery_only_answers_enrolled_leaf_with_signed_endpoint_and_nonce() {
373        let dir = tempfile::tempdir().unwrap();
374        #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; fs::set_permissions(dir.path(), fs::Permissions::from_mode(0o700)).unwrap(); }
375        let parent = SigningKey::from_bytes(&[13; 32]);
376        let user_id = clawft_types::project::cert::key_id(&parent.verifying_key().to_bytes());
377        let machine = SigningKey::from_bytes(&[14; 32]);
378        let leaf = SigningKey::from_bytes(&[15; 32]);
379        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
380        let cert = LeafCertificate::issue(&parent, format!("user:{user_id}"), machine.verifying_key().to_bytes(), leaf.verifying_key().to_bytes(), 1, now - 1, now + 60, vec!["input.publish".into()]).unwrap();
381        let id = cert.leaf_id();
382        let ingress = LeafIngress::open(dir.path(), machine.clone()).unwrap();
383        fs::write(ingress.registry.join(format!("{id}.json")), serde_json::json!({"node_id":id,"cert":cert}).to_string()).unwrap();
384        let endpoint = "127.0.0.1:9489".to_string();
385        let ad = ingress.advertisement(&id, endpoint.clone(), now, [9; 16]).unwrap();
386        ad.verify_for(&machine.verifying_key().to_bytes(), &cert.parent_scope, now, &[9; 16]).unwrap();
387        assert_eq!(ad.endpoint, endpoint);
388        assert!(ad.verify_for(&machine.verifying_key().to_bytes(), &cert.parent_scope, now, &[8; 16]).is_err());
389        assert!(ingress.advertisement(&"f".repeat(32), endpoint, now, [9; 16]).is_none());
390    }
391}
clawft_kernel/mesh_leaf.rs

clawft_kernel/
mesh_leaf.rs
