1use chrono::Utc;
8use serde::Deserialize;
9use serde_json::Value;
10use sha2::{Digest, Sha256};
11use std::collections::{HashMap, VecDeque};
12use std::fmt::Write as _;
13use std::fs;
14use std::sync::{LazyLock, Mutex};
15use uuid::Uuid;
16use wasmtime::{
17 Caller, Config, Engine, Extern, Linker, Module, Store, StoreLimits, StoreLimitsBuilder,
18};
19use wasmtime_wasi::WasiCtxBuilder;
20use wasmtime_wasi::p1::WasiP1Ctx;
21use wasmtime_wasi::p2::pipe::{MemoryInputPipe, MemoryOutputPipe};
22
23use super::{ArtifactType, CapabilityExecutor, ExecutorCapability, ExecutorError, ExecutorOutput};
24use crate::events::types::{LifecycleStatus, TraverseEvent};
25use traverse_contracts::{EventReference, ServiceType};
26
27pub const SUPPORTED_HOST_ABI_VERSION: &str = "1.0.0";
29
30const HOST_ABI_V1_WHITELIST: &str = include_str!("host_abi_v1.json");
31const DEFAULT_FUEL_BUDGET: u64 = 5_000_000;
32const DEFAULT_MEMORY_LIMIT_BYTES: usize = 8 * 1024 * 1024;
33const DEFAULT_TABLE_ELEMENT_LIMIT: usize = 1_024;
34const DEFAULT_INSTANCE_LIMIT: usize = 1;
35const DEFAULT_TABLE_LIMIT: usize = 8;
36const DEFAULT_LINEAR_MEMORY_LIMIT: usize = 1;
37const DEFAULT_MODULE_CACHE_MAX_ENTRIES: usize = 64;
38
39const MAX_EVENT_EMIT_PAYLOAD_BYTES: usize = 64 * 1024;
43
44const EMIT_EVENT_OK: i32 = 0;
47const EMIT_EVENT_ERR_INVALID_PAYLOAD: i32 = -1;
52const EMIT_EVENT_ERR_UNDECLARED_EVENT: i32 = -2;
55const EMIT_EVENT_ERR_NOT_SUBSCRIBABLE: i32 = -3;
58
59static HOST_ABI_V1_WHITELIST_CACHE: LazyLock<Result<HostAbiWhitelist, String>> =
60 LazyLock::new(|| {
61 serde_json::from_str::<HostAbiWhitelist>(HOST_ABI_V1_WHITELIST).map_err(|e| e.to_string())
62 });
63
64#[derive(Debug, Clone, PartialEq, Eq)]
66pub struct HostAbiImport {
67 pub module: String,
69 pub name: String,
71}
72
73#[derive(Debug, Clone, PartialEq, Eq)]
75pub struct HostAbiValidation {
76 pub abi_version: String,
78 pub imports: Vec<HostAbiImport>,
80}
81
82#[derive(Debug, Clone, Deserialize)]
83struct HostAbiWhitelist {
84 abi_version: String,
85 imports: Vec<HostAbiWhitelistImport>,
86}
87
88#[derive(Debug, Clone, Deserialize)]
89struct HostAbiWhitelistImport {
90 module: String,
91 name: String,
92}
93
94#[must_use]
96pub fn supported_host_abi_versions() -> &'static [&'static str] {
97 &[SUPPORTED_HOST_ABI_VERSION]
98}
99
100pub fn verify_wasm_host_abi_bytes(
107 wasm_bytes: &[u8],
108 abi_version: &str,
109) -> Result<HostAbiValidation, ExecutorError> {
110 let engine = Engine::default();
111 let module = Module::from_binary(&engine, wasm_bytes).map_err(|e| {
112 ExecutorError::MalformedWasmArtifact {
113 error_code: "malformed_wasm_artifact".to_string(),
114 detail: format!("module compile: {e}"),
115 }
116 })?;
117 validate_module_imports(&module, abi_version)
118}
119
120#[derive(Debug)]
124pub struct WasmExecutor {
125 engine: Engine,
126 limits: WasmExecutionLimits,
127 module_cache: Mutex<CompiledModuleCache>,
128}
129
130impl WasmExecutor {
131 pub fn new() -> Result<Self, ExecutorError> {
137 Self::with_limits(WasmExecutionLimits::default())
138 }
139
140 pub fn with_limits(limits: WasmExecutionLimits) -> Result<Self, ExecutorError> {
146 Self::with_limits_and_cache_config(limits, WasmModuleCacheConfig::default())
147 }
148
149 pub fn with_limits_and_cache_config(
155 limits: WasmExecutionLimits,
156 cache_config: WasmModuleCacheConfig,
157 ) -> Result<Self, ExecutorError> {
158 let mut config = Config::new();
159 config.consume_fuel(true);
160 let engine = Engine::new(&config)
161 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("engine config: {e}")))?;
162 Ok(Self {
163 engine,
164 limits,
165 module_cache: Mutex::new(CompiledModuleCache::new(cache_config.max_entries)),
166 })
167 }
168
169 #[must_use]
171 pub fn module_cache_stats(&self) -> WasmModuleCacheStats {
172 let cache = self
173 .module_cache
174 .lock()
175 .unwrap_or_else(std::sync::PoisonError::into_inner);
176 cache.stats()
177 }
178}
179
180#[derive(Debug, Clone, Copy, PartialEq, Eq)]
182pub struct WasmExecutionLimits {
183 pub fuel_budget: u64,
185 pub memory_bytes: usize,
187 pub table_elements: usize,
189 pub instances: usize,
191 pub tables: usize,
193 pub memories: usize,
195}
196
197impl Default for WasmExecutionLimits {
198 fn default() -> Self {
199 Self {
200 fuel_budget: DEFAULT_FUEL_BUDGET,
201 memory_bytes: DEFAULT_MEMORY_LIMIT_BYTES,
202 table_elements: DEFAULT_TABLE_ELEMENT_LIMIT,
203 instances: DEFAULT_INSTANCE_LIMIT,
204 tables: DEFAULT_TABLE_LIMIT,
205 memories: DEFAULT_LINEAR_MEMORY_LIMIT,
206 }
207 }
208}
209
210#[derive(Debug, Clone, Copy, PartialEq, Eq)]
212pub struct WasmModuleCacheConfig {
213 pub max_entries: usize,
215}
216
217impl Default for WasmModuleCacheConfig {
218 fn default() -> Self {
219 Self {
220 max_entries: DEFAULT_MODULE_CACHE_MAX_ENTRIES,
221 }
222 }
223}
224
225#[derive(Debug, Clone, Copy, PartialEq, Eq)]
227pub struct WasmModuleCacheStats {
228 pub entries: usize,
230 pub hits: u64,
232 pub misses: u64,
234 pub evictions: u64,
236}
237
238#[derive(Debug, Clone)]
239struct CachedModule {
240 module: Module,
241 validation: HostAbiValidation,
242}
243
244#[derive(Debug)]
245struct CompiledModuleCache {
246 max_entries: usize,
247 entries: HashMap<String, CachedModule>,
248 insertion_order: VecDeque<String>,
249 hits: u64,
250 misses: u64,
251 evictions: u64,
252}
253
254impl CompiledModuleCache {
255 fn new(max_entries: usize) -> Self {
256 Self {
257 max_entries: max_entries.max(1),
258 entries: HashMap::new(),
259 insertion_order: VecDeque::new(),
260 hits: 0,
261 misses: 0,
262 evictions: 0,
263 }
264 }
265
266 fn get(&mut self, checksum: &str, abi_version: &str) -> Option<CachedModule> {
267 let cached = self.entries.get(checksum)?;
268 if cached.validation.abi_version != abi_version {
269 self.misses += 1;
270 return None;
271 }
272 self.hits += 1;
273 Some(cached.clone())
274 }
275
276 fn insert(&mut self, checksum: String, cached: CachedModule) {
277 self.misses += 1;
278 while self.entries.len() >= self.max_entries {
279 if let Some(oldest) = self.insertion_order.pop_front()
280 && self.entries.remove(&oldest).is_some()
281 {
282 self.evictions += 1;
283 }
284 }
285 self.insertion_order.push_back(checksum.clone());
286 self.entries.insert(checksum, cached);
287 }
288
289 fn stats(&self) -> WasmModuleCacheStats {
290 WasmModuleCacheStats {
291 entries: self.entries.len(),
292 hits: self.hits,
293 misses: self.misses,
294 evictions: self.evictions,
295 }
296 }
297}
298
299struct WasmStoreState {
300 wasi: WasiP1Ctx,
301 limits: StoreLimits,
302 capability_id: String,
307 emits: Vec<EventReference>,
308 service_type: ServiceType,
309 emitted_events: Vec<TraverseEvent>,
311}
312
313impl CapabilityExecutor for WasmExecutor {
314 fn execute(
315 &self,
316 capability: &ExecutorCapability,
317 input: &Value,
318 ) -> Result<ExecutorOutput, ExecutorError> {
319 if capability.artifact_type != ArtifactType::Wasm {
320 return Err(ExecutorError::UnsupportedArtifactType);
321 }
322
323 let wasm_path = capability.wasm_binary_path.as_deref().ok_or_else(|| {
325 ExecutorError::BinaryLoadFailed("no wasm_binary_path set".to_string())
326 })?;
327
328 let binary = fs::read(wasm_path).map_err(|e| {
329 ExecutorError::BinaryLoadFailed(format!("cannot read {wasm_path}: {e}"))
330 })?;
331
332 if let Some(expected) = capability.wasm_checksum.as_deref() {
334 let actual = sha256_hex(&binary);
335 if actual != expected {
336 return Err(ExecutorError::ChecksumMismatch {
337 expected: expected.to_string(),
338 actual,
339 });
340 }
341 }
342
343 let abi_version = capability
344 .host_abi_version
345 .as_deref()
346 .unwrap_or(SUPPORTED_HOST_ABI_VERSION);
347
348 self.run_wasm(
349 &binary,
350 input,
351 abi_version,
352 &capability.capability_id,
353 &capability.emits,
354 capability.service_type.clone(),
355 )
356 }
357}
358
359impl WasmExecutor {
360 pub fn run_bytes(&self, wasm_bytes: &[u8], input: &Value) -> Result<Value, ExecutorError> {
373 self.run_bytes_with_host_abi(wasm_bytes, input, SUPPORTED_HOST_ABI_VERSION)
374 }
375
376 pub fn run_bytes_with_host_abi(
382 &self,
383 wasm_bytes: &[u8],
384 input: &Value,
385 abi_version: &str,
386 ) -> Result<Value, ExecutorError> {
387 self.run_wasm(
388 wasm_bytes,
389 input,
390 abi_version,
391 "test-capability",
392 &[],
393 ServiceType::Stateless,
394 )
395 .map(|output| output.value)
396 }
397
398 pub fn run_bytes_with_capability(
406 &self,
407 wasm_bytes: &[u8],
408 input: &Value,
409 capability_id: &str,
410 emits: &[EventReference],
411 service_type: ServiceType,
412 ) -> Result<ExecutorOutput, ExecutorError> {
413 self.run_wasm(
414 wasm_bytes,
415 input,
416 SUPPORTED_HOST_ABI_VERSION,
417 capability_id,
418 emits,
419 service_type,
420 )
421 }
422
423 #[allow(clippy::too_many_arguments)]
424 fn run_wasm(
425 &self,
426 wasm_bytes: &[u8],
427 input: &Value,
428 abi_version: &str,
429 capability_id: &str,
430 emits: &[EventReference],
431 service_type: ServiceType,
432 ) -> Result<ExecutorOutput, ExecutorError> {
433 let input_json = serde_json::to_string(input)
434 .map_err(|e| ExecutorError::ExecutionFailed(format!("input serialization: {e}")))?;
435
436 let cached_module = self.compiled_module(wasm_bytes, abi_version)?;
437
438 let stdout_pipe = MemoryOutputPipe::new(65536);
440 let stdout_ref = stdout_pipe.clone();
441
442 let wasi_ctx: WasiP1Ctx = WasiCtxBuilder::new()
445 .stdin(MemoryInputPipe::new(input_json.into_bytes()))
446 .stdout(stdout_pipe)
447 .build_p1();
448
449 let mut linker: Linker<WasmStoreState> = Linker::new(&self.engine);
450 wasmtime_wasi::p1::add_to_linker_sync(&mut linker, |s| &mut s.wasi)
451 .map_err(|e| ExecutorError::RuntimeSetupFailed(e.to_string()))?;
452 linker
453 .func_wrap("traverse_host", "emit_event", handle_emit_event)
454 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("func_wrap emit_event: {e}")))?;
455
456 let mut store = Store::new(
457 &self.engine,
458 WasmStoreState {
459 wasi: wasi_ctx,
460 limits: self.store_limits(),
461 capability_id: capability_id.to_string(),
462 emits: emits.to_vec(),
463 service_type,
464 emitted_events: Vec::new(),
465 },
466 );
467 store.limiter(|state| &mut state.limits);
468 store
469 .set_fuel(self.limits.fuel_budget)
470 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("set fuel: {e}")))?;
471
472 linker
473 .module(&mut store, "", &cached_module.module)
474 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("module link: {e}")))?;
475
476 linker
477 .get_default(&mut store, "")
478 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("get_default: {e}")))?
479 .typed::<(), ()>(&store)
480 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("typed: {e}")))?
481 .call(&mut store, ())
482 .map_err(|error| classify_wasm_execution_error(&error))?;
483
484 let raw_output = stdout_ref.contents();
486
487 let value = serde_json::from_slice::<Value>(&raw_output).map_err(|e| {
488 ExecutorError::OutputDeserializationFailed(format!(
489 "stdout is not valid JSON: {e} — raw: {}",
490 String::from_utf8_lossy(&raw_output)
491 ))
492 })?;
493
494 Ok(ExecutorOutput {
495 value,
496 emitted_events: store.into_data().emitted_events,
497 })
498 }
499
500 fn store_limits(&self) -> StoreLimits {
501 StoreLimitsBuilder::new()
502 .memory_size(self.limits.memory_bytes)
503 .table_elements(self.limits.table_elements)
504 .instances(self.limits.instances)
505 .tables(self.limits.tables)
506 .memories(self.limits.memories)
507 .trap_on_grow_failure(true)
508 .build()
509 }
510
511 fn compiled_module(
512 &self,
513 wasm_bytes: &[u8],
514 abi_version: &str,
515 ) -> Result<CachedModule, ExecutorError> {
516 let checksum = sha256_hex(wasm_bytes);
517 {
518 let mut cache = self
519 .module_cache
520 .lock()
521 .unwrap_or_else(std::sync::PoisonError::into_inner);
522 if let Some(cached) = cache.get(&checksum, abi_version) {
523 return Ok(cached);
524 }
525 }
526
527 let module = Module::from_binary(&self.engine, wasm_bytes).map_err(|e| {
528 ExecutorError::MalformedWasmArtifact {
529 error_code: "malformed_wasm_artifact".to_string(),
530 detail: format!("module compile: {e}"),
531 }
532 })?;
533 let validation = validate_module_imports(&module, abi_version)?;
534 let cached = CachedModule { module, validation };
535
536 let mut cache = self
537 .module_cache
538 .lock()
539 .unwrap_or_else(std::sync::PoisonError::into_inner);
540 cache.insert(checksum, cached.clone());
541 Ok(cached)
542 }
543}
544
545fn handle_emit_event(mut caller: Caller<'_, WasmStoreState>, ptr: i32, len: i32) -> i32 {
553 if caller.data().service_type != ServiceType::Subscribable {
556 return EMIT_EVENT_ERR_NOT_SUBSCRIBABLE;
557 }
558
559 if ptr < 0 || len < 0 {
561 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
562 }
563 #[allow(clippy::cast_sign_loss)]
564 let (ptr, len) = (ptr as usize, len as usize);
565 if len > MAX_EVENT_EMIT_PAYLOAD_BYTES {
566 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
567 }
568
569 let Some(Extern::Memory(memory)) = caller.get_export("memory") else {
570 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
571 };
572
573 let mut buffer = vec![0u8; len];
574 if memory.read(&caller, ptr, &mut buffer).is_err() {
577 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
578 }
579
580 let Ok(payload) = serde_json::from_slice::<Value>(&buffer) else {
581 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
582 };
583 let Some(event_type) = payload
584 .get("event_id")
585 .and_then(Value::as_str)
586 .map(str::to_string)
587 else {
588 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
589 };
590 let Some(version) = payload
591 .get("version")
592 .and_then(Value::as_str)
593 .map(str::to_string)
594 else {
595 return EMIT_EVENT_ERR_INVALID_PAYLOAD;
596 };
597 let data = payload
598 .get("payload")
599 .cloned()
600 .unwrap_or_else(|| Value::Object(serde_json::Map::new()));
601
602 let declared = caller
604 .data()
605 .emits
606 .iter()
607 .any(|decl| decl.event_id == event_type && decl.version == version);
608 if !declared {
609 return EMIT_EVENT_ERR_UNDECLARED_EVENT;
610 }
611
612 let capability_id = caller.data().capability_id.clone();
613 let event = TraverseEvent {
614 id: Uuid::new_v4().to_string(),
615 source: format!("traverse-runtime/{capability_id}"),
616 event_type: event_type.clone(),
617 datacontenttype: "application/json".to_string(),
618 time: Utc::now().to_rfc3339(),
619 data,
620 owner: capability_id.clone(),
621 version: version.clone(),
622 lifecycle_status: LifecycleStatus::Active,
623 deduplication_id: Some(format!("{capability_id}:{event_type}:{version}")),
624 ordering_scope: Some(capability_id),
625 correlation_id: None,
626 causation_id: None,
627 subject_id: None,
628 actor_id: None,
629 };
630 caller.data_mut().emitted_events.push(event);
631 EMIT_EVENT_OK
632}
633
634fn classify_wasm_execution_error(error: &wasmtime::Error) -> ExecutorError {
635 let display = error.to_string();
636 let debug = format!("{error:?}");
637 if display.contains("all fuel consumed by WebAssembly")
638 || debug.contains("all fuel consumed by WebAssembly")
639 {
640 return ExecutorError::Timeout(debug);
641 }
642 if display.contains("forcing trap when growing") || debug.contains("forcing trap when growing")
643 {
644 return ExecutorError::ResourceExhausted(debug);
645 }
646 ExecutorError::ExecutionFailed(display)
647}
648
649fn sha256_hex(data: &[u8]) -> String {
650 let mut hasher = Sha256::new();
651 hasher.update(data);
652 hasher
653 .finalize()
654 .iter()
655 .fold(String::new(), |mut acc, byte| {
656 let _ = write!(acc, "{byte:02x}");
657 acc
658 })
659}
660
661fn validate_module_imports(
662 module: &Module,
663 abi_version: &str,
664) -> Result<HostAbiValidation, ExecutorError> {
665 let whitelist = host_abi_whitelist(abi_version)?;
666 let mut imports = module
667 .imports()
668 .map(|import| HostAbiImport {
669 module: import.module().to_string(),
670 name: import.name().to_string(),
671 })
672 .collect::<Vec<_>>();
673 imports.sort_by(|a, b| a.module.cmp(&b.module).then_with(|| a.name.cmp(&b.name)));
674
675 for import in &imports {
676 if !whitelist
677 .imports
678 .iter()
679 .any(|allowed| allowed.module == import.module && allowed.name == import.name)
680 {
681 return Err(ExecutorError::UnauthorizedHostImport {
682 error_code: "unauthorized_host_import".to_string(),
683 abi_version: abi_version.to_string(),
684 module: import.module.clone(),
685 name: import.name.clone(),
686 });
687 }
688 }
689
690 Ok(HostAbiValidation {
691 abi_version: whitelist.abi_version,
692 imports,
693 })
694}
695
696fn host_abi_whitelist(abi_version: &str) -> Result<HostAbiWhitelist, ExecutorError> {
697 if abi_version != SUPPORTED_HOST_ABI_VERSION {
698 return Err(ExecutorError::UnsupportedAbiVersion {
699 error_code: "unsupported_abi_version".to_string(),
700 requested: abi_version.to_string(),
701 supported: supported_host_abi_versions().join(", "),
702 });
703 }
704
705 HOST_ABI_V1_WHITELIST_CACHE
706 .as_ref()
707 .cloned()
708 .map_err(|e| ExecutorError::RuntimeSetupFailed(format!("invalid ABI whitelist: {e}")))
709}