core_ipc/
message.rs

1//! IPC message envelope.
2
3use core_types::{AgentId, DaemonId, InstallationId, SecurityLevel, Timestamp, TrustVector};
4use serde::{Deserialize, Serialize};
5use std::fmt;
6use std::time::Instant;
7use uuid::Uuid;
8
9/// Current wire format version. Increment on any field addition/removal.
10///
11/// WIRE FORMAT CONTRACT:
12///
13/// v2 fields: `wire_version`, `msg_id`, `correlation_id`, `sender`,
14/// `timestamp`, `payload`, `security_level`, `verified_sender_name`
15///
16/// All v2 binaries must be deployed atomically (single compilation unit).
17/// Adding fields requires incrementing this constant and updating the decode
18/// path to handle both old and new versions during rolling upgrades.
19pub const WIRE_VERSION: u8 = 3;
20
21/// The IPC bus message envelope wrapping any payload type.
22///
23/// Debug is manually implemented to delegate to `T`'s Debug impl.
24/// When `T = EventKind`, the payload's custom Debug redacts secret fields.
25#[derive(Clone, Serialize, Deserialize)]
26pub struct Message<T> {
27    /// Wire format version. Always serialized first.
28    /// Receivers should check this before interpreting remaining fields.
29    pub wire_version: u8,
30    /// Unique message identifier (UUID v7 for time-ordering).
31    pub msg_id: Uuid,
32    /// Correlation ID for request-response patterns.
33    pub correlation_id: Option<Uuid>,
34    /// Sender daemon identity.
35    pub sender: DaemonId,
36    /// Dual-clock timestamp.
37    pub timestamp: Timestamp,
38    /// The event or request payload.
39    pub payload: T,
40    /// Access control level for this message.
41    pub security_level: SecurityLevel,
42    /// Server-stamped verified sender name from Noise IK registry lookup.
43    ///
44    /// Set by `route_frame()` in the bus server — never trust client-supplied values.
45    /// `None` for unregistered clients (CLI, Open clearance).
46    /// Note: no `skip_serializing_if` — postcard uses positional encoding, so the
47    /// field must always be present in the wire format for decode compatibility.
48    pub verified_sender_name: Option<String>,
49
50    // -- v3 fields (appended for positional encoding safety) --
51    /// Installation identity of the sender.
52    /// No `skip_serializing_if` — postcard positional encoding requires all fields present.
53    pub origin_installation: Option<InstallationId>,
54    /// Agent identity of the sender.
55    pub agent_id: Option<AgentId>,
56    /// Trust snapshot at time of message creation.
57    pub trust_snapshot: Option<TrustVector>,
58}
59
60/// Context for constructing outbound messages.
61///
62/// Carries the sender's identity information so that `Message::new()` can
63/// populate all v3 fields without callers needing to pass them individually.
64#[derive(Debug, Clone, Serialize, Deserialize)]
65pub struct MessageContext {
66    pub sender: DaemonId,
67    pub installation: Option<InstallationId>,
68    pub agent_id: Option<AgentId>,
69    pub trust_snapshot: Option<TrustVector>,
70}
71
72impl MessageContext {
73    /// Create a minimal context with just a daemon ID (no v3 fields).
74    #[must_use]
75    pub fn new(sender: DaemonId) -> Self {
76        Self {
77            sender,
78            installation: None,
79            agent_id: None,
80            trust_snapshot: None,
81        }
82    }
83}
84
85impl<T: fmt::Debug> fmt::Debug for Message<T> {
86    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
87        f.debug_struct("Message")
88            .field("msg_id", &self.msg_id)
89            .field("correlation_id", &self.correlation_id)
90            .field("sender", &self.sender)
91            .field("security_level", &self.security_level)
92            .field("agent_id", &self.agent_id)
93            .field(
94                "origin_installation",
95                &self.origin_installation.as_ref().map(|i| &i.id),
96            )
97            .field("payload", &self.payload)
98            .finish_non_exhaustive()
99    }
100}
101
102impl<T: Serialize> Message<T> {
103    /// Create a new message with a fresh UUID v7 and current timestamp.
104    #[must_use]
105    pub fn new(
106        ctx: &MessageContext,
107        payload: T,
108        security_level: SecurityLevel,
109        epoch: Instant,
110    ) -> Self {
111        Self {
112            wire_version: WIRE_VERSION,
113            msg_id: Uuid::now_v7(),
114            correlation_id: None,
115            timestamp: Timestamp::now(epoch),
116            sender: ctx.sender,
117            payload,
118            security_level,
119            verified_sender_name: None,
120            origin_installation: ctx.installation.clone(),
121            agent_id: ctx.agent_id,
122            trust_snapshot: ctx.trust_snapshot.clone(),
123        }
124    }
125
126    /// Set a correlation ID for request-response linking.
127    #[must_use]
128    pub fn with_correlation(mut self, id: Uuid) -> Self {
129        self.correlation_id = Some(id);
130        self
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use core_types::{DaemonId, EventKind, SecurityLevel};
138
139    fn make_test_ctx() -> MessageContext {
140        MessageContext::new(DaemonId::new())
141    }
142
143    fn make_test_message() -> Message<EventKind> {
144        let ctx = make_test_ctx();
145        let payload = EventKind::StatusRequest;
146        Message::new(&ctx, payload, SecurityLevel::Internal, Instant::now())
147    }
148
149    // SECURITY INVARIANT: Message::new() must always set wire_version to the
150    // current WIRE_VERSION constant. A zero or stale version breaks decode
151    // compatibility and could bypass version-gated validation.
152    #[test]
153    fn message_new_sets_wire_version() {
154        let msg = make_test_message();
155        assert_eq!(msg.wire_version, WIRE_VERSION);
156        assert_eq!(msg.wire_version, 3);
157    }
158
159    // SECURITY INVARIANT: New messages must have no verified_sender_name (server
160    // stamps this) and no correlation_id (only responses have one). A client
161    // pre-setting verified_sender_name could impersonate another daemon.
162    #[test]
163    fn message_new_defaults_are_safe() {
164        let msg = make_test_message();
165        assert!(msg.verified_sender_name.is_none());
166        assert!(msg.correlation_id.is_none());
167        assert!(msg.origin_installation.is_none());
168        assert!(msg.agent_id.is_none());
169        assert!(msg.trust_snapshot.is_none());
170    }
171
172    // SECURITY INVARIANT: wire_version must survive postcard encode/decode
173    // roundtrip. If positional encoding drops or reorders it, version-gated
174    // fields will be misinterpreted.
175    #[test]
176    fn message_roundtrip_preserves_wire_version() {
177        let msg = make_test_message();
178        let bytes = crate::framing::encode_frame(&msg).unwrap();
179        let decoded: Message<EventKind> = crate::framing::decode_frame(&bytes).unwrap();
180        assert_eq!(decoded.wire_version, WIRE_VERSION);
181    }
182
183    #[test]
184    fn with_correlation_sets_id() {
185        let msg = make_test_message();
186        let corr_id = Uuid::now_v7();
187        let msg = msg.with_correlation(corr_id);
188        assert_eq!(msg.correlation_id, Some(corr_id));
189    }
190
191    #[test]
192    fn v3_fields_populated_roundtrip() {
193        let ctx = MessageContext {
194            sender: DaemonId::new(),
195            installation: Some(InstallationId {
196                id: Uuid::from_u128(1),
197                org_ns: None,
198                namespace: Uuid::from_u128(2),
199                machine_binding: None,
200            }),
201            agent_id: Some(AgentId::from_uuid(Uuid::from_u128(3))),
202            trust_snapshot: Some(TrustVector {
203                authn_strength: core_types::TrustLevel::High,
204                authz_freshness: std::time::Duration::from_secs(30),
205                delegation_depth: 1,
206                device_posture: 0.9,
207                network_exposure: core_types::NetworkTrust::Local,
208                agent_type: core_types::AgentType::Human,
209            }),
210        };
211        let msg = Message::new(
212            &ctx,
213            EventKind::StatusRequest,
214            SecurityLevel::Internal,
215            Instant::now(),
216        );
217
218        let bytes = crate::framing::encode_frame(&msg).unwrap();
219        let decoded: Message<EventKind> = crate::framing::decode_frame(&bytes).unwrap();
220
221        assert_eq!(decoded.wire_version, WIRE_VERSION);
222        assert_eq!(
223            decoded.origin_installation.as_ref().unwrap().id,
224            Uuid::from_u128(1)
225        );
226        assert_eq!(
227            decoded.agent_id.unwrap(),
228            AgentId::from_uuid(Uuid::from_u128(3))
229        );
230        assert!(decoded.trust_snapshot.is_some());
231        let tv = decoded.trust_snapshot.unwrap();
232        assert_eq!(tv.authn_strength, core_types::TrustLevel::High);
233        assert_eq!(tv.delegation_depth, 1);
234    }
235
236    #[test]
237    fn v3_fields_none_roundtrip() {
238        let ctx = MessageContext::new(DaemonId::new());
239        let msg = Message::new(
240            &ctx,
241            EventKind::StatusRequest,
242            SecurityLevel::Open,
243            Instant::now(),
244        );
245
246        let bytes = crate::framing::encode_frame(&msg).unwrap();
247        let decoded: Message<EventKind> = crate::framing::decode_frame(&bytes).unwrap();
248
249        assert!(decoded.origin_installation.is_none());
250        assert!(decoded.agent_id.is_none());
251        assert!(decoded.trust_snapshot.is_none());
252    }
253
254    #[test]
255    fn v3_event_variants_roundtrip() {
256        use core_types::*;
257
258        let ctx = MessageContext::new(DaemonId::new());
259        let epoch = Instant::now();
260
261        let agent = AgentId::from_uuid(Uuid::from_u128(10));
262        let installation = InstallationId {
263            id: Uuid::from_u128(20),
264            org_ns: None,
265            namespace: Uuid::from_u128(30),
266            machine_binding: None,
267        };
268        let req_id = Uuid::from_u128(40);
269        let deleg_id = Uuid::from_u128(50);
270        let session_id = Uuid::from_u128(60);
271        let ts = Timestamp::now(epoch);
272
273        let variants: Vec<EventKind> = vec![
274            EventKind::AgentConnected {
275                agent_id: agent,
276                agent_type: AgentType::Human,
277                attestations: vec![AttestationType::UCred],
278            },
279            EventKind::AgentDisconnected {
280                agent_id: agent,
281                reason: "done".into(),
282            },
283            EventKind::InstallationCreated {
284                id: installation.clone(),
285                org: None,
286                machine_binding_present: false,
287            },
288            EventKind::ProfileIdMigrated {
289                name: TrustProfileName::try_from("work").unwrap(),
290                old_id: ProfileId::from_uuid(Uuid::from_u128(1)),
291                new_id: ProfileId::from_uuid(Uuid::from_u128(2)),
292            },
293            EventKind::AuthorizationRequired {
294                request_id: req_id,
295                operation: "secret.read".into(),
296                missing_attestations: vec![],
297                expires_at: ts.clone(),
298            },
299            EventKind::AuthorizationGrant {
300                request_id: req_id,
301                delegator: agent,
302                scope: CapabilitySet::empty(),
303                ttl_seconds: 300,
304                point_of_use_filter: None,
305            },
306            EventKind::AuthorizationDenied {
307                request_id: req_id,
308                reason: "nope".into(),
309            },
310            EventKind::AuthorizationTimeout { request_id: req_id },
311            EventKind::DelegationRevoked {
312                delegation_id: deleg_id,
313                revoker: agent,
314                reason: "expired".into(),
315            },
316            EventKind::HeartbeatRenewed {
317                delegation_id: deleg_id,
318                renewal_source: agent,
319                next_deadline: ts,
320            },
321            EventKind::FederationSessionEstablished {
322                session_id,
323                remote_installation: installation,
324            },
325            EventKind::FederationSessionTerminated {
326                session_id,
327                reason: "closed".into(),
328            },
329            EventKind::PostureEvaluated {
330                secure_boot: Some(true),
331                disk_encrypted: Some(true),
332                screen_locked: None,
333                composite_score: 0.95,
334            },
335        ];
336
337        for (i, variant) in variants.into_iter().enumerate() {
338            let msg = Message::new(&ctx, variant, SecurityLevel::Internal, epoch);
339            let bytes = crate::framing::encode_frame(&msg)
340                .unwrap_or_else(|e| panic!("encode failed for variant {i}: {e}"));
341            let decoded: Message<EventKind> = crate::framing::decode_frame(&bytes)
342                .unwrap_or_else(|e| panic!("decode failed for variant {i}: {e}"));
343            assert_eq!(
344                decoded.wire_version, WIRE_VERSION,
345                "variant {i} wire version mismatch"
346            );
347        }
348    }
349}
core_ipc/message.rs

core_ipc/
message.rs
