diff --git a/Cargo.lock b/Cargo.lock index 7d7e48e6..948ce5ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -530,6 +530,15 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +[[package]] +name = "encoding_rs" +version = "0.8.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +dependencies = [ + "cfg-if", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -924,6 +933,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + [[package]] name = "idna" version = "1.1.0" @@ -1815,6 +1830,20 @@ dependencies = [ "sha2", ] +[[package]] +name = "st2-resource-wasip2" +version = "0.1.0" +dependencies = [ + "libc", + "serde", + "serde_json", + "sha2", + "st2-resource-protocol", + "tempfile", + "wasmtime", + "wat", +] + [[package]] name = "st2-wire" version = "0.1.0" @@ -2341,6 +2370,18 @@ dependencies = [ "wasmparser 0.258.0", ] +[[package]] +name = "wasm-metadata" +version = "0.254.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b01df5f3b4ca7881e843f3bc0fb8a3905d79c68692250dcb8e33e698705ccdb6" +dependencies = [ + "anyhow", + "indexmap", + "wasm-encoder 0.254.0", + "wasmparser 0.254.0", +] + [[package]] name = "wasmparser" version = "0.254.0" @@ -2387,6 +2428,7 @@ dependencies = [ "bitflags", "bumpalo", "cc", + "encoding_rs", "futures", "libc", "log", @@ -2397,12 +2439,15 @@ dependencies = [ "postcard", "pulley-interpreter", "rustix", + "semver", "serde", "serde_derive", "smallvec", "target-lexicon", "wasmparser 0.254.0", "wasmtime-environ", + "wasmtime-internal-component-macro", + "wasmtime-internal-component-util", "wasmtime-internal-core", "wasmtime-internal-cranelift", "wasmtime-internal-fiber", @@ -2445,6 +2490,21 @@ dependencies = [ "wasmtime-internal-core", ] +[[package]] +name = "wasmtime-internal-component-macro" +version = "48.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3d2b2f9f89a65bb2277d82fbdd49263d27d8233bc717d750c4b13a23d57721c" +dependencies = [ + "anyhow", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasmtime-internal-component-util", + "wasmtime-internal-wit-bindgen", + "wit-parser", +] + [[package]] name = "wasmtime-internal-component-util" version = "48.0.1" @@ -2546,6 +2606,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "wasmtime-internal-wit-bindgen" +version = "48.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07429ab53576255be2d298e04b2e1f381e39645465adbaebd69314b5430165c0" +dependencies = [ + "anyhow", + "bitflags", + "heck", + "indexmap", + "wit-component", + "wit-parser", +] + [[package]] name = "wast" version = "258.0.0" @@ -2707,6 +2781,44 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "wit-component" +version = "0.254.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0e65bb94c369b3c4741ce3d1d2704b1fec93db7c540df0e521a097e7ceeb5be" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder 0.254.0", + "wasm-metadata", + "wasmparser 0.254.0", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.254.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1655131e4f7d3f0cb141f6eca71315ca40eff0f3d4de7cff0a82bacedd8c89b4" +dependencies = [ + "anyhow", + "hashbrown 0.17.1", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-ident", + "wasmparser 0.254.0", +] + [[package]] name = "writeable" version = "0.6.4" diff --git a/Cargo.toml b/Cargo.toml index 167ea193..478d3ae1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ # `package.version` out of this file as the single source of truth for the # build, and a virtual root has no `[package]` to read. [workspace] -members = ["crates/agent-spec", "crates/st2-resource-protocol", "crates/st2-wire", "crates/demo-resolver-wasm"] +members = ["crates/agent-spec", "crates/st2-resource-protocol", "crates/st2-resource-wasip2", "crates/st2-wire", "crates/demo-resolver-wasm"] default-members = [".", "crates/agent-spec", "crates/st2-resource-protocol", "crates/st2-wire"] [package] diff --git a/crates/st2-resource-protocol/src/lib.rs b/crates/st2-resource-protocol/src/lib.rs index 6256b64a..640c0246 100644 --- a/crates/st2-resource-protocol/src/lib.rs +++ b/crates/st2-resource-protocol/src/lib.rs @@ -719,7 +719,7 @@ pub struct Publication { } impl Publication { - fn validate(&self) -> Result<(), ProtocolError> { + pub fn validate(&self) -> Result<(), ProtocolError> { validate_topics(&self.topics)?; validate_facts(self.facts.as_deref().unwrap_or_default()) } @@ -742,6 +742,25 @@ pub enum ObservationResult { }, } +impl ObservationResult { + pub fn validate(&self) -> Result<(), ProtocolError> { + match self { + Self::Unchanged => Ok(()), + Self::Failed { diagnostic } => { + if let Some(diagnostic) = diagnostic + && diagnostic.len() > MAX_OBSERVATION_DIAGNOSTIC_BYTES + { + return Err(ProtocolError::ObservationDiagnosticTooLarge { + actual: diagnostic.len(), + }); + } + Ok(()) + } + Self::Published { publication } => publication.validate(), + } + } +} + #[derive(Deserialize)] #[serde( tag = "status", @@ -1042,20 +1061,7 @@ fn validate_runtime_message(message: &RuntimeMessage) -> Result<(), ProtocolErro if *demand_watermark == 0 { return Err(ProtocolError::InvalidDemandWatermark); } - match result { - ObservationResult::Unchanged => Ok(()), - ObservationResult::Failed { diagnostic } => { - if let Some(diagnostic) = diagnostic - && diagnostic.len() > MAX_OBSERVATION_DIAGNOSTIC_BYTES - { - return Err(ProtocolError::ObservationDiagnosticTooLarge { - actual: diagnostic.len(), - }); - } - Ok(()) - } - ObservationResult::Published { publication } => publication.validate(), - } + result.validate() } } } @@ -1071,7 +1077,7 @@ fn validate_facts(facts: &[ResourceFact]) -> Result<(), ProtocolError> { .map_err(ProtocolError::InvalidFacts) } -fn validate_topics(topics: &[String]) -> Result<(), ProtocolError> { +pub fn validate_topics(topics: &[String]) -> Result<(), ProtocolError> { let mut unique = BTreeSet::new(); for topic in topics { if topic.is_empty() { @@ -1272,6 +1278,27 @@ mod tests { ); } + #[test] + fn observation_result_validation_is_reusable_outside_line_framing() { + assert!(ObservationResult::Unchanged.validate().is_ok()); + assert!(matches!( + ObservationResult::Failed { + diagnostic: Some("x".repeat(MAX_OBSERVATION_DIAGNOSTIC_BYTES + 1)), + } + .validate(), + Err(ProtocolError::ObservationDiagnosticTooLarge { .. }) + )); + let mut invalid = publication(b"duplicate topic"); + invalid.topics = vec!["same".to_owned(), "same".to_owned()]; + assert!(matches!( + ObservationResult::Published { + publication: invalid, + } + .validate(), + Err(ProtocolError::InvalidTopics(_)) + )); + } + #[test] fn fact_wire_shape_distinguishes_omission_from_explicit_null() { let mut message = publish(b"fact"); diff --git a/crates/st2-resource-wasip2/Cargo.toml b/crates/st2-resource-wasip2/Cargo.toml new file mode 100644 index 00000000..9251bdd5 --- /dev/null +++ b/crates/st2-resource-wasip2/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "st2-resource-wasip2" +version = "0.1.0" +edition = "2024" +description = "Wasmtime Component Model executor for typed st2 resource observations." +license = "MIT" +build = "build.rs" + +[dependencies] +libc = { version = "0.2", optional = true } +serde = { version = "1", features = ["derive"], optional = true } +serde_json = "1" +sha2 = { version = "0.10", optional = true } +st2-resource-protocol = { path = "../st2-resource-protocol" } +wasmtime = { version = "=48.0.1", optional = true, default-features = false, features = [ + "component-model", + "cranelift", + "runtime", + "std", +] } + +[dev-dependencies] +tempfile = "3" +wat = "1" + +[features] +default = [] +runtime = ["dep:libc", "dep:serde", "dep:sha2", "dep:wasmtime"] + +[[test]] +name = "executor" +required-features = ["runtime"] diff --git a/crates/st2-resource-wasip2/build.rs b/crates/st2-resource-wasip2/build.rs new file mode 100644 index 00000000..1464c0a6 --- /dev/null +++ b/crates/st2-resource-wasip2/build.rs @@ -0,0 +1,5 @@ +fn main() { + let target = std::env::var("TARGET").expect("Cargo always supplies TARGET to build scripts"); + println!("cargo:rustc-env=ST2_WASIP2_TARGET={target}"); + println!("cargo:rerun-if-env-changed=ST2_EXECUTOR_BUILD_IDENTITY"); +} diff --git a/crates/st2-resource-wasip2/src/bindings.rs b/crates/st2-resource-wasip2/src/bindings.rs new file mode 100644 index 00000000..4a2ed031 --- /dev/null +++ b/crates/st2-resource-wasip2/src/bindings.rs @@ -0,0 +1,4 @@ +wasmtime::component::bindgen!({ + path: "wit", + world: "provider", +}); diff --git a/crates/st2-resource-wasip2/src/cache.rs b/crates/st2-resource-wasip2/src/cache.rs new file mode 100644 index 00000000..c6bbb2ee --- /dev/null +++ b/crates/st2-resource-wasip2/src/cache.rs @@ -0,0 +1,829 @@ +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read as _, Write}; +use std::path::{Path, PathBuf}; +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, +}; + +use serde::{Deserialize, Serialize}; +use wasmtime::{Engine, component::Component}; + +use crate::{ComponentDigest, sha256_hex}; + +const MANIFEST_LIMIT_BYTES: u64 = 64 * 1024; +const ARTIFACT_LIMIT_BYTES: u64 = 512 * 1024 * 1024; +static TEMP_FILE_SEQUENCE: AtomicU64 = AtomicU64::new(0); + + +#[derive(Debug, Clone)] +pub struct PrivateArtifactCache { + root: Arc, +} + +impl PrivateArtifactCache { + pub fn open(path: impl Into) -> Result { + let path = path.into(); + create_private_directory(&path).map_err(CacheOpenError::Io)?; + let metadata = fs::symlink_metadata(&path).map_err(CacheOpenError::Io)?; + if !metadata.is_dir() || metadata.file_type().is_symlink() { + return Err(CacheOpenError::NotPrivateDirectory); + } + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt as _; + if metadata.mode() & 0o022 != 0 || metadata.uid() != effective_uid() { + return Err(CacheOpenError::NotPrivateDirectory); + } + } + Ok(Self { + root: Arc::new(path), + }) + } + + pub fn root(&self) -> &Path { + self.root.as_ref() + } +} + +#[derive(Debug)] +pub enum CacheOpenError { + Io(io::Error), + NotPrivateDirectory, +} + +impl std::fmt::Display for CacheOpenError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(error) => write!(formatter, "cannot open artifact cache: {error}"), + Self::NotPrivateDirectory => formatter.write_str( + "artifact cache must be a real effective-UID-owned directory not writable by group or other users", + ), + } + } +} + +impl std::error::Error for CacheOpenError {} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CacheRejection { + ManifestTooLarge, + ManifestUnreadable(String), + ManifestInvalid(String), + ComponentDigest, + ExecutorBuildIdentity, + WasmtimeVersion, + Target, + EngineCompatibility, + ConfigIdentity, + ArtifactLength, + ArtifactDigest, + ArtifactUnreadable(String), + UntrustedOwnership(String), + Deserialization(String), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CacheDisposition { + MemoryHit, + DiskHit, + CompiledAndStored, + CompiledWithoutCache, + RejectedAndCompiled(CacheRejection), + CompiledButNotStored(String), +} + +#[derive(Clone, Serialize)] +pub(crate) struct CacheIdentity { + pub(crate) executor_build_identity: String, + pub(crate) wasmtime_version: &'static str, + pub(crate) target: &'static str, + pub(crate) engine_compatibility: String, + pub(crate) config_identity: String, +} + +impl CacheIdentity { + fn key(&self) -> String { + let encoded = + serde_json::to_vec(self).expect("serializing cache identity fields cannot fail"); + sha256_hex(&encoded) + } +} + +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ArtifactManifest { + component_digest: String, + executor_build_identity: String, + wasmtime_version: String, + target: String, + engine_compatibility: String, + config_identity: String, + artifact_length: u64, + artifact_digest: String, +} + +pub(crate) enum CacheLookup { + Miss, + Hit(Component), + Rejected(CacheRejection), +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SecuredEntryKind { + Directory, + RegularFile, + Symlink, + Other, +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy)] +struct SecurityMetadata { + kind: SecuredEntryKind, + uid: u32, + mode: u32, +} + +#[cfg(unix)] +trait SecurityMetadataOracle { + fn metadata(&self, path: &Path) -> Result; +} + +#[cfg(unix)] +struct FilesystemMetadata; + +#[cfg(unix)] +impl SecurityMetadataOracle for FilesystemMetadata { + fn metadata(&self, path: &Path) -> Result { + use std::os::unix::fs::MetadataExt as _; + + let metadata = fs::symlink_metadata(path).map_err(|error| error.to_string())?; + let file_type = metadata.file_type(); + let kind = if file_type.is_symlink() { + SecuredEntryKind::Symlink + } else if metadata.is_dir() { + SecuredEntryKind::Directory + } else if metadata.is_file() { + SecuredEntryKind::RegularFile + } else { + SecuredEntryKind::Other + }; + Ok(SecurityMetadata { + kind, + uid: metadata.uid(), + mode: metadata.mode(), + }) + } +} + +#[cfg(unix)] +#[derive(Debug, PartialEq, Eq)] +enum OwnershipViolation { + OutsideRoot, + Metadata { path: PathBuf, error: String }, + WrongKind { + path: PathBuf, + expected: SecuredEntryKind, + actual: SecuredEntryKind, + }, + WrongOwner { + path: PathBuf, + expected: u32, + actual: u32, + }, + WritableByOther { path: PathBuf }, +} + +#[cfg(unix)] +impl std::fmt::Display for OwnershipViolation { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::OutsideRoot => formatter.write_str("cache entry is outside the cache root"), + Self::Metadata { path, error } => { + write!(formatter, "cannot inspect {}: {error}", path.display()) + } + Self::WrongKind { + path, + expected, + actual, + } => write!( + formatter, + "{} has kind {actual:?}; expected {expected:?}", + path.display(), + ), + Self::WrongOwner { + path, + expected, + actual, + } => write!( + formatter, + "{} is owned by UID {actual}; expected effective UID {expected}", + path.display(), + ), + Self::WritableByOther { path } => write!( + formatter, + "{} is writable by group or other users", + path.display(), + ), + } + } +} + +#[cfg(unix)] +fn validate_owned_cache_entry_with( + root: &Path, + entry: &Path, + effective_uid: u32, + oracle: &impl SecurityMetadataOracle, +) -> Result<(), OwnershipViolation> { + let relative = entry + .strip_prefix(root) + .map_err(|_| OwnershipViolation::OutsideRoot)?; + validate_secured_entry( + root, + SecuredEntryKind::Directory, + effective_uid, + oracle, + )?; + let mut current = root.to_path_buf(); + let mut components = relative.components().peekable(); + while let Some(component) = components.next() { + let std::path::Component::Normal(component) = component else { + return Err(OwnershipViolation::OutsideRoot); + }; + current.push(component); + let expected = if components.peek().is_some() { + SecuredEntryKind::Directory + } else { + SecuredEntryKind::RegularFile + }; + validate_secured_entry(¤t, expected, effective_uid, oracle)?; + } + Ok(()) +} + +#[cfg(unix)] +fn validate_secured_entry( + path: &Path, + expected: SecuredEntryKind, + effective_uid: u32, + oracle: &impl SecurityMetadataOracle, +) -> Result<(), OwnershipViolation> { + let metadata = oracle + .metadata(path) + .map_err(|error| OwnershipViolation::Metadata { + path: path.to_path_buf(), + error, + })?; + if metadata.kind != expected { + return Err(OwnershipViolation::WrongKind { + path: path.to_path_buf(), + expected, + actual: metadata.kind, + }); + } + if metadata.uid != effective_uid { + return Err(OwnershipViolation::WrongOwner { + path: path.to_path_buf(), + expected: effective_uid, + actual: metadata.uid, + }); + } + if metadata.mode & 0o022 != 0 { + return Err(OwnershipViolation::WritableByOther { + path: path.to_path_buf(), + }); + } + Ok(()) +} + +#[cfg(unix)] +fn effective_uid() -> u32 { + // SAFETY: geteuid has no preconditions and reads process credentials. + unsafe { libc::geteuid() } +} + +pub(crate) fn load( + cache: &PrivateArtifactCache, + engine: &Engine, + component_digest: ComponentDigest, + identity: &CacheIdentity, +) -> CacheLookup { + let manifest_path = manifest_path(cache, component_digest, identity); + let manifest_bytes = match read_regular_bounded(&manifest_path, MANIFEST_LIMIT_BYTES) { + Ok(Some(bytes)) => bytes, + Ok(None) => return CacheLookup::Miss, + Err(ReadError::TooLarge) => { + return CacheLookup::Rejected(CacheRejection::ManifestTooLarge); + } + Err(ReadError::Io(error)) => { + return CacheLookup::Rejected(CacheRejection::ManifestUnreadable(error)); + } + }; + let manifest: ArtifactManifest = match serde_json::from_slice(&manifest_bytes) { + Ok(manifest) => manifest, + Err(error) => { + return CacheLookup::Rejected(CacheRejection::ManifestInvalid(error.to_string())); + } + }; + if manifest.component_digest != component_digest.to_string() { + return CacheLookup::Rejected(CacheRejection::ComponentDigest); + } + if manifest.executor_build_identity != identity.executor_build_identity { + return CacheLookup::Rejected(CacheRejection::ExecutorBuildIdentity); + } + if manifest.wasmtime_version != identity.wasmtime_version { + return CacheLookup::Rejected(CacheRejection::WasmtimeVersion); + } + if manifest.target != identity.target { + return CacheLookup::Rejected(CacheRejection::Target); + } + if manifest.engine_compatibility != identity.engine_compatibility { + return CacheLookup::Rejected(CacheRejection::EngineCompatibility); + } + if manifest.config_identity != identity.config_identity { + return CacheLookup::Rejected(CacheRejection::ConfigIdentity); + } + if manifest.artifact_length > ARTIFACT_LIMIT_BYTES { + return CacheLookup::Rejected(CacheRejection::ArtifactLength); + } + if !is_sha256(&manifest.artifact_digest) { + return CacheLookup::Rejected(CacheRejection::ArtifactDigest); + } + let artifact_path = cache + .root() + .join("objects") + .join(format!("{}.cwasm", manifest.artifact_digest)); + let artifact = match read_regular_bounded(&artifact_path, manifest.artifact_length) { + Ok(Some(bytes)) if bytes.len() as u64 == manifest.artifact_length => bytes, + Ok(Some(_)) | Err(ReadError::TooLarge) => { + return CacheLookup::Rejected(CacheRejection::ArtifactLength); + } + Ok(None) => { + return CacheLookup::Rejected(CacheRejection::ArtifactUnreadable( + "artifact is missing".to_owned(), + )); + } + Err(ReadError::Io(error)) => { + return CacheLookup::Rejected(CacheRejection::ArtifactUnreadable(error)); + } + }; + if sha256_hex(&artifact) != manifest.artifact_digest { + return CacheLookup::Rejected(CacheRejection::ArtifactDigest); + } + #[cfg(unix)] + { + let oracle = FilesystemMetadata; + for path in [&manifest_path, &artifact_path] { + if let Err(error) = + validate_owned_cache_entry_with(cache.root(), path, effective_uid(), &oracle) + { + return CacheLookup::Rejected(CacheRejection::UntrustedOwnership( + error.to_string(), + )); + } + } + } + + // The cache root, internal ancestors, manifest and artifact are effective-UID-owned and + // non-writable by other users, and every manifest field plus the exact artifact bytes has + // been verified before crossing Wasmtime's unsafe AOT boundary. + let component = unsafe { Component::deserialize(engine, &artifact) }; + match component { + Ok(component) => CacheLookup::Hit(component), + Err(error) => CacheLookup::Rejected(CacheRejection::Deserialization(error.to_string())), + } +} + +pub(crate) fn store( + cache: &PrivateArtifactCache, + component: &Component, + component_digest: ComponentDigest, + identity: &CacheIdentity, +) -> Result<(), String> { + let artifact = component.serialize().map_err(|error| error.to_string())?; + if artifact.len() as u64 > ARTIFACT_LIMIT_BYTES { + return Err("serialized component exceeds the artifact cache limit".to_owned()); + } + let artifact_digest = sha256_hex(&artifact); + let artifact_length = artifact.len() as u64; + let object_directory = cache.root().join("objects"); + create_private_directory(&object_directory).map_err(|error| error.to_string())?; + let artifact_path = object_directory.join(format!("{artifact_digest}.cwasm")); + create_immutable(&artifact_path, &artifact)?; + + let manifest = ArtifactManifest { + component_digest: component_digest.to_string(), + executor_build_identity: identity.executor_build_identity.clone(), + wasmtime_version: identity.wasmtime_version.to_owned(), + target: identity.target.to_owned(), + engine_compatibility: identity.engine_compatibility.clone(), + config_identity: identity.config_identity.clone(), + artifact_length, + artifact_digest, + }; + let manifest_bytes = serde_json::to_vec(&manifest).map_err(|error| error.to_string())?; + let path = manifest_path(cache, component_digest, identity); + let directory = path + .parent() + .expect("manifest paths always have a parent directory"); + create_private_directory(directory).map_err(|error| error.to_string())?; + create_immutable(&path, &manifest_bytes) +} + +fn manifest_path( + cache: &PrivateArtifactCache, + component_digest: ComponentDigest, + identity: &CacheIdentity, +) -> PathBuf { + cache + .root() + .join("manifests") + .join(component_digest.to_string()) + .join(format!("{}.json", identity.key())) +} + +fn create_immutable(path: &Path, bytes: &[u8]) -> Result<(), String> { + let parent = path + .parent() + .ok_or_else(|| "cache entry path has no parent directory".to_owned())?; + let (temporary_path, mut file) = loop { + let sequence = TEMP_FILE_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let file_name = path + .file_name() + .ok_or_else(|| "cache entry path has no file name".to_owned())? + .to_string_lossy(); + let temporary_path = + parent.join(format!(".{file_name}.tmp-{}-{sequence}", std::process::id())); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(0o600); + } + match options.open(&temporary_path) { + Ok(file) => break (temporary_path, file), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue, + Err(error) => return Err(error.to_string()), + } + }; + + if let Err(error) = file.write_all(bytes).and_then(|()| file.sync_all()) { + drop(file); + let _ = fs::remove_file(&temporary_path); + return Err(error.to_string()); + } + drop(file); + + match publish_no_replace(&temporary_path, path) { + Ok(()) => sync_directory(parent).map_err(|error| error.to_string()), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + let _ = fs::remove_file(&temporary_path); + verify_immutable(path, bytes)?; + sync_directory(parent).map_err(|error| error.to_string()) + } + Err(error) => { + let _ = fs::remove_file(&temporary_path); + Err(error.to_string()) + } + } +} + +fn verify_immutable(path: &Path, bytes: &[u8]) -> Result<(), String> { + match read_regular_bounded(path, bytes.len() as u64) { + Ok(Some(existing)) if existing == bytes => Ok(()), + Ok(Some(_)) | Err(ReadError::TooLarge) => { + Err("immutable cache entry already exists with different bytes".to_owned()) + } + Ok(None) => Err("immutable cache entry disappeared during publication".to_owned()), + Err(ReadError::Io(error)) => Err(error), + } +} + +#[cfg(target_os = "linux")] +fn publish_no_replace(from: &Path, to: &Path) -> Result<(), io::Error> { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt as _; + + let c_from = CString::new(from.as_os_str().as_bytes()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "temporary path contains NUL"))?; + let c_to = CString::new(to.as_os_str().as_bytes()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "cache path contains NUL"))?; + // Both paths are in the same private cache directory. RENAME_NOREPLACE gives publication + // a single atomic winner without ever exposing the temporary file at the final name. + let result = unsafe { + libc::renameat2( + libc::AT_FDCWD, + c_from.as_ptr(), + libc::AT_FDCWD, + c_to.as_ptr(), + libc::RENAME_NOREPLACE, + ) + }; + if result == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + if matches!( + error.raw_os_error(), + Some(code) if code == libc::ENOSYS || code == libc::EINVAL + ) { + return publish_with_hard_link(from, to); + } + Err(error) +} + +#[cfg(target_os = "macos")] +fn publish_no_replace(from: &Path, to: &Path) -> Result<(), io::Error> { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt as _; + + let from = CString::new(from.as_os_str().as_bytes()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "temporary path contains NUL"))?; + let to = CString::new(to.as_os_str().as_bytes()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "cache path contains NUL"))?; + let result = unsafe { libc::renamex_np(from.as_ptr(), to.as_ptr(), libc::RENAME_EXCL) }; + if result == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } +} + +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +fn publish_no_replace(from: &Path, to: &Path) -> Result<(), io::Error> { + publish_with_hard_link(from, to) +} + +#[cfg(not(target_os = "macos"))] +fn publish_with_hard_link(from: &Path, to: &Path) -> Result<(), io::Error> { + fs::hard_link(from, to)?; + fs::remove_file(from) +} + +fn sync_directory(path: &Path) -> Result<(), io::Error> { + #[cfg(unix)] + { + File::open(path)?.sync_all() + } + #[cfg(not(unix))] + { + let _ = path; + Ok(()) + } +} + +fn create_private_directory(path: &Path) -> Result<(), io::Error> { + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt as _; + let mut builder = fs::DirBuilder::new(); + builder.recursive(true).mode(0o700); + builder.create(path) + } + #[cfg(not(unix))] + { + fs::create_dir_all(path) + } +} + +enum ReadError { + TooLarge, + Io(String), +} + +fn read_regular_bounded(path: &Path, limit: u64) -> Result>, ReadError> { + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK); + } + let file = match options.open(path) { + Ok(file) => file, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(ReadError::Io(error.to_string())), + }; + let metadata = file + .metadata() + .map_err(|error| ReadError::Io(error.to_string()))?; + if !metadata.is_file() { + return Err(ReadError::Io("entry is not a regular file".to_owned())); + } + if metadata.len() > limit { + return Err(ReadError::TooLarge); + } + let mut bytes = Vec::new(); + file.take(limit.saturating_add(1)) + .read_to_end(&mut bytes) + .map_err(|error| ReadError::Io(error.to_string()))?; + if bytes.len() as u64 > limit { + return Err(ReadError::TooLarge); + } + Ok(Some(bytes)) +} + +fn is_sha256(value: &str) -> bool { + value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + #[cfg(unix)] + use std::collections::HashMap; + use std::sync::{Arc, Barrier}; + + use super::*; + + fn identity() -> CacheIdentity { + CacheIdentity { + executor_build_identity: "executor-a".to_owned(), + wasmtime_version: "48.0.1", + target: "x86_64-unknown-linux-gnu", + engine_compatibility: "compat-a".to_owned(), + config_identity: "config-a".to_owned(), + } + } + + #[cfg(unix)] + struct FakeMetadata { + entries: HashMap, + } + + #[cfg(unix)] + impl SecurityMetadataOracle for FakeMetadata { + fn metadata(&self, path: &Path) -> Result { + self.entries + .get(path) + .copied() + .ok_or_else(|| "missing fake metadata".to_owned()) + } + } + + #[cfg(unix)] + fn owned_manifest_tree(uid: u32) -> (PathBuf, PathBuf, FakeMetadata) { + let root = PathBuf::from("/cache"); + let manifest_directory = root.join("manifests"); + let digest_directory = manifest_directory.join("digest"); + let manifest = digest_directory.join("identity.json"); + let entries = [ + ( + root.clone(), + SecurityMetadata { + kind: SecuredEntryKind::Directory, + uid, + mode: 0o700, + }, + ), + ( + manifest_directory, + SecurityMetadata { + kind: SecuredEntryKind::Directory, + uid, + mode: 0o700, + }, + ), + ( + digest_directory, + SecurityMetadata { + kind: SecuredEntryKind::Directory, + uid, + mode: 0o700, + }, + ), + ( + manifest.clone(), + SecurityMetadata { + kind: SecuredEntryKind::RegularFile, + uid, + mode: 0o600, + }, + ), + ] + .into_iter() + .collect(); + (root, manifest, FakeMetadata { entries }) + } + + #[cfg(unix)] + #[test] + fn ownership_policy_checks_root_ancestors_and_entry() { + let effective_uid = 42; + let (root, manifest, trusted) = owned_manifest_tree(effective_uid); + assert_eq!( + validate_owned_cache_entry_with(&root, &manifest, effective_uid, &trusted), + Ok(()) + ); + + for path in [ + root.clone(), + root.join("manifests"), + root.join("manifests/digest"), + manifest.clone(), + ] { + let (_, _, mut untrusted) = owned_manifest_tree(effective_uid); + untrusted.entries.get_mut(&path).unwrap().uid = 7; + assert_eq!( + validate_owned_cache_entry_with( + &root, + &manifest, + effective_uid, + &untrusted, + ), + Err(OwnershipViolation::WrongOwner { + path, + expected: effective_uid, + actual: 7, + }) + ); + } + } + + + #[test] + fn cache_key_covers_every_compatibility_field() { + let base = identity(); + let mut variants = Vec::new(); + + let mut changed = base.clone(); + changed.executor_build_identity = "executor-b".to_owned(); + variants.push(changed); + let mut changed = base.clone(); + changed.wasmtime_version = "48.0.2"; + variants.push(changed); + let mut changed = base.clone(); + changed.target = "aarch64-unknown-linux-gnu"; + variants.push(changed); + let mut changed = base.clone(); + changed.engine_compatibility = "compat-b".to_owned(); + variants.push(changed); + let mut changed = base.clone(); + changed.config_identity = "config-b".to_owned(); + variants.push(changed); + + let keys = std::iter::once(base.key()) + .chain(variants.iter().map(CacheIdentity::key)) + .collect::>(); + assert_eq!(keys.len(), variants.len() + 1); + } + + #[test] + fn concurrent_immutable_writers_publish_one_complete_entry() { + let directory = tempfile::tempdir().unwrap(); + let path = Arc::new(directory.path().join("entry")); + let bytes = Arc::new(vec![0x5a; 64 * 1024]); + let barrier = Arc::new(Barrier::new(8)); + let writers = (0..8) + .map(|_| { + let path = Arc::clone(&path); + let bytes = Arc::clone(&bytes); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + create_immutable(&path, &bytes) + }) + }) + .collect::>(); + + for writer in writers { + writer.join().unwrap().unwrap(); + } + assert_eq!(fs::read(path.as_ref()).unwrap(), *bytes); + let residue = fs::read_dir(directory.path()) + .unwrap() + .filter_map(Result::ok) + .filter(|entry| entry.file_name().to_string_lossy().contains(".tmp-")) + .count(); + assert_eq!(residue, 0); + } + + #[test] + fn truncated_crash_temp_is_never_published_as_the_final_entry() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("entry"); + let residue = directory.path().join(".entry.tmp-crash"); + fs::write(&residue, b"partial").unwrap(); + + create_immutable(&path, b"complete immutable bytes").unwrap(); + + assert_eq!(fs::read(path).unwrap(), b"complete immutable bytes"); + assert_eq!(fs::read(residue).unwrap(), b"partial"); + } + + #[test] + fn truncated_final_entry_is_rejected_without_overwrite() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("entry"); + fs::write(&path, b"partial").unwrap(); + + let error = create_immutable(&path, b"complete immutable bytes").unwrap_err(); + + assert!(error.contains("different bytes")); + assert_eq!(fs::read(path).unwrap(), b"partial"); + } +} diff --git a/crates/st2-resource-wasip2/src/lib.rs b/crates/st2-resource-wasip2/src/lib.rs new file mode 100644 index 00000000..f6857072 --- /dev/null +++ b/crates/st2-resource-wasip2/src/lib.rs @@ -0,0 +1,1146 @@ +//! Typed, capability-closed Component Model execution for resource observations. + +use serde_json::Value; +use st2_resource_protocol::SnapshotDigest; + +#[derive(Debug, Clone, PartialEq)] +pub struct ObservationRequest { + pub uri: String, + pub selector: Value, + pub prior_digest: Option, + pub demand_watermark: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum SchedulingCapability { + Demand, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ProviderDescriptor { + pub capabilities: Vec, + pub selector_schema: Value, + pub default_selector: Value, + pub topics: Vec, + pub snapshot_media_type: String, + pub snapshot_schema_id: String, +} + +#[cfg(feature = "runtime")] +mod bindings; +#[cfg(feature = "runtime")] +mod cache; +#[cfg(feature = "runtime")] +mod limits; + +#[cfg(feature = "runtime")] +pub use cache::{CacheDisposition, CacheOpenError, CacheRejection, PrivateArtifactCache}; + +#[cfg(feature = "runtime")] +mod runtime { + use std::collections::{BTreeSet, HashMap}; + use std::fmt; + use std::hash::{DefaultHasher, Hash as _, Hasher as _}; + use std::sync::atomic::{AtomicU8, Ordering}; + use std::sync::{Arc, Condvar, Mutex}; + + use sha2::{Digest as _, Sha256}; + use st2_resource_protocol::{ + FactError, FactValue, ObservationResult, ProtocolError, Publication, ResourceFact, + SnapshotBytes, SnapshotSizeError, MAX_SELECTOR_BYTES, validate_topics, + }; + use wasmtime::component::{Component, Linker}; + use wasmtime::{Config, Engine, Store, Trap, UpdateDeadline}; + + use crate::bindings::Provider; + use crate::bindings::exports::provider_api as guest; + use crate::cache::{self, CacheDisposition, CacheIdentity, CacheLookup, PrivateArtifactCache}; + use crate::limits::InvocationLimits; + use crate::{ObservationRequest, ProviderDescriptor, SchedulingCapability}; + + pub const WASMTIME_VERSION: &str = "48.0.1"; + pub const DEFAULT_MAX_COMPONENT_BYTES: usize = 16 * 1024 * 1024; + pub const DEFAULT_FUEL_PER_OBSERVATION: u64 = 10_000_000; + pub const DEFAULT_MAX_MEMORY_BYTES: usize = 64 * 1024 * 1024; + pub const DEFAULT_MAX_TABLE_ELEMENTS: usize = 10_000; + pub const DEFAULT_MAX_INSTANCES: usize = 128; + pub const DEFAULT_MAX_TABLES: usize = 64; + pub const DEFAULT_MAX_MEMORIES: usize = 64; + const MAX_DESCRIPTOR_JSON_BYTES: usize = 64 * 1024; + + #[derive(Debug, Clone, PartialEq, Eq)] + pub struct RuntimeConfig { + pub max_component_bytes: usize, + pub fuel_per_observation: u64, + pub max_memory_bytes: usize, + pub max_table_elements: usize, + pub max_instances: usize, + pub max_tables: usize, + pub max_memories: usize, + } + + impl Default for RuntimeConfig { + fn default() -> Self { + Self { + max_component_bytes: DEFAULT_MAX_COMPONENT_BYTES, + fuel_per_observation: DEFAULT_FUEL_PER_OBSERVATION, + max_memory_bytes: DEFAULT_MAX_MEMORY_BYTES, + max_table_elements: DEFAULT_MAX_TABLE_ELEMENTS, + max_instances: DEFAULT_MAX_INSTANCES, + max_tables: DEFAULT_MAX_TABLES, + max_memories: DEFAULT_MAX_MEMORIES, + } + } + } + + impl RuntimeConfig { + fn validate(&self) -> Result<(), BuildError> { + let nonzero = self.max_component_bytes > 0 + && self.fuel_per_observation > 0 + && self.max_memory_bytes > 0 + && self.max_table_elements > 0 + && self.max_instances > 0 + && self.max_tables > 0 + && self.max_memories > 0; + if nonzero { + Ok(()) + } else { + Err(BuildError::InvalidConfig("all limits must be nonzero")) + } + } + + fn identity(&self) -> String { + sha256_hex( + format!( + "component-model=1;fuel=1;epoch=1;component-bytes={};fuel-per-call={};memory={};table-elements={};instances={};tables={};memories={}", + self.max_component_bytes, + self.fuel_per_observation, + self.max_memory_bytes, + self.max_table_elements, + self.max_instances, + self.max_tables, + self.max_memories, + ) + .as_bytes(), + ) + } + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub enum InterruptionReason { + Cancelled, + TimedOut, + } + + struct InvocationControlState { + reason: AtomicU8, + changed: Mutex<()>, + wake: Condvar, + } + + #[derive(Clone)] + pub struct InvocationControl { + state: Arc, + } + + impl InvocationControl { + fn new() -> Self { + Self { + state: Arc::new(InvocationControlState { + reason: AtomicU8::new(INTERRUPTION_NONE), + changed: Mutex::new(()), + wake: Condvar::new(), + }), + } + } + + pub fn interruption_reason(&self) -> Option { + match self.state.reason.load(Ordering::Acquire) { + INTERRUPTION_CANCELLED => Some(InterruptionReason::Cancelled), + INTERRUPTION_TIMED_OUT => Some(InterruptionReason::TimedOut), + _ => None, + } + } + + pub fn wait_for_interruption(&self) -> InterruptionReason { + let mut guard = self + .state + .changed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + loop { + if let Some(reason) = self.interruption_reason() { + return reason; + } + guard = self + .state + .wake + .wait(guard) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + + fn interrupt(&self, reason: u8) -> bool { + let guard = self + .state + .changed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let changed = self + .state + .reason + .compare_exchange( + INTERRUPTION_NONE, + reason, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok(); + if changed { + self.state.wake.notify_all(); + } + drop(guard); + changed + } + } + + #[derive(Debug, Clone, Copy)] + pub enum CapabilityPhase<'a> { + Describe, + Observe(&'a ObservationRequest), + } + + #[derive(Clone)] + pub struct CapabilityContext<'a> { + phase: CapabilityPhase<'a>, + control: InvocationControl, + } + + impl<'a> CapabilityContext<'a> { + pub fn phase(&self) -> CapabilityPhase<'a> { + self.phase + } + + pub fn control(&self) -> &InvocationControl { + &self.control + } + } + + pub trait CapabilityModule: Send + Sync + 'static { + type Invocation: Send + 'static; + + fn import_names(&self) -> &'static [&'static str]; + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error>; + + fn begin(&self, context: CapabilityContext<'_>) -> Self::Invocation; + } + + #[derive(Debug, Clone, Copy, Default)] + pub struct NoCapabilities; + + impl CapabilityModule for NoCapabilities { + type Invocation = (); + + fn import_names(&self) -> &'static [&'static str] { + &[] + } + + fn add_to_linker( + &self, + _linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + Ok(()) + } + + fn begin(&self, _context: CapabilityContext<'_>) -> Self::Invocation {} + } + + pub struct InvocationStore { + capability: T, + control: InvocationControl, + limits: InvocationLimits, + } + + impl InvocationStore { + pub fn capability(&self) -> &T { + &self.capability + } + + pub fn capability_mut(&mut self) -> &mut T { + &mut self.capability + } + + pub fn control(&self) -> &InvocationControl { + &self.control + } + } + + pub struct Executor { + engine: Engine, + linker: Arc>>, + capabilities: Arc, + allowed_imports: Arc>, + components: Mutex>>, + config: RuntimeConfig, + cache: Option, + cache_identity: CacheIdentity, + runtime_token: Arc<()>, + } + + impl Executor { + pub fn closed( + config: RuntimeConfig, + cache: Option, + ) -> Result { + Self::new(config, cache, NoCapabilities) + } + } + + impl Executor { + pub fn new( + config: RuntimeConfig, + cache: Option, + capabilities: C, + ) -> Result { + config.validate()?; + let mut wasmtime_config = Config::new(); + wasmtime_config + .wasm_component_model(true) + .consume_fuel(true) + .epoch_interruption(true); + let engine = Engine::new(&wasmtime_config) + .map_err(|error| BuildError::Engine(error.to_string()))?; + let mut linker = Linker::new(&engine); + capabilities + .add_to_linker(&mut linker) + .map_err(|error| BuildError::Linker(error.to_string()))?; + let allowed_imports = capabilities.import_names().iter().copied().collect(); + let mut compatibility_hasher = DefaultHasher::new(); + engine + .precompile_compatibility_hash() + .hash(&mut compatibility_hasher); + let cache_identity = CacheIdentity { + executor_build_identity: executor_build_identity().to_owned(), + wasmtime_version: WASMTIME_VERSION, + target: env!("ST2_WASIP2_TARGET"), + engine_compatibility: format!("{:016x}", compatibility_hasher.finish()), + config_identity: config.identity(), + }; + Ok(Self { + engine, + linker: Arc::new(linker), + capabilities: Arc::new(capabilities), + allowed_imports: Arc::new(allowed_imports), + components: Mutex::new(HashMap::new()), + config, + cache, + cache_identity, + runtime_token: Arc::new(()), + }) + } + + pub fn load(&self, bytes: &[u8]) -> Result { + if bytes.len() > self.config.max_component_bytes { + return Err(LoadError::ComponentTooLarge { + actual: bytes.len(), + maximum: self.config.max_component_bytes, + }); + } + let digest = ComponentDigest::of(bytes); + // Compilation is intentionally serialized per executor: racing the same digest must + // not manufacture a second compiled identity outside the one immutable cache. + let mut components = self + .components + .lock() + .map_err(|_| LoadError::Internal("compiled component cache lock poisoned"))?; + if let Some(component) = components.get(&digest).cloned() { + return Ok(self.loaded(digest, component, CacheDisposition::MemoryHit)); + } + + let lookup = self.cache.as_ref().map_or(CacheLookup::Miss, |cache| { + cache::load(cache, &self.engine, digest, &self.cache_identity) + }); + let (component, disposition) = match lookup { + CacheLookup::Hit(component) => (component, CacheDisposition::DiskHit), + CacheLookup::Miss => { + let component = self.compile(bytes)?; + let disposition = match &self.cache { + Some(cache) => match cache::store( + cache, + &component, + digest, + &self.cache_identity, + ) { + Ok(()) => CacheDisposition::CompiledAndStored, + Err(error) => CacheDisposition::CompiledButNotStored(error), + }, + None => CacheDisposition::CompiledWithoutCache, + }; + (component, disposition) + } + CacheLookup::Rejected(rejection) => { + let component = self.compile(bytes)?; + (component, CacheDisposition::RejectedAndCompiled(rejection)) + } + }; + self.admit_imports(&component)?; + let component = Arc::new(component); + components.insert(digest, Arc::clone(&component)); + Ok(self.loaded(digest, component, disposition)) + } + + pub fn interruption_handle(&self) -> InterruptionHandle { + InterruptionHandle { + engine: self.engine.clone(), + control: InvocationControl::new(), + runtime_token: Arc::clone(&self.runtime_token), + } + } + + pub fn describe( + &self, + component: &LoadedComponent, + interruption: Option<&InterruptionHandle>, + ) -> Result { + if !Arc::ptr_eq(&self.runtime_token, &component.runtime_token) { + return Err(DescribeError::WrongExecutor); + } + if let Some(interruption) = interruption + && !Arc::ptr_eq(&self.runtime_token, &interruption.runtime_token) + { + return Err(DescribeError::WrongExecutor); + } + let control = interruption.map_or_else(InvocationControl::new, |handle| { + handle.control.clone() + }); + if let Some(reason) = control.interruption_reason() { + return Err(DescribeError::from(reason)); + } + let mut store = self + .begin_invocation(CapabilityPhase::Describe, &control) + .map_err(|error| DescribeError::Instantiation(error.to_string()))?; + let bindings = Provider::instantiate( + &mut store, + component.component.as_ref(), + self.linker.as_ref(), + ) + .map_err(|error| { + classify_execution_error(&store, &control.state.reason, error, true).describe() + })?; + let result = bindings + .provider_api() + .call_describe(&mut store) + .map_err(|error| { + classify_execution_error(&store, &control.state.reason, error, false).describe() + })?; + if let Some(reason) = control.interruption_reason() { + return Err(DescribeError::from(reason)); + } + let descriptor = result.map_err(GuestDescriptorError::from)?; + map_descriptor(descriptor).map_err(DescribeError::InvalidDescriptor) + } + + pub fn observe( + &self, + component: &LoadedComponent, + request: &ObservationRequest, + interruption: Option<&InterruptionHandle>, + ) -> Result { + if !Arc::ptr_eq(&self.runtime_token, &component.runtime_token) { + return Err(ObserveError::WrongExecutor); + } + if let Some(interruption) = interruption + && !Arc::ptr_eq(&self.runtime_token, &interruption.runtime_token) + { + return Err(ObserveError::WrongExecutor); + } + if request.uri.len() > 64 * 1024 { + return Err(ObserveError::InvalidRequest("URI exceeds 64 KiB")); + } + if matches!(request.demand_watermark, Some(0)) { + return Err(ObserveError::InvalidRequest( + "demand watermark must be positive", + )); + } + let selector_json = serde_json::to_string(&request.selector) + .map_err(|_| ObserveError::InvalidRequest("selector is not JSON"))?; + if selector_json.len() > MAX_SELECTOR_BYTES { + return Err(ObserveError::InvalidRequest( + "selector exceeds the protocol limit", + )); + } + let guest_request = guest::ObserveRequest { + uri: request.uri.clone(), + selector_json, + prior_digest: request + .prior_digest + .map(|digest| digest.as_bytes().to_vec()), + demand_watermark: request.demand_watermark, + }; + let control = interruption.map_or_else(InvocationControl::new, |handle| { + handle.control.clone() + }); + if let Some(reason) = control.interruption_reason() { + return Err(ObserveError::from(reason)); + } + let mut store = self + .begin_invocation(CapabilityPhase::Observe(request), &control) + .map_err(|error| ObserveError::Instantiation(error.to_string()))?; + let bindings = Provider::instantiate( + &mut store, + component.component.as_ref(), + self.linker.as_ref(), + ) + .map_err(|error| { + classify_execution_error(&store, &control.state.reason, error, true).observe() + })?; + let result = bindings + .provider_api() + .call_observe(&mut store, &guest_request) + .map_err(|error| { + classify_execution_error(&store, &control.state.reason, error, false).observe() + })?; + if let Some(reason) = control.interruption_reason() { + return Err(ObserveError::from(reason)); + } + map_proposal(result).map_err(ObserveError::InvalidProposal) + } + + fn begin_invocation( + &self, + phase: CapabilityPhase<'_>, + control: &InvocationControl, + ) -> Result>, wasmtime::Error> { + let state = InvocationStore { + capability: self.capabilities.begin(CapabilityContext { + phase, + control: control.clone(), + }), + control: control.clone(), + limits: InvocationLimits::new(&self.config), + }; + let mut store = Store::new(&self.engine, state); + store.limiter(|state| &mut state.limits); + store.set_fuel(self.config.fuel_per_observation)?; + arm_epoch_deadline(&mut store, control); + Ok(store) + } + + fn compile(&self, bytes: &[u8]) -> Result { + let component = Component::new(&self.engine, bytes) + .map_err(|error| LoadError::Compilation(format!("{error:#}")))?; + self.admit_imports(&component)?; + Ok(component) + } + + fn admit_imports(&self, component: &Component) -> Result<(), LoadError> { + let forbidden: Vec = component + .component_type() + .imports(&self.engine) + .map(|(name, _)| name) + .filter(|name| !self.allowed_imports.contains(name)) + .map(str::to_owned) + .collect(); + if forbidden.is_empty() { + Ok(()) + } else { + Err(LoadError::ForbiddenImports(forbidden)) + } + } + + fn loaded( + &self, + digest: ComponentDigest, + component: Arc, + cache_disposition: CacheDisposition, + ) -> LoadedComponent { + LoadedComponent { + digest, + component, + cache_disposition, + runtime_token: Arc::clone(&self.runtime_token), + } + } + } + + pub struct LoadedComponent { + digest: ComponentDigest, + component: Arc, + cache_disposition: CacheDisposition, + runtime_token: Arc<()>, + } + + impl LoadedComponent { + pub fn digest(&self) -> ComponentDigest { + self.digest + } + + pub fn cache_disposition(&self) -> &CacheDisposition { + &self.cache_disposition + } + } + + const INTERRUPTION_NONE: u8 = 0; + const INTERRUPTION_CANCELLED: u8 = 1; + const INTERRUPTION_TIMED_OUT: u8 = 2; + + #[derive(Clone)] + pub struct InterruptionHandle { + engine: Engine, + control: InvocationControl, + runtime_token: Arc<()>, + } + + impl InterruptionHandle { + pub fn cancel(&self) -> bool { + self.interrupt(INTERRUPTION_CANCELLED) + } + + pub fn time_out(&self) -> bool { + self.interrupt(INTERRUPTION_TIMED_OUT) + } + + fn interrupt(&self, reason: u8) -> bool { + let changed = self.control.interrupt(reason); + if changed { + self.engine.increment_epoch(); + } + changed + } + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub enum LimitKind { + Memory, + Table, + } + + #[derive(Debug)] + pub enum BuildError { + InvalidConfig(&'static str), + Engine(String), + Linker(String), + } + + impl fmt::Display for BuildError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidConfig(message) => write!(formatter, "invalid executor config: {message}"), + Self::Engine(message) => write!(formatter, "cannot create Wasmtime engine: {message}"), + Self::Linker(message) => write!(formatter, "cannot create capability linker: {message}"), + } + } + } + + impl std::error::Error for BuildError {} + + #[derive(Debug)] + pub enum LoadError { + ComponentTooLarge { actual: usize, maximum: usize }, + Compilation(String), + ForbiddenImports(Vec), + Internal(&'static str), + } + + impl fmt::Display for LoadError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ComponentTooLarge { actual, maximum } => write!( + formatter, + "component is {actual} bytes; maximum is {maximum}", + ), + Self::Compilation(message) => write!(formatter, "component compilation failed: {message}"), + Self::ForbiddenImports(imports) => { + write!(formatter, "component imports are not admitted: {}", imports.join(", ")) + } + Self::Internal(message) => formatter.write_str(message), + } + } + } + + impl std::error::Error for LoadError {} + + #[derive(Debug, Clone, PartialEq, Eq)] + pub enum GuestDescriptorError { + InvalidDescriptor(String), + Unavailable(String), + } + + impl From for GuestDescriptorError { + fn from(error: guest::DescriptorError) -> Self { + match error { + guest::DescriptorError::InvalidDescriptor(message) => { + Self::InvalidDescriptor(message) + } + guest::DescriptorError::Unavailable(message) => Self::Unavailable(message), + } + } + } + + impl fmt::Display for GuestDescriptorError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidDescriptor(message) => { + write!(formatter, "guest descriptor is invalid: {message}") + } + Self::Unavailable(message) => { + write!(formatter, "guest descriptor is unavailable: {message}") + } + } + } + } + + impl std::error::Error for GuestDescriptorError {} + + #[derive(Debug)] + pub enum DescriptorValidationError { + DuplicateCapability, + SelectorSchemaTooLarge { actual: usize }, + SelectorSchemaJson(serde_json::Error), + DefaultSelectorTooLarge { actual: usize }, + DefaultSelectorJson(serde_json::Error), + InvalidTopics(ProtocolError), + EmptySnapshotMediaType, + EmptySnapshotSchemaId, + } + + impl fmt::Display for DescriptorValidationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::DuplicateCapability => { + formatter.write_str("descriptor capabilities must be unique") + } + Self::SelectorSchemaTooLarge { actual } => write!( + formatter, + "selector schema is {actual} bytes; maximum is {MAX_DESCRIPTOR_JSON_BYTES}", + ), + Self::SelectorSchemaJson(error) => { + write!(formatter, "selector schema is not valid JSON: {error}") + } + Self::DefaultSelectorTooLarge { actual } => write!( + formatter, + "default selector is {actual} bytes; maximum is {MAX_SELECTOR_BYTES}", + ), + Self::DefaultSelectorJson(error) => { + write!(formatter, "default selector is not valid JSON: {error}") + } + Self::InvalidTopics(error) => error.fmt(formatter), + Self::EmptySnapshotMediaType => { + formatter.write_str("snapshot media type must not be empty") + } + Self::EmptySnapshotSchemaId => { + formatter.write_str("snapshot schema ID must not be empty") + } + } + } + } + + impl std::error::Error for DescriptorValidationError {} + + #[derive(Debug)] + pub enum DescribeError { + WrongExecutor, + Instantiation(String), + Invocation(String), + Trap(Trap), + FuelExhausted, + Cancelled, + TimedOut, + ResourceLimit(LimitKind), + Guest(GuestDescriptorError), + InvalidDescriptor(DescriptorValidationError), + } + + impl From for DescribeError { + fn from(error: GuestDescriptorError) -> Self { + Self::Guest(error) + } + } + + impl From for DescribeError { + fn from(reason: InterruptionReason) -> Self { + match reason { + InterruptionReason::Cancelled => Self::Cancelled, + InterruptionReason::TimedOut => Self::TimedOut, + } + } + } + + impl fmt::Display for DescribeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::WrongExecutor => formatter + .write_str("component or interruption handle belongs to another executor"), + Self::Instantiation(message) => { + write!(formatter, "component instantiation failed: {message}") + } + Self::Invocation(message) => write!(formatter, "typed describe call failed: {message}"), + Self::Trap(trap) => write!(formatter, "guest trapped: {trap}"), + Self::FuelExhausted => formatter.write_str("guest exhausted its fuel allowance"), + Self::Cancelled => formatter.write_str("descriptor call was cancelled"), + Self::TimedOut => formatter.write_str("descriptor call timed out"), + Self::ResourceLimit(kind) => write!(formatter, "guest exceeded its {kind:?} limit"), + Self::Guest(error) => error.fmt(formatter), + Self::InvalidDescriptor(error) => { + write!(formatter, "guest descriptor is invalid: {error}") + } + } + } + } + + impl std::error::Error for DescribeError {} + + #[derive(Debug)] + pub enum ProposalError { + Snapshot(SnapshotSizeError), + Fact(FactError), + Protocol(ProtocolError), + } + + impl fmt::Display for ProposalError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Snapshot(error) => error.fmt(formatter), + Self::Fact(error) => error.fmt(formatter), + Self::Protocol(error) => error.fmt(formatter), + } + } + } + + impl std::error::Error for ProposalError {} + + #[derive(Debug)] + pub enum ObserveError { + WrongExecutor, + InvalidRequest(&'static str), + Instantiation(String), + Invocation(String), + Trap(Trap), + FuelExhausted, + Cancelled, + TimedOut, + ResourceLimit(LimitKind), + InvalidProposal(ProposalError), + } + + impl From for ObserveError { + fn from(reason: InterruptionReason) -> Self { + match reason { + InterruptionReason::Cancelled => Self::Cancelled, + InterruptionReason::TimedOut => Self::TimedOut, + } + } + } + + impl fmt::Display for ObserveError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::WrongExecutor => formatter.write_str("component or interruption handle belongs to another executor"), + Self::InvalidRequest(message) => write!(formatter, "invalid observation request: {message}"), + Self::Instantiation(message) => write!(formatter, "component instantiation failed: {message}"), + Self::Invocation(message) => write!(formatter, "typed observation call failed: {message}"), + Self::Trap(trap) => write!(formatter, "guest trapped: {trap}"), + Self::FuelExhausted => formatter.write_str("guest exhausted its fuel allowance"), + Self::Cancelled => formatter.write_str("observation was cancelled"), + Self::TimedOut => formatter.write_str("observation timed out"), + Self::ResourceLimit(kind) => write!(formatter, "guest exceeded its {kind:?} limit"), + Self::InvalidProposal(error) => write!(formatter, "guest proposal is invalid: {error}"), + } + } + } + + impl std::error::Error for ObserveError {} + + #[derive(Clone, Copy, PartialEq, Eq, Hash)] + pub struct ComponentDigest([u8; 32]); + + impl ComponentDigest { + pub fn of(bytes: &[u8]) -> Self { + Self(Sha256::digest(bytes).into()) + } + + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } + } + + impl fmt::Debug for ComponentDigest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, formatter) + } + } + + impl fmt::Display for ComponentDigest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + for byte in self.0 { + write!(formatter, "{byte:02x}")?; + } + Ok(()) + } + } + + fn arm_epoch_deadline(store: &mut Store, control: &InvocationControl) { + arm_epoch_deadline_with_hook(store, control, || {}); + } + + fn arm_epoch_deadline_with_hook( + store: &mut Store, + control: &InvocationControl, + before_deadline: impl FnOnce(), + ) { + let callback_state = Arc::clone(&control.state); + store.epoch_deadline_callback(move |_| { + if callback_state.reason.load(Ordering::Acquire) == INTERRUPTION_NONE { + Ok(UpdateDeadline::Continue(1)) + } else { + Ok(UpdateDeadline::Interrupt) + } + }); + before_deadline(); + store.set_epoch_deadline(1); + // Cancellation may have incremented the engine epoch just before the deadline was armed. + // Rechecking closes that lost-tick window without requiring a second external tick. + if control.interruption_reason().is_some() { + store.set_epoch_deadline(0); + } + } + + enum InvocationFailure { + Instantiation(String), + Invocation(String), + Trap(Trap), + FuelExhausted, + Cancelled, + TimedOut, + ResourceLimit(LimitKind), + } + + impl InvocationFailure { + fn observe(self) -> ObserveError { + match self { + Self::Instantiation(message) => ObserveError::Instantiation(message), + Self::Invocation(message) => ObserveError::Invocation(message), + Self::Trap(trap) => ObserveError::Trap(trap), + Self::FuelExhausted => ObserveError::FuelExhausted, + Self::Cancelled => ObserveError::Cancelled, + Self::TimedOut => ObserveError::TimedOut, + Self::ResourceLimit(kind) => ObserveError::ResourceLimit(kind), + } + } + + fn describe(self) -> DescribeError { + match self { + Self::Instantiation(message) => DescribeError::Instantiation(message), + Self::Invocation(message) => DescribeError::Invocation(message), + Self::Trap(trap) => DescribeError::Trap(trap), + Self::FuelExhausted => DescribeError::FuelExhausted, + Self::Cancelled => DescribeError::Cancelled, + Self::TimedOut => DescribeError::TimedOut, + Self::ResourceLimit(kind) => DescribeError::ResourceLimit(kind), + } + } + } + + fn classify_execution_error( + store: &Store>, + reason: &AtomicU8, + error: wasmtime::Error, + instantiating: bool, + ) -> InvocationFailure { + if let Some(kind) = store.data().limits.exceeded() { + return InvocationFailure::ResourceLimit(kind); + } + match reason.load(Ordering::Acquire) { + INTERRUPTION_CANCELLED => return InvocationFailure::Cancelled, + INTERRUPTION_TIMED_OUT => return InvocationFailure::TimedOut, + _ => {} + } + if let Some(trap) = error.downcast_ref::() { + return match trap { + Trap::OutOfFuel => InvocationFailure::FuelExhausted, + trap => InvocationFailure::Trap(*trap), + }; + } + if instantiating { + InvocationFailure::Instantiation(error.to_string()) + } else { + InvocationFailure::Invocation(error.to_string()) + } + } + + fn map_descriptor( + descriptor: guest::ProviderDescriptor, + ) -> Result { + let mut capabilities = Vec::with_capacity(descriptor.capabilities.len()); + let mut unique_capabilities = BTreeSet::new(); + for capability in descriptor.capabilities { + let capability = match capability { + guest::SchedulingCapability::Demand => SchedulingCapability::Demand, + }; + if !unique_capabilities.insert(capability) { + return Err(DescriptorValidationError::DuplicateCapability); + } + capabilities.push(capability); + } + if descriptor.selector_schema_json.len() > MAX_DESCRIPTOR_JSON_BYTES { + return Err(DescriptorValidationError::SelectorSchemaTooLarge { + actual: descriptor.selector_schema_json.len(), + }); + } + let selector_schema = serde_json::from_str(&descriptor.selector_schema_json) + .map_err(DescriptorValidationError::SelectorSchemaJson)?; + let default_selector = serde_json::from_str(&descriptor.default_selector_json) + .map_err(DescriptorValidationError::DefaultSelectorJson)?; + let default_selector_bytes = serde_json::to_vec(&default_selector) + .expect("serializing a decoded JSON value cannot fail") + .len(); + if default_selector_bytes > MAX_SELECTOR_BYTES { + return Err(DescriptorValidationError::DefaultSelectorTooLarge { + actual: default_selector_bytes, + }); + } + validate_topics(&descriptor.topics).map_err(DescriptorValidationError::InvalidTopics)?; + if descriptor.snapshot_media_type.is_empty() { + return Err(DescriptorValidationError::EmptySnapshotMediaType); + } + if descriptor.snapshot_schema_id.is_empty() { + return Err(DescriptorValidationError::EmptySnapshotSchemaId); + } + Ok(ProviderDescriptor { + capabilities, + selector_schema, + default_selector, + topics: descriptor.topics, + snapshot_media_type: descriptor.snapshot_media_type, + snapshot_schema_id: descriptor.snapshot_schema_id, + }) + } + + fn map_proposal( + proposal: guest::ObservationResult, + ) -> Result { + let proposal = match proposal { + guest::ObservationResult::Unchanged => ObservationResult::Unchanged, + guest::ObservationResult::Failed(diagnostic) => { + ObservationResult::Failed { diagnostic } + } + guest::ObservationResult::Published(publication) => { + let facts = publication + .facts + .map(|facts| { + facts + .into_iter() + .map(map_fact) + .collect::, _>>() + }) + .transpose()?; + ObservationResult::Published { + publication: Publication { + schema_id: publication.schema_id, + media_type: publication.media_type, + bytes: SnapshotBytes::new(publication.bytes) + .map_err(ProposalError::Snapshot)?, + topics: publication.topics, + facts, + }, + } + } + }; + proposal.validate().map_err(ProposalError::Protocol)?; + Ok(proposal) + } + + fn map_fact(fact: guest::Fact) -> Result { + ResourceFact::new( + fact.key, + map_fact_value(fact.before), + map_fact_value(fact.after), + ) + .map_err(ProposalError::Fact) + } + + fn map_fact_value(value: guest::FactValue) -> FactValue { + match value { + guest::FactValue::Omitted => FactValue::Omitted, + guest::FactValue::Null => FactValue::Null, + guest::FactValue::Value(value) => FactValue::Value(value), + } + } + + pub(crate) fn sha256_hex(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let mut result = String::with_capacity(64); + for byte in digest { + use std::fmt::Write as _; + write!(result, "{byte:02x}").expect("writing into a String cannot fail"); + } + result + } + + fn executor_build_identity() -> &'static str { + option_env!("ST2_EXECUTOR_BUILD_IDENTITY") + .unwrap_or(concat!(env!("CARGO_PKG_NAME"), "@", env!("CARGO_PKG_VERSION"))) + } + #[cfg(test)] + mod tests { + use std::sync::{Arc, Barrier}; + + use super::*; + + #[test] + fn cancellation_before_deadline_arm_forces_the_current_epoch() { + let mut config = Config::new(); + config.epoch_interruption(true); + let engine = Engine::new(&config).unwrap(); + let control = InvocationControl::new(); + let handle = InterruptionHandle { + engine: engine.clone(), + control: control.clone(), + runtime_token: Arc::new(()), + }; + let barrier = Arc::new(Barrier::new(2)); + let cancelling_barrier = Arc::clone(&barrier); + let cancelling = std::thread::spawn(move || { + cancelling_barrier.wait(); + assert!(handle.cancel()); + cancelling_barrier.wait(); + }); + let mut store = Store::new(&engine, ()); + arm_epoch_deadline_with_hook(&mut store, &control, || { + barrier.wait(); + barrier.wait(); + }); + cancelling.join().unwrap(); + + let module = wasmtime::Module::new( + &engine, + wat::parse_str( + "(module (func (export \"run\") (loop $spin (br $spin))))", + ) + .unwrap(), + ) + .unwrap(); + let instance = wasmtime::Instance::new(&mut store, &module, &[]).unwrap(); + let run = instance.get_typed_func::<(), ()>(&mut store, "run").unwrap(); + let error = run.call(&mut store, ()).unwrap_err(); + assert_eq!(error.downcast_ref::(), Some(&Trap::Interrupt)); + } + } +} + +#[cfg(feature = "runtime")] +pub use runtime::{ + BuildError, CapabilityContext, CapabilityModule, CapabilityPhase, ComponentDigest, + DescribeError, DescriptorValidationError, Executor, GuestDescriptorError, InterruptionHandle, + InterruptionReason, InvocationControl, InvocationStore, LimitKind, LoadError, LoadedComponent, + NoCapabilities, ObserveError, ProposalError, RuntimeConfig, DEFAULT_FUEL_PER_OBSERVATION, + DEFAULT_MAX_COMPONENT_BYTES, DEFAULT_MAX_INSTANCES, DEFAULT_MAX_MEMORIES, + DEFAULT_MAX_MEMORY_BYTES, DEFAULT_MAX_TABLE_ELEMENTS, DEFAULT_MAX_TABLES, WASMTIME_VERSION, +}; + +#[cfg(feature = "runtime")] +pub(crate) use runtime::sha256_hex; diff --git a/crates/st2-resource-wasip2/src/limits.rs b/crates/st2-resource-wasip2/src/limits.rs new file mode 100644 index 00000000..26ab2fc8 --- /dev/null +++ b/crates/st2-resource-wasip2/src/limits.rs @@ -0,0 +1,69 @@ +use wasmtime::{Error, ResourceLimiter}; + +use crate::{LimitKind, RuntimeConfig}; + +pub(crate) struct InvocationLimits { + memory_bytes: usize, + table_elements: usize, + instances: usize, + tables: usize, + memories: usize, + exceeded: Option, +} + +impl InvocationLimits { + pub(crate) fn new(config: &RuntimeConfig) -> Self { + Self { + memory_bytes: config.max_memory_bytes, + table_elements: config.max_table_elements, + instances: config.max_instances, + tables: config.max_tables, + memories: config.max_memories, + exceeded: None, + } + } + + pub(crate) fn exceeded(&self) -> Option { + self.exceeded + } +} + +impl ResourceLimiter for InvocationLimits { + fn memory_growing( + &mut self, + _current: usize, + desired: usize, + maximum: Option, + ) -> Result { + if desired > self.memory_bytes || maximum.is_some_and(|maximum| desired > maximum) { + self.exceeded = Some(LimitKind::Memory); + return Err(Error::msg("guest linear-memory limit exceeded")); + } + Ok(true) + } + + fn table_growing( + &mut self, + _current: usize, + desired: usize, + maximum: Option, + ) -> Result { + if desired > self.table_elements || maximum.is_some_and(|maximum| desired > maximum) { + self.exceeded = Some(LimitKind::Table); + return Err(Error::msg("guest table-element limit exceeded")); + } + Ok(true) + } + + fn instances(&self) -> usize { + self.instances + } + + fn tables(&self) -> usize { + self.tables + } + + fn memories(&self) -> usize { + self.memories + } +} diff --git a/crates/st2-resource-wasip2/tests/executor.rs b/crates/st2-resource-wasip2/tests/executor.rs new file mode 100644 index 00000000..33851a59 --- /dev/null +++ b/crates/st2-resource-wasip2/tests/executor.rs @@ -0,0 +1,472 @@ +use std::fs; +use std::sync::{Arc, Barrier}; + +use serde_json::json; +use st2_resource_protocol::ObservationResult; +use st2_resource_wasip2::{ + CacheDisposition, CacheRejection, CapabilityContext, CapabilityModule, Executor, + InvocationStore, LimitKind, LoadError, NoCapabilities, ObservationRequest, ObserveError, + PrivateArtifactCache, RuntimeConfig, SchedulingCapability, +}; +use wasmtime::component::{HasSelf, Linker}; + +mod fixture { + wasmtime::component::bindgen!({ + path: "wit", + world: "fixture-host", + }); +} + +use fixture::st2::resource_provider::provider_api::{ + DescriptorError, Fact, FactValue, Host, ObservationResult as WitObservationResult, + ObserveRequest, ProviderDescriptor as WitProviderDescriptor, Publication, + SchedulingCapability as WitSchedulingCapability, +}; + +const PROVIDER_IMPORT: &str = "st2:resource-provider/provider-api@0.1.0"; +const NO_EFFECT: &str = include_str!("fixtures/no-effect-component.wat"); + +#[derive(Default)] +struct FixtureInvocation { + calls: u8, +} + +fn wit_descriptor() -> WitProviderDescriptor { + WitProviderDescriptor { + capabilities: vec![WitSchedulingCapability::Demand], + selector_schema_json: r#"{"type":"object"}"#.to_owned(), + default_selector_json: "{}".to_owned(), + topics: vec!["fixture".to_owned()], + snapshot_media_type: "application/octet-stream".to_owned(), + snapshot_schema_id: "dev.compounding.fixture/v1".to_owned(), + } +} + +impl Host for InvocationStore { + fn describe(&mut self) -> Result { + Ok(wit_descriptor()) + } + + fn observe(&mut self, request: ObserveRequest) -> WitObservationResult { + self.capability_mut().calls += 1; + let calls = self.capability().calls; + let publication_byte = request + .demand_watermark + .map_or(calls, |watermark| u8::try_from(watermark).unwrap()); + let topics = if request.uri == "fixture://invalid" { + vec!["duplicate".to_owned(), "duplicate".to_owned()] + } else { + vec!["fixture".to_owned()] + }; + WitObservationResult::Published(Publication { + schema_id: "dev.compounding.fixture/v1".to_owned(), + media_type: "application/octet-stream".to_owned(), + bytes: vec![publication_byte], + topics, + facts: Some(vec![Fact { + key: "calls".to_owned(), + before: FactValue::Omitted, + after: FactValue::Value(calls.to_string()), + }]), + }) + } +} + +struct FixtureCapabilities; + +impl CapabilityModule for FixtureCapabilities { + type Invocation = FixtureInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[PROVIDER_IMPORT] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + fixture::FixtureHost::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, _context: CapabilityContext<'_>) -> Self::Invocation { + FixtureInvocation::default() + } +} +struct BlockingCapabilities { + entered: Arc, +} + +struct BlockingInvocation { + entered: Arc, +} + +impl Host for InvocationStore { + fn describe(&mut self) -> Result { + Ok(wit_descriptor()) + } + + fn observe(&mut self, _request: ObserveRequest) -> WitObservationResult { + self.capability().entered.wait(); + let reason = self.control().wait_for_interruption(); + WitObservationResult::Failed(Some(format!("{reason:?}"))) + } +} + +impl CapabilityModule for BlockingCapabilities { + type Invocation = BlockingInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[PROVIDER_IMPORT] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + fixture::FixtureHost::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, _context: CapabilityContext<'_>) -> Self::Invocation { + BlockingInvocation { + entered: Arc::clone(&self.entered), + } + } +} + +fn executor(config: RuntimeConfig, cache: Option) -> Executor { + Executor::new(config, cache, FixtureCapabilities).unwrap() +} + +fn request() -> ObservationRequest { + ObservationRequest { + uri: "fixture://resource".to_owned(), + selector: json!({"region": "local"}), + prior_digest: None, + demand_watermark: None, + } +} + +fn component(wat: &str) -> Vec { + wat::parse_str(wat).unwrap() +} + +fn with_core_behavior(core: &str) -> Vec { + component(&NO_EFFECT.replacen("(component", &format!("(component\n{core}"), 1)) +} + +fn publication_bytes(result: ObservationResult) -> Vec { + match result { + ObservationResult::Published { publication } => publication.bytes.into_vec(), + other => panic!("expected publication, got {other:?}"), + } +} + +#[test] +fn executes_the_repository_provider_world() { + let executor = executor(RuntimeConfig::default(), None); + let loaded = executor.load(&component(NO_EFFECT)).unwrap(); + let descriptor = executor.describe(&loaded, None).unwrap(); + assert_eq!(descriptor.capabilities, vec![SchedulingCapability::Demand]); + assert_eq!(descriptor.default_selector, json!({})); + let mut demand = request(); + demand.demand_watermark = Some(7); + let result = executor.observe(&loaded, &demand, None).unwrap(); + assert_eq!(publication_bytes(result), vec![7]); + demand.demand_watermark = Some(0); + assert!(matches!( + executor.observe(&loaded, &demand, None), + Err(ObserveError::InvalidRequest( + "demand watermark must be positive" + )) + )); +} + +#[test] +fn host_rejects_a_semantically_invalid_guest_proposal() { + let executor = executor(RuntimeConfig::default(), None); + let loaded = executor.load(&component(NO_EFFECT)).unwrap(); + let mut invalid = request(); + invalid.uri = "fixture://invalid".to_owned(); + assert!(matches!( + executor.observe(&loaded, &invalid, None), + Err(ObserveError::InvalidProposal(_)) + )); +} + +#[test] +fn a_trap_is_contained_and_structured() { + let bytes = with_core_behavior( + "(core module $behavior (func $start unreachable) (start $start))\n\ + (core instance $running (instantiate $behavior))", + ); + let executor = executor(RuntimeConfig::default(), None); + let loaded = executor.load(&bytes).unwrap(); + assert!(matches!( + executor.observe(&loaded, &request(), None), + Err(ObserveError::Trap(_)) + )); +} + +#[test] +fn fuel_exhaustion_is_distinct_from_guest_traps() { + let bytes = with_core_behavior( + "(core module $behavior\n\ + (func $start (loop $spin (br $spin)))\n\ + (start $start))\n\ + (core instance $running (instantiate $behavior))", + ); + let mut config = RuntimeConfig::default(); + config.fuel_per_observation = 10_000; + let executor = executor(config, None); + let loaded = executor.load(&bytes).unwrap(); + assert!(matches!( + executor.observe(&loaded, &request(), None), + Err(ObserveError::FuelExhausted) + )); +} + +#[test] +fn linear_memory_limit_is_enforced_during_instantiation() { + let bytes = with_core_behavior( + "(core module $behavior (memory 2))\n\ + (core instance $running (instantiate $behavior))", + ); + let mut config = RuntimeConfig::default(); + config.max_memory_bytes = 64 * 1024; + let executor = executor(config, None); + let loaded = executor.load(&bytes).unwrap(); + assert!(matches!( + executor.observe(&loaded, &request(), None), + Err(ObserveError::ResourceLimit(LimitKind::Memory)) + )); +} + +#[test] +fn instance_count_limit_is_enforced_as_an_instantiation_error() { + let mut config = RuntimeConfig::default(); + config.max_instances = 1; + let executor = executor(config, None); + let loaded = executor.load(&component(NO_EFFECT)).unwrap(); + assert!(matches!( + executor.observe(&loaded, &request(), None), + Err(ObserveError::Instantiation(_)) + )); +} + +#[test] +fn deterministic_epoch_handle_classifies_timeout_and_cancel() { + let bytes = with_core_behavior( + "(core module $behavior\n\ + (func $start (loop $spin (br $spin)))\n\ + (start $start))\n\ + (core instance $running (instantiate $behavior))", + ); + let mut config = RuntimeConfig::default(); + config.fuel_per_observation = u64::MAX; + let executor = executor(config, None); + let loaded = executor.load(&bytes).unwrap(); + + let timeout = executor.interruption_handle(); + assert!(timeout.time_out()); + assert!(matches!( + executor.observe(&loaded, &request(), Some(&timeout)), + Err(ObserveError::TimedOut) + )); + + let cancellation = executor.interruption_handle(); + assert!(cancellation.cancel()); + assert!(matches!( + executor.observe(&loaded, &request(), Some(&cancellation)), + Err(ObserveError::Cancelled) + )); + + let healthy = executor.load(&component(NO_EFFECT)).unwrap(); + assert_eq!( + publication_bytes(executor.observe(&healthy, &request(), None).unwrap()), + vec![1] + ); +} +enum BlockedInterruption { + Cancel, + TimeOut, +} + +fn interrupt_blocked_import(interruption: BlockedInterruption) -> ObserveError { + let entered = Arc::new(Barrier::new(2)); + let executor = Executor::new( + RuntimeConfig::default(), + None, + BlockingCapabilities { + entered: Arc::clone(&entered), + }, + ) + .unwrap(); + let loaded = executor.load(&component(NO_EFFECT)).unwrap(); + let handle = executor.interruption_handle(); + let trigger = handle.clone(); + let observing = + std::thread::spawn(move || executor.observe(&loaded, &request(), Some(&handle))); + entered.wait(); + match interruption { + BlockedInterruption::Cancel => assert!(trigger.cancel()), + BlockedInterruption::TimeOut => assert!(trigger.time_out()), + } + observing.join().unwrap().unwrap_err() +} + +#[test] +fn cancellation_wakes_a_blocked_capability_import() { + assert!(matches!( + interrupt_blocked_import(BlockedInterruption::Cancel), + ObserveError::Cancelled + )); +} + +#[test] +fn timeout_wakes_a_blocked_capability_import() { + assert!(matches!( + interrupt_blocked_import(BlockedInterruption::TimeOut), + ObserveError::TimedOut + )); +} + +#[test] +fn every_observation_gets_fresh_store_state() { + let executor = executor(RuntimeConfig::default(), None); + let loaded = executor.load(&component(NO_EFFECT)).unwrap(); + assert_eq!( + publication_bytes(executor.observe(&loaded, &request(), None).unwrap()), + vec![1] + ); + assert_eq!( + publication_bytes(executor.observe(&loaded, &request(), None).unwrap()), + vec![1] + ); +} + +#[test] +fn imports_are_rejected_unless_an_explicit_capability_module_admits_them() { + let executor = Executor::::closed(RuntimeConfig::default(), None).unwrap(); + let result = executor.load(&component(NO_EFFECT)); + assert!(matches!( + result, + Err(LoadError::ForbiddenImports(imports)) if imports == [PROVIDER_IMPORT] + )); +} + +#[test] +fn verified_aot_artifact_is_reused_by_a_new_executor() { + let temporary = tempfile::tempdir().unwrap(); + let cache = PrivateArtifactCache::open(temporary.path().join("cache")).unwrap(); + let bytes = component(NO_EFFECT); + let first = executor(RuntimeConfig::default(), Some(cache.clone())); + let loaded = first.load(&bytes).unwrap(); + assert_eq!(loaded.cache_disposition(), &CacheDisposition::CompiledAndStored); + drop(first); + + let second = executor(RuntimeConfig::default(), Some(cache)); + let loaded = second.load(&bytes).unwrap(); + assert_eq!(loaded.cache_disposition(), &CacheDisposition::DiskHit); +} + +#[test] +fn changed_runtime_identity_gets_a_clean_cache_miss_and_new_manifest() { + let temporary = tempfile::tempdir().unwrap(); + let cache = PrivateArtifactCache::open(temporary.path().join("cache")).unwrap(); + let bytes = component(NO_EFFECT); + let first = executor(RuntimeConfig::default(), Some(cache.clone())); + let first_loaded = first.load(&bytes).unwrap(); + assert_eq!( + first_loaded.cache_disposition(), + &CacheDisposition::CompiledAndStored + ); + drop(first); + + let mut changed_config = RuntimeConfig::default(); + changed_config.fuel_per_observation += 1; + let second = executor(changed_config, Some(cache.clone())); + let second_loaded = second.load(&bytes).unwrap(); + assert_eq!( + second_loaded.cache_disposition(), + &CacheDisposition::CompiledAndStored + ); + + let manifests = fs::read_dir( + cache + .root() + .join("manifests") + .join(second_loaded.digest().to_string()), + ) + .unwrap() + .count(); + assert_eq!(manifests, 2); +} + +#[test] +fn corrupt_artifact_is_never_deserialized() { + let temporary = tempfile::tempdir().unwrap(); + let cache = PrivateArtifactCache::open(temporary.path().join("cache")).unwrap(); + let bytes = component(NO_EFFECT); + let first = executor(RuntimeConfig::default(), Some(cache.clone())); + first.load(&bytes).unwrap(); + drop(first); + + let artifact = fs::read_dir(cache.root().join("objects")) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let mut corrupt = fs::read(&artifact).unwrap(); + corrupt[0] ^= 0xff; + fs::write(artifact, corrupt).unwrap(); + + let second = executor(RuntimeConfig::default(), Some(cache)); + let loaded = second.load(&bytes).unwrap(); + assert_eq!( + loaded.cache_disposition(), + &CacheDisposition::RejectedAndCompiled(CacheRejection::ArtifactDigest) + ); + assert_eq!( + publication_bytes(second.observe(&loaded, &request(), None).unwrap()), + vec![1] + ); +} + +#[test] +fn manifest_engine_and_config_identity_mismatches_invalidate_aot() { + for field in ["engineCompatibility", "configIdentity"] { + let temporary = tempfile::tempdir().unwrap(); + let cache = PrivateArtifactCache::open(temporary.path().join("cache")).unwrap(); + let bytes = component(NO_EFFECT); + let first = executor(RuntimeConfig::default(), Some(cache.clone())); + let loaded = first.load(&bytes).unwrap(); + let manifest_directory = cache + .root() + .join("manifests") + .join(loaded.digest().to_string()); + drop(first); + let manifest = fs::read_dir(manifest_directory) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let mut value: serde_json::Value = + serde_json::from_slice(&fs::read(&manifest).unwrap()).unwrap(); + value[field] = json!("mismatched"); + fs::write(&manifest, serde_json::to_vec(&value).unwrap()).unwrap(); + + let second = executor(RuntimeConfig::default(), Some(cache)); + let loaded = second.load(&bytes).unwrap(); + let expected = if field == "engineCompatibility" { + CacheRejection::EngineCompatibility + } else { + CacheRejection::ConfigIdentity + }; + assert_eq!( + loaded.cache_disposition(), + &CacheDisposition::RejectedAndCompiled(expected) + ); + } +} diff --git a/crates/st2-resource-wasip2/tests/fixtures/no-effect-component.wat b/crates/st2-resource-wasip2/tests/fixtures/no-effect-component.wat new file mode 100644 index 00000000..f7689c20 --- /dev/null +++ b/crates/st2-resource-wasip2/tests/fixtures/no-effect-component.wat @@ -0,0 +1,145 @@ +(component + (type $provider-api (instance + (type $scheduling-capability' (enum "demand")) + (export "scheduling-capability" (type $scheduling-capability (eq $scheduling-capability'))) + (type $provider-descriptor' (record + (field "capabilities" (list $scheduling-capability)) + (field "selector-schema-json" string) + (field "default-selector-json" string) + (field "topics" (list string)) + (field "snapshot-media-type" string) + (field "snapshot-schema-id" string))) + (export "provider-descriptor" (type $provider-descriptor (eq $provider-descriptor'))) + (type $descriptor-error' (variant + (case "invalid-descriptor" string) + (case "unavailable" string))) + (export "descriptor-error" (type $descriptor-error (eq $descriptor-error'))) + (type $observe-request' (record + (field "uri" string) + (field "selector-json" string) + (field "prior-digest" (option (list u8))) + (field "demand-watermark" (option u64)))) + (export "observe-request" (type $observe-request (eq $observe-request'))) + (type $fact-value' (variant + (case "omitted") + (case "null") + (case "value" string))) + (export "fact-value" (type $fact-value (eq $fact-value'))) + (type $fact' (record + (field "key" string) + (field "before" $fact-value) + (field "after" $fact-value))) + (export "fact" (type $fact (eq $fact'))) + (type $publication' (record + (field "schema-id" string) + (field "media-type" string) + (field "bytes" (list u8)) + (field "topics" (list string)) + (field "facts" (option (list $fact))))) + (export "publication" (type $publication (eq $publication'))) + (type $observation-result' (variant + (case "unchanged") + (case "failed" (option string)) + (case "published" $publication))) + (export "observation-result" (type $observation-result (eq $observation-result'))) + (type $describe-result (result $provider-descriptor (error $descriptor-error))) + (type $describe (func (result $describe-result))) + (export "describe" (func (type $describe))) + (type $observe (func (param "request" $observe-request) (result $observation-result))) + (export "observe" (func (type $observe))) + )) + (import "st2:resource-provider/provider-api@0.1.0" (instance $host (type $provider-api))) + (alias export $host "scheduling-capability" (type $scheduling-capability)) + (alias export $host "provider-descriptor" (type $provider-descriptor)) + (alias export $host "descriptor-error" (type $descriptor-error)) + (alias export $host "observe-request" (type $observe-request)) + (alias export $host "fact-value" (type $fact-value)) + (alias export $host "fact" (type $fact)) + (alias export $host "publication" (type $publication)) + (alias export $host "observation-result" (type $observation-result)) + (alias export $host "describe" (func $host-describe)) + (alias export $host "observe" (func $host-observe)) + (type $describe-result (result $provider-descriptor (error $descriptor-error))) + (type $describe-type (func (result $describe-result))) + (type $observe-type (func (param "request" $observe-request) (result $observation-result))) + (core module $abi + (memory (export "memory") 1) + (global $heap (mut i32) (i32.const 1024)) + (func (export "realloc") (param i32 i32 i32 i32) (result i32) + (local $pointer i32) + global.get $heap + local.get 2 + i32.const 1 + i32.sub + i32.add + i32.const 0 + local.get 2 + i32.sub + i32.and + local.tee $pointer + local.get 3 + i32.add + global.set $heap + local.get $pointer) + ) + (core instance $abi-instance (instantiate $abi)) + (alias core export $abi-instance "memory" (core memory $memory)) + (alias core export $abi-instance "realloc" (core func $realloc)) + (core func $lowered-describe (canon lower (func $host-describe) + (memory $memory) (realloc $realloc) string-encoding=utf8)) + (core func $lowered-observe (canon lower (func $host-observe) + (memory $memory) (realloc $realloc) string-encoding=utf8)) + (core instance $lowered-instance + (export "describe" (func $lowered-describe)) + (export "observe" (func $lowered-observe))) + (core module $adapter + (import "host" "describe" (func $lowered-describe (param i32))) + (import "host" "observe" + (func $lowered-observe + (param i32 i32 i32 i32 i32 i32 i32 i32 i64 i32))) + (func (export "describe") (result i32) + i32.const 0 + call $lowered-describe + i32.const 0) + (func (export "observe") + (param $p0 i32) (param $p1 i32) (param $p2 i32) (param $p3 i32) + (param $p4 i32) (param $p5 i32) (param $p6 i32) (param $p7 i32) + (param $p8 i64) + (result i32) + local.get $p0 + local.get $p1 + local.get $p2 + local.get $p3 + local.get $p4 + local.get $p5 + local.get $p6 + local.get $p7 + local.get $p8 + i32.const 0 + call $lowered-observe + i32.const 0) + ) + (core instance $adapter-instance (instantiate $adapter + (with "host" (instance $lowered-instance)))) + (alias core export $adapter-instance "describe" (core func $adapted-describe)) + (alias core export $adapter-instance "observe" (core func $adapted-observe)) + (func $implemented-describe (type $describe-type) + (canon lift (core func $adapted-describe) + (memory $memory) (realloc $realloc) string-encoding=utf8)) + (func $implemented-observe (type $observe-type) + (canon lift (core func $adapted-observe) + (memory $memory) (realloc $realloc) string-encoding=utf8)) + (instance $api + (export "scheduling-capability" (type $scheduling-capability)) + (export "provider-descriptor" (type $provider-descriptor)) + (export "descriptor-error" (type $descriptor-error)) + (export "observe-request" (type $observe-request)) + (export "fact-value" (type $fact-value)) + (export "fact" (type $fact)) + (export "publication" (type $publication)) + (export "observation-result" (type $observation-result)) + (export "describe" (func $implemented-describe)) + (export "observe" (func $implemented-observe)) + ) + (export "provider-api" (instance $api)) +) diff --git a/crates/st2-resource-wasip2/wit/provider.wit b/crates/st2-resource-wasip2/wit/provider.wit new file mode 100644 index 00000000..2c7bc520 --- /dev/null +++ b/crates/st2-resource-wasip2/wit/provider.wit @@ -0,0 +1,65 @@ +package st2:resource-provider@0.1.0; + +interface provider-api { + enum scheduling-capability { + demand, + } + + record provider-descriptor { + capabilities: list, + selector-schema-json: string, + default-selector-json: string, + topics: list, + snapshot-media-type: string, + snapshot-schema-id: string, + } + + variant descriptor-error { + invalid-descriptor(string), + unavailable(string), + } + + record observe-request { + uri: string, + selector-json: string, + prior-digest: option>, + demand-watermark: option, + } + + variant fact-value { + omitted, + null, + value(string), + } + + record fact { + key: string, + before: fact-value, + after: fact-value, + } + + record publication { + schema-id: string, + media-type: string, + bytes: list, + topics: list, + facts: option>, + } + + variant observation-result { + unchanged, + failed(option), + published(publication), + } + + describe: func() -> result; + observe: func(request: observe-request) -> observation-result; +} + +world provider { + export provider-api: provider-api; +} + +world fixture-host { + import provider-api; +} diff --git a/flake.nix b/flake.nix index 9864f429..f10e9044 100644 --- a/flake.nix +++ b/flake.nix @@ -79,6 +79,7 @@ # env var, captured at compile time by `option_env!` (see # src/version.rs). A derivation env var change rebuilds the crate. CLI_BUILD_STAMP = buildStamp; + ST2_EXECUTOR_BUILD_IDENTITY = buildStamp; AGENT_SPEC_REVISION = agentSpecRevision; # The hook integration test executes the shipped Bash scripts with @@ -179,6 +180,22 @@ ]; }); + # The default workspace remains Wasmtime-free; this focused gate opts the Component Model + # executor into its runtime feature and drives its fixture and cache trust boundary. + st2Wasip2ExecutorCheck = st2.overrideAttrs (old: { + pname = "st2-resource-wasip2-check"; + nativeBuildInputs = (old.nativeBuildInputs or [ ]) ++ [ pkgs.lld ]; + cargoTestFlags = [ + "-p" + "st2-resource-wasip2" + "--features" + "runtime" + "--lib" + "--test" + "executor" + ]; + }); + # Narrow sandbox-safe integration gate for the atomic snapshot boundary. The main package # deliberately omits the broad doctor suite because some doctor cases exercise facilities # unavailable in the Nix sandbox. A dedicated target containing exactly one test makes the @@ -296,6 +313,7 @@ checks.parked-recovery = st2ParkedRecovery; checks.otel-export = st2OtelExport; checks.wasm-resolver-feature = st2WasmResolverCheck; + checks.wasip2-resource-executor = st2Wasip2ExecutorCheck; # Exercise the shipped binary, not a cargo-side surrogate: its version entrypoint runs and # the same artifact strictly admits a catalog carrying a real wasm profile module. checks.wasm-resolver-artifact = pkgs.runCommand "st2-wasm-resolver-artifact-${version}" { } ''