1. 1//! Universal CloudEvents 1.0 envelope with W3C Distributed Tracing and Merkle causal hash chaining.
2. 2
3. 3use chrono::{DateTime, Utc};
4. 4use serde::{Deserialize, Serialize};
5. 5use sha2::{Digest, Sha256};
6. 6use uuid::Uuid;
7. 7
8. 8/// W3C Distributed Trace Context.
9. 9#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
10. 10pub struct TraceContext {
11. 11 /// W3C traceparent string: "00-<trace_id_32_hex>-<parent_id_16_hex>-<trace_flags_2_hex>".
12. 12 pub traceparent: String,
13. 13 /// Optional W3C tracestate vendor key-value pairs.
14. 14 pub tracestate: Option<String>,
15. 15}
16. 16
17. 17impl TraceContext {
18. 18 /// Generate a new root W3C trace context.
19. 19 pub fn new_root() -> Self {
20. 20 let trace_id = format!("{:032x}", Uuid::now_v7().as_u128());
21. 21 let parent_id = format!("{:016x}", (Uuid::now_v7().as_u128() & 0xFFFFFFFFFFFFFFFF));
22. 22 let traceparent = format!("00-{trace_id}-{parent_id}-01");
23. 23 Self {
24. 24 traceparent,
25. 25 tracestate: None,
26. 26 }
27. 27 }
28. 28
29. 29 /// Create a child trace context inheriting the parent trace_id.
30. 30 pub fn child_span(&self) -> Self {
31. 31 let parts: Vec<&str> = self.traceparent.split('-').collect();
32. 32 if parts.len() == 4 {
33. 33 let trace_id = parts[1];
34. 34 let new_span_id = format!("{:016x}", (Uuid::now_v7().as_u128() & 0xFFFFFFFFFFFFFFFF));
35. 35 let traceparent = format!("00-{trace_id}-{new_span_id}-01");
36. 36 Self {
37. 37 traceparent,
38. 38 tracestate: self.tracestate.clone(),
39. 39 }
40. 40 } else {
41. 41 Self::new_root()
42. 42 }
43. 43 }
44. 44
45. 45 /// Extract the 32-character hex trace_id.
46. 46 pub fn trace_id(&self) -> &str {
47. 47 let parts: Vec<&str> = self.traceparent.split('-').collect();
48. 48 if parts.len() == 4 { parts[1] } else { "" }
49. 49 }
50. 50
51. 51 /// Extract the 16-character hex span_id.
52. 52 pub fn span_id(&self) -> &str {
53. 53 let parts: Vec<&str> = self.traceparent.split('-').collect();
54. 54 if parts.len() == 4 { parts[2] } else { "" }
55. 55 }
56. 56}
57. 57
58. 58/// Metadata describing payload encryption.
59. 59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60. 60pub struct EncryptionMeta {
61. 61 /// Algorithm identifier (e.g. "AES-256-GCM").
62. 62 pub algorithm: String,
63. 63 /// Key identifier used for derivation.
64. 64 pub key_id: String,
65. 65 /// Initialization vector / nonce hex string (12 bytes / 24 hex characters).
66. 66 pub nonce_hex: String,
67. 67}
68. 68
69. 69/// Causal lineage and Merkle chain coordinates.
70. 70#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
71. 71pub struct CausalLineage {
72. 72 pub workspace_id: String,
73. 73 #[serde(skip_serializing_if = "Option::is_none")]
74. 74 pub session_id: Option<String>,
75. 75 #[serde(skip_serializing_if = "Option::is_none")]
76. 76 pub parent_agent_id: Option<String>,
77. 77 pub origin_agent_id: String,
78. 78 pub seq_num: u64,
79. 79 pub prev_event_hash: String,
80. 80 pub payload_hash: String,
81. 81}
82. 82
83. 83/// Universal CloudEvents-compatible message envelope.
84. 84#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85. 85pub struct MessageEnvelope {
86. 86 pub specversion: String,
87. 87 pub id: String,
88. 88 pub source: String,
89. 89 #[serde(rename = "type")]
90. 90 pub event_type: String,
91. 91 pub time: DateTime<Utc>,
92. 92 pub datacontenttype: String,
93. 93 pub trace: TraceContext,
94. 94 pub lineage: CausalLineage,
95. 95 pub payload: Vec<u8>,
96. 96 pub encrypted: bool,
97. 97 #[serde(skip_serializing_if = "Option::is_none")]
98. 98 pub encryption: Option<EncryptionMeta>,
99. 99}
100. 100
101. 101impl MessageEnvelope {
102. 102 /// Genesis hash used for the first event in a stream or session.
103. 103 pub const GENESIS_HASH: &'static str =
104. 104 "0000000000000000000000000000000000000000000000000000000000000000";
105. 105
106. 106 /// Construct a new unencrypted message envelope.
107. 107 #[allow(clippy::too_many_arguments)]
108. 108 pub fn new(
109. 109 source: impl Into<String>,
110. 110 event_type: impl Into<String>,
111. 111 workspace_id: impl Into<String>,
112. 112 origin_agent_id: impl Into<String>,
113. 113 seq_num: u64,
114. 114 prev_event_hash: impl Into<String>,
115. 115 payload: Vec<u8>,
116. 116 trace: Option<TraceContext>,
117. 117 ) -> Self {
118. 118 let mut hasher = Sha256::new();
119. 119 hasher.update(&payload);
120. 120 let payload_hash = hex::encode(hasher.finalize());
121. 121
122. 122 let lineage = CausalLineage {
123. 123 workspace_id: workspace_id.into(),
124. 124 session_id: None,
125. 125 parent_agent_id: None,
126. 126 origin_agent_id: origin_agent_id.into(),
127. 127 seq_num,
128. 128 prev_event_hash: prev_event_hash.into(),
129. 129 payload_hash,
130. 130 };
131. 131
132. 132 Self {
133. 133 specversion: "1.0".to_string(),
134. 134 id: Uuid::now_v7().to_string(),
135. 135 source: source.into(),
136. 136 event_type: event_type.into(),
137. 137 time: Utc::now(),
138. 138 datacontenttype: "application/json".to_string(),
139. 139 trace: trace.unwrap_or_else(TraceContext::new_root),
140. 140 lineage,
141. 141 payload,
142. 142 encrypted: false,
143. 143 encryption: None,
144. 144 }
145. 145 }
146. 146
147. 147 /// Attach session and parent context.
148. 148 pub fn with_session(mut self, session_id: impl Into<String>) -> Self {
149. 149 self.lineage.session_id = Some(session_id.into());
150. 150 self
151. 151 }
152. 152
153. 153 pub fn with_parent_agent(mut self, parent_agent_id: impl Into<String>) -> Self {
154. 154 self.lineage.parent_agent_id = Some(parent_agent_id.into());
155. 155 self
156. 156 }
157. 157
158. 158 /// Compute cryptographic hash of envelope headers and payload hash (Merkle leaf).
159. 159 pub fn compute_hash(&self) -> String {
160. 160 let mut hasher = Sha256::new();
161. 161 hasher.update(self.specversion.as_bytes());
162. 162 hasher.update(b"|");
163. 163 hasher.update(self.id.as_bytes());
164. 164 hasher.update(b"|");
165. 165 hasher.update(self.source.as_bytes());
166. 166 hasher.update(b"|");
167. 167 hasher.update(self.event_type.as_bytes());
168. 168 hasher.update(b"|");
169. 169 hasher.update(self.time.to_rfc3339().as_bytes());
170. 170 hasher.update(b"|");
171. 171 hasher.update(self.lineage.workspace_id.as_bytes());
172. 172 hasher.update(b"|");
173. 173 hasher.update(self.lineage.origin_agent_id.as_bytes());
174. 174 hasher.update(b"|");
175. 175 hasher.update(self.lineage.seq_num.to_be_bytes());
176. 176 hasher.update(b"|");
177. 177 hasher.update(self.lineage.prev_event_hash.as_bytes());
178. 178 hasher.update(b"|");
179. 179 hasher.update(self.lineage.payload_hash.as_bytes());
180. 180 hex::encode(hasher.finalize())
181. 181 }
182. 182
183. 183 /// Verify causal link between previous and current envelope in a stream.
184. 184 pub fn verify_merkle_link(prev: &MessageEnvelope, current: &MessageEnvelope) -> bool {
185. 185 if current.lineage.seq_num != prev.lineage.seq_num + 1 {
186. 186 return false;
187. 187 }
188. 188 let expected_prev_hash = prev.compute_hash();
189. 189 subtle::ConstantTimeEq::ct_eq(
190. 190 current.lineage.prev_event_hash.as_bytes(),
191. 191 expected_prev_hash.as_bytes(),
192. 192 )
193. 193 .into()
194. 194 }
195. 195
196. 196 /// Additional Authenticated Data (AAD) for AES-256-GCM encryption.
197. 197 pub fn aad_bytes(&self) -> Vec<u8> {
198. 198 format!(
199. 199 "{}:{}:{}:{}",
200. 200 self.id, self.event_type, self.lineage.workspace_id, self.lineage.seq_num
201. 201 )
202. 202 .into_bytes()
203. 203 }
204. 204}
205. 205