clawft_kernel/workload_ctl/
msg.rs

1//! The `workload.ctl` message set (ADR-099 section 7).
2//!
3//! Every request is a [`CtlRequest`] signed by the requesting node's Ed25519
4//! key ([`SignedCtl`]). It names the requester (the node id derived from the
5//! signing key), the target `workload-host`, a random nonce and an expiry.
6//! Mutating methods also carry the id of the chained governance decision
7//! that authorised them. Responses are signed by the target and bound to
8//! the request nonce, so a response cannot be replayed onto another request.
9//!
10//! Verification order on the target: size, signature, structure, key
11//! binding, addressee, time window, authorised controller key, nonce
12//! freshness. The nonce is recorded only after everything else passed, so a
13//! forged request cannot burn a legitimate nonce.
14
15use std::collections::HashMap;
16use std::sync::Mutex;
17
18use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
19use serde::{Deserialize, Serialize};
20use serde_json::Value;
21
22use crate::node_registry::node_id_from_pubkey;
23use crate::workload_pkg::codec::{hex_decode_exact, hex_encode};
24
25/// Service name the control plane addresses on every node.
26pub const WORKLOAD_HOST_SERVICE: &str = "workload-host";
27/// Wire version.
28pub const CTL_VERSION: u32 = 1;
29/// Domain tag signed before a request payload.
30pub const REQUEST_DOMAIN: &[u8] = b"weftos.workload_ctl.request.v1\0";
31/// Domain tag signed before a response payload.
32pub const RESPONSE_DOMAIN: &[u8] = b"weftos.workload_ctl.response.v1\0";
33/// Longest request lifetime a target accepts.
34pub const MAX_TTL_MS: u64 = 300_000;
35/// Tolerated clock skew for `issued_at_ms`.
36pub const MAX_SKEW_MS: u64 = 30_000;
37/// Largest signed payload.
38pub const MAX_CTL_BYTES: usize = 256 * 1024;
39/// Most nonces remembered at once (fail closed when full).
40pub const MAX_NONCES: usize = 8192;
41/// Target id a `describe` may use before the caller knows the node id.
42pub const ANY_TARGET: &str = "*";
43
44/// Methods of the message set. Anything else under `workload.` is denied.
45pub mod method {
46    /// Signed facts and the `workload-host` advertisement (read-only).
47    pub const DESCRIBE: &str = "workload.describe";
48    /// Fetch, verify, admit, load and optionally start a placed workload.
49    pub const PLACE: &str = "workload.place";
50    /// Like place, without starting and without a placement decision.
51    pub const LOAD: &str = "workload.load";
52    /// Start a loaded instance.
53    pub const START: &str = "workload.start";
54    /// Stop a running instance.
55    pub const STOP: &str = "workload.stop";
56    /// Unload an instance.
57    pub const UNLOAD: &str = "workload.unload";
58    /// Instance status (read-only).
59    pub const STATUS: &str = "workload.status";
60    /// Last captured output of an instance (read-only).
61    pub const LOGS: &str = "workload.logs";
62    /// Every method of the set.
63    pub const ALL: &[&str] = &[DESCRIBE, PLACE, LOAD, START, STOP, UNLOAD, STATUS, LOGS];
64
65    /// Dashboard reporter status on the target node (read-only). Not part of
66    /// [`ALL`]: it is a node-admin method served by the node's
67    /// [`NodeAdmin`](crate::workload_ctl::NodeAdmin) hook, not by an adapter.
68    pub const DASHBOARD_STATUS: &str = "dashboard.status";
69    /// Rotate the target node's dashboard token (changes state; carries a
70    /// decision id and is chained on both nodes).
71    pub const DASHBOARD_ROTATE: &str = "dashboard.token.rotate";
72    /// Node-admin methods: signed like the rest of the set, but answered by
73    /// the node's own admin hook rather than a workload adapter.
74    pub const NODE_ADMIN: &[&str] = &[DASHBOARD_STATUS, DASHBOARD_ROTATE];
75
76    /// True for a node-admin method.
77    pub fn is_node_admin(m: &str) -> bool {
78        NODE_ADMIN.contains(&m)
79    }
80
81    /// Methods that only take down what is running. Trust policy never
82    /// blocks these on the controller: an operator can always stop and
83    /// unload what they placed.
84    pub fn is_teardown(m: &str) -> bool {
85        matches!(m, STOP | UNLOAD)
86    }
87
88    /// Methods that change state (they need a decision id).
89    pub fn mutates(m: &str) -> bool {
90        matches!(m, PLACE | LOAD | START | STOP | UNLOAD | DASHBOARD_ROTATE)
91    }
92}
93
94/// One control request.
95#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
96#[serde(deny_unknown_fields)]
97pub struct CtlRequest {
98    /// [`CTL_VERSION`].
99    pub version: u32,
100    /// Method (`workload.place`, ...).
101    pub method: String,
102    /// Requesting node id (derived from the signing key).
103    pub requester: String,
104    /// Target node id, or [`ANY_TARGET`] for `describe`.
105    pub target: String,
106    /// 32 lower-case hex characters, fresh per request.
107    pub nonce: String,
108    /// Issue time, ms since the Unix epoch.
109    pub issued_at_ms: u64,
110    /// Expiry, ms since the Unix epoch.
111    pub expires_at_ms: u64,
112    /// Chain hash of the requester's governance decision (mutations).
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub decision_id: Option<String>,
115    /// Method-specific body.
116    #[serde(default)]
117    pub body: Value,
118}
119
120/// Signed envelope for a request or response. `payload` is the exact JSON
121/// that was signed, so verification never re-serialises.
122#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
123#[serde(deny_unknown_fields)]
124pub struct SignedCtl {
125    /// Exact JSON bytes of the request / response.
126    pub payload: String,
127    /// Hex Ed25519 public key (32 bytes).
128    pub public_key: String,
129    /// Hex Ed25519 signature (64 bytes) over domain || payload.
130    pub signature: String,
131}
132
133/// Why a target refused a request. Stable names (chained).
134#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
135#[serde(rename_all = "snake_case")]
136pub enum RefusalCode {
137    /// Signature, key or structure check failed.
138    Signature,
139    /// Outside its time window.
140    Expired,
141    /// Nonce already seen.
142    Replay,
143    /// Addressed to another node.
144    NotForMe,
145    /// Signed by a key this node does not accept as a controller.
146    Unauthorized,
147    /// Not a method of the message set (default deny).
148    UnknownMethod,
149    /// Malformed body or missing decision id.
150    InvalidRequest,
151    /// Governance denied the transition on this node.
152    Governance,
153    /// The admission self-check disagreed with the placement.
154    Admission,
155    /// The payload could not be fetched.
156    Fetch,
157    /// The payload failed verification.
158    Verify,
159    /// The runtime adapter failed.
160    Runtime,
161    /// No such instance on this node.
162    UnknownInstance,
163}
164
165impl RefusalCode {
166    /// Stable name.
167    pub fn as_str(self) -> &'static str {
168        match self {
169            Self::Signature => "signature",
170            Self::Expired => "expired",
171            Self::Replay => "replay",
172            Self::NotForMe => "not_for_me",
173            Self::Unauthorized => "unauthorized",
174            Self::UnknownMethod => "unknown_method",
175            Self::InvalidRequest => "invalid_request",
176            Self::Governance => "governance",
177            Self::Admission => "admission",
178            Self::Fetch => "fetch",
179            Self::Verify => "verify",
180            Self::Runtime => "runtime",
181            Self::UnknownInstance => "unknown_instance",
182        }
183    }
184}
185
186/// A refusal: code plus human-readable reason.
187#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
188#[error("{} refused: {reason}", code.as_str())]
189pub struct Refusal {
190    /// Code.
191    pub code: RefusalCode,
192    /// Reason.
193    pub reason: String,
194}
195
196impl Refusal {
197    /// Build a refusal.
198    pub fn new(code: RefusalCode, reason: impl Into<String>) -> Self {
199        Self {
200            code,
201            reason: reason.into(),
202        }
203    }
204}
205
206/// What the target did.
207#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
208#[serde(tag = "outcome", rename_all = "snake_case")]
209pub enum CtlOutcome {
210    /// Done; method-specific result.
211    Ok {
212        /// Result.
213        result: Value,
214    },
215    /// Refused.
216    Refused {
217        /// Why.
218        refusal: Refusal,
219    },
220}
221
222/// One control response.
223#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
224#[serde(deny_unknown_fields)]
225pub struct CtlResponse {
226    /// [`CTL_VERSION`].
227    pub version: u32,
228    /// Method answered.
229    pub method: String,
230    /// Responding node id (derived from its key).
231    pub responder: String,
232    /// Nonce of the request this answers.
233    pub request_nonce: String,
234    /// Outcome.
235    pub outcome: CtlOutcome,
236}
237
238pub(crate) fn signed_bytes(domain: &[u8], payload: &str) -> Vec<u8> {
239    let mut v = Vec::with_capacity(domain.len() + payload.len());
240    v.extend_from_slice(domain);
241    v.extend_from_slice(payload.as_bytes());
242    v
243}
244
245pub(crate) fn sign(domain: &[u8], payload: String, key: &SigningKey) -> SignedCtl {
246    let sig = key.sign(&signed_bytes(domain, &payload));
247    SignedCtl {
248        payload,
249        public_key: hex_encode(&key.verifying_key().to_bytes()),
250        signature: hex_encode(&sig.to_bytes()),
251    }
252}
253
254/// Check the envelope signature; returns the signer's raw key.
255pub(crate) fn open(domain: &[u8], s: &SignedCtl) -> Result<[u8; 32], Refusal> {
256    let bad = |m: &str| Refusal::new(RefusalCode::Signature, m);
257    if s.payload.len() > MAX_CTL_BYTES {
258        return Err(bad("payload too large"));
259    }
260    let pk = hex_decode_exact::<32>(&s.public_key).ok_or_else(|| bad("bad public key"))?;
261    let sig = hex_decode_exact::<64>(&s.signature).ok_or_else(|| bad("bad signature encoding"))?;
262    let vk = VerifyingKey::from_bytes(&pk).map_err(|_| bad("bad public key"))?;
263    vk.verify(
264        &signed_bytes(domain, &s.payload),
265        &Signature::from_bytes(&sig),
266    )
267    .map_err(|_| bad("signature does not verify"))?;
268    Ok(pk)
269}
270
271/// A fresh random nonce (32 hex).
272pub fn fresh_nonce() -> String {
273    let b: [u8; 16] = rand::random();
274    hex_encode(&b)
275}
276
277impl CtlRequest {
278    /// Build a request from `key`'s node, valid for `ttl_ms` from `now_ms`.
279    pub fn new(
280        key: &SigningKey,
281        method: &str,
282        target: &str,
283        now_ms: u64,
284        ttl_ms: u64,
285        decision_id: Option<String>,
286        body: Value,
287    ) -> Self {
288        Self {
289            version: CTL_VERSION,
290            method: method.to_string(),
291            requester: node_id_from_pubkey(&key.verifying_key().to_bytes()),
292            target: target.to_string(),
293            nonce: fresh_nonce(),
294            issued_at_ms: now_ms,
295            expires_at_ms: now_ms.saturating_add(ttl_ms),
296            decision_id,
297            body,
298        }
299    }
300
301    /// Sign with `key`. The requester must be `key`'s node id.
302    pub fn sign(&self, key: &SigningKey) -> Result<SignedCtl, Refusal> {
303        let payload = serde_json::to_string(self)
304            .map_err(|e| Refusal::new(RefusalCode::InvalidRequest, e.to_string()))?;
305        Ok(sign(REQUEST_DOMAIN, payload, key))
306    }
307}
308
309impl CtlResponse {
310    /// Sign with the responder's key.
311    pub fn sign(&self, key: &SigningKey) -> SignedCtl {
312        let payload = serde_json::to_string(self).unwrap_or_else(|_| "{}".into());
313        sign(RESPONSE_DOMAIN, payload, key)
314    }
315}
316
317/// Replay guard: nonces seen, kept until their request expires.
318#[derive(Debug, Default)]
319pub struct NonceGuard {
320    seen: Mutex<HashMap<String, u64>>,
321}
322
323impl NonceGuard {
324    /// Empty guard.
325    pub fn new() -> Self {
326        Self::default()
327    }
328
329    /// Record `nonce` (valid until `expires_ms`). False if already seen or
330    /// the guard is full of unexpired nonces (fail closed).
331    pub fn admit(&self, nonce: &str, expires_ms: u64, now_ms: u64) -> bool {
332        let Ok(mut seen) = self.seen.lock() else {
333            return false;
334        };
335        if seen.contains_key(nonce) {
336            return false;
337        }
338        if seen.len() >= MAX_NONCES {
339            seen.retain(|_, exp| *exp > now_ms);
340            if seen.len() >= MAX_NONCES {
341                return false;
342            }
343        }
344        seen.insert(nonce.to_string(), expires_ms);
345        true
346    }
347}
348
349/// Who may control this node: a set of controller public keys.
350pub trait ControllerPolicy: Send + Sync {
351    /// True if a request signed by `public_key` may be served.
352    fn allows(&self, public_key: &[u8; 32]) -> bool;
353}
354
355impl ControllerPolicy for Vec<[u8; 32]> {
356    fn allows(&self, public_key: &[u8; 32]) -> bool {
357        self.contains(public_key)
358    }
359}
360
361/// Verify a request addressed to `me` at `now_ms`. See the module docs for
362/// the order of checks.
363pub fn verify_request(
364    s: &SignedCtl,
365    me: &str,
366    now_ms: u64,
367    controllers: &dyn ControllerPolicy,
368    nonces: &NonceGuard,
369) -> Result<CtlRequest, Refusal> {
370    let pk = open(REQUEST_DOMAIN, s)?;
371    let req: CtlRequest = serde_json::from_str(&s.payload)
372        .map_err(|e| Refusal::new(RefusalCode::Signature, format!("malformed request: {e}")))?;
373    if req.version != CTL_VERSION {
374        return Err(Refusal::new(
375            RefusalCode::InvalidRequest,
376            "unsupported version",
377        ));
378    }
379    let derived = node_id_from_pubkey(&pk);
380    if req.requester != derived {
381        return Err(Refusal::new(
382            RefusalCode::Signature,
383            format!("requester {} is not the signing key's node", req.requester),
384        ));
385    }
386    let any_ok = req.target == ANY_TARGET && req.method == method::DESCRIBE;
387    if req.target != me && !any_ok {
388        return Err(Refusal::new(
389            RefusalCode::NotForMe,
390            format!("addressed to {}", req.target),
391        ));
392    }
393    if req.expires_at_ms <= now_ms {
394        return Err(Refusal::new(RefusalCode::Expired, "request expired"));
395    }
396    if req.issued_at_ms > now_ms.saturating_add(MAX_SKEW_MS) {
397        return Err(Refusal::new(
398            RefusalCode::Expired,
399            "request issued in the future",
400        ));
401    }
402    if req.expires_at_ms.saturating_sub(req.issued_at_ms) > MAX_TTL_MS
403        || req.expires_at_ms < req.issued_at_ms
404    {
405        return Err(Refusal::new(
406            RefusalCode::Expired,
407            "request lifetime out of bounds",
408        ));
409    }
410    if !crate::workload_pkg::codec::is_lower_hex(&req.nonce, 32) {
411        return Err(Refusal::new(
412            RefusalCode::InvalidRequest,
413            "nonce must be 32 hex",
414        ));
415    }
416    if !controllers.allows(&pk) {
417        return Err(Refusal::new(
418            RefusalCode::Unauthorized,
419            format!("{derived} is not an authorised controller of this node"),
420        ));
421    }
422    if !nonces.admit(&req.nonce, req.expires_at_ms, now_ms) {
423        return Err(Refusal::new(RefusalCode::Replay, "nonce already used"));
424    }
425    Ok(req)
426}
427
428/// Verify a response to `request` from the node whose key is `expected`
429/// (`None` accepts any key, then binds `responder` to it: `describe` to an
430/// unknown node). Returns the response and the responder's key.
431pub fn verify_response(
432    s: &SignedCtl,
433    request: &CtlRequest,
434    expected: Option<&[u8; 32]>,
435) -> Result<(CtlResponse, [u8; 32]), Refusal> {
436    let pk = open(RESPONSE_DOMAIN, s)?;
437    if expected.is_some_and(|e| e != &pk) {
438        return Err(Refusal::new(
439            RefusalCode::Signature,
440            "response signed by another key",
441        ));
442    }
443    let resp: CtlResponse = serde_json::from_str(&s.payload)
444        .map_err(|e| Refusal::new(RefusalCode::Signature, format!("malformed response: {e}")))?;
445    if resp.responder != node_id_from_pubkey(&pk) {
446        return Err(Refusal::new(
447            RefusalCode::Signature,
448            "responder is not the signing key",
449        ));
450    }
451    if resp.request_nonce != request.nonce || resp.method != request.method {
452        return Err(Refusal::new(
453            RefusalCode::Replay,
454            "response answers another request",
455        ));
456    }
457    if request.target != ANY_TARGET && resp.responder != request.target {
458        return Err(Refusal::new(
459            RefusalCode::NotForMe,
460            "response from another node",
461        ));
462    }
463    Ok((resp, pk))
464}
465
466#[cfg(test)]
467#[path = "tests_msg.rs"]
468mod tests;
clawft_kernel/workload_ctl/msg.rs

clawft_kernel/workload_ctl/
msg.rs
