From 3a13301f758c001b41836f43f55d5c860c07d17c Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 03:51:57 +0900 Subject: [PATCH] feat: deliver exact Plugin artifacts through current Host authority Bind remote and in-process delivery to authenticated Worker lineage, live installation permission and persisted exact pins. Verify canonical bytes and complete descriptors in a bounded Runtime-only atomic cache; cache hits reauthorize and process-crash staging never becomes visible. Shares the exact PluginInstallation permission shape published for T-793; no HTTPS or launch/restore integration is included. Depends on the T-465 source ancestry and the shared Host execution identity helper. --- Cargo.lock | 4 + crates/manifest/src/plugin.rs | 23 + crates/plugin-runtime/Cargo.toml | 5 +- crates/plugin-runtime/README.md | 4 +- crates/plugin-runtime/src/delivery.rs | 509 ++++++++ crates/plugin-runtime/src/delivery/tests.rs | 985 ++++++++++++++ crates/plugin-runtime/src/lib.rs | 1 + crates/server-api/src/lib.rs | 20 + crates/server-api/src/plugin.rs | 34 + crates/server-api/src/plugin_types.rs | 53 + crates/worker-runtime/Cargo.toml | 4 +- .../worker-runtime/src/workspace_request.rs | 2 + .../src/workspace_request/plugin_delivery.rs | 182 +++ crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/lib.rs | 1 + crates/workspace-server/src/plugin_catalog.rs | 12 +- .../workspace-server/src/plugin_delivery.rs | 146 +++ .../src/plugin_delivery/tests.rs | 591 +++++++++ crates/workspace-server/src/server.rs | 29 + crates/workspace-server/src/server_plugin.rs | 4 +- .../src/server_plugin_delivery.rs | 215 +++ .../src/server_plugin_delivery_tests.rs | 1153 +++++++++++++++++ web/workspace/src/lib/generated/plugin-api.ts | 4 + 23 files changed, 3971 insertions(+), 11 deletions(-) create mode 100644 crates/plugin-runtime/src/delivery.rs create mode 100644 crates/plugin-runtime/src/delivery/tests.rs create mode 100644 crates/worker-runtime/src/workspace_request/plugin_delivery.rs create mode 100644 crates/workspace-server/src/plugin_delivery.rs create mode 100644 crates/workspace-server/src/plugin_delivery/tests.rs create mode 100644 crates/workspace-server/src/server_plugin_delivery.rs create mode 100644 crates/workspace-server/src/server_plugin_delivery_tests.rs diff --git a/Cargo.lock b/Cargo.lock index ea40ae292..27472dcb8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7318,6 +7318,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", + "tokio-util", "toml", "tools", "tower", @@ -7327,6 +7328,7 @@ dependencies = [ "uuid", "workdir", "worker", + "yoi-plugin-runtime", "zeroize", ] @@ -7468,11 +7470,13 @@ name = "yoi-plugin-runtime" version = "0.1.0" dependencies = [ "async-trait", + "fs4", "jsonschema", "manifest", "serde", "serde_json", "server-api", + "sha2 0.11.0", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/crates/manifest/src/plugin.rs b/crates/manifest/src/plugin.rs index 33eb85330..7511a97b4 100644 --- a/crates/manifest/src/plugin.rs +++ b/crates/manifest/src/plugin.rs @@ -602,6 +602,29 @@ pub fn verify_plugin_artifact( Ok(package) } +/// Static schema identity shared by catalog publication and Runtime verification. +/// Derivation belongs with the package contract so consumers cannot invent a +/// different schema hash or namespace while agreeing on archive bytes. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PluginConfigSchemaIdentity { + pub identity: String, + pub version: String, + pub digest: String, + pub namespace: String, +} + +pub fn plugin_config_schema_identity( + manifest: &PluginPackageManifest, +) -> PluginConfigSchemaIdentity { + let digest = artifact_digest(manifest.config_schema.as_bytes()); + PluginConfigSchemaIdentity { + identity: format!("{}/config", manifest.id), + version: manifest.version.clone(), + namespace: format!("{}/config/{}/{}", manifest.id, manifest.version, digest), + digest, + } +} + fn artifact_digest(bytes: &[u8]) -> String { format!("sha256:{}", hex_lower(&Sha256::digest(bytes))) } diff --git a/crates/plugin-runtime/Cargo.toml b/crates/plugin-runtime/Cargo.toml index 2f377fa15..e31a1a7b5 100644 --- a/crates/plugin-runtime/Cargo.toml +++ b/crates/plugin-runtime/Cargo.toml @@ -9,10 +9,13 @@ rust-version.workspace = true async-trait.workspace = true manifest.workspace = true server-api.workspace = true +sha2.workspace = true +fs4.workspace = true +tempfile.workspace = true serde = { workspace = true, features = ["derive"] } serde_json.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["rt", "macros", "sync", "time"] } +tokio = { workspace = true, features = ["rt", "macros", "sync", "time", "fs", "io-util"] } tokio-util = { version = "0.7", features = ["rt"] } jsonschema = { version = "0.33.0", default-features = false } wasmparser = "0.244.0" diff --git a/crates/plugin-runtime/README.md b/crates/plugin-runtime/README.md index c70471a70..ecfd5d6e3 100644 --- a/crates/plugin-runtime/README.md +++ b/crates/plugin-runtime/README.md @@ -1,6 +1,8 @@ # yoi-plugin-runtime -Runtime-owned, standalone execution of exact v2 Tool Components. [`Executor`] compiles only artifacts reverified through `manifest::plugin`, then creates a fresh Store and instance for each invocation. It does not resolve catalogs, config, Worker plans, delivery caches or repository paths, and does not register Tools with a Worker. +Runtime-owned verification, byte caching and standalone execution of exact v2 Tool Components. [`Executor`] compiles only artifacts reverified through `manifest::plugin`, then creates a fresh Store and instance for each invocation. It does not resolve catalogs, config, Worker plans or repository paths, and does not register Tools with a Worker. + +[`delivery`] caches canonical bytes under the authenticated Server authority and artifact digest. Its source adapter must reauthorize the exact saved plan selection on every hit and after transfer. Byte deduplication shares neither metadata nor installation/config/grant authority. A private Runtime directory, bounded staging, process-released locking and atomic publication prevent partial files from becoming executor inputs; abandoned staging is discarded under the same lock. Cache deletion never releases Server pins. The cache is not an offline execution lease, and verified bytes still require the execution Host's fresh admission. [`host`] is an explicit asynchronous injection boundary, not an implementation of Server authority or HTTPS. Every invocation (including pure computation) requires live admission and result-publication authorization. Providers must bind their authenticated installation/plan/subject and narrowed grants to [`host::InvocationContext`], authorize every protected operation, and make pending work cancellation-safe. There is no default provider or product allow-all fallback. Dropping a provider future must release temporary resources; an already dispatched external effect is not rolled back. Durable effect receipts, secret resolution, URL/DNS/TLS policy and real network transport belong to the Host provider, not this sandbox. diff --git a/crates/plugin-runtime/src/delivery.rs b/crates/plugin-runtime/src/delivery.rs new file mode 100644 index 000000000..5bfad2163 --- /dev/null +++ b/crates/plugin-runtime/src/delivery.rs @@ -0,0 +1,509 @@ +//! Runtime-owned byte cache, not a source of admission, configuration or grants. +//! +//! The caller supplies a private Runtime directory and an expectation from its +//! trusted Host plan, never a model-selected path or owner-supplied expectation. +//! Every resolution checks the live owner twice, including hits. The returned +//! bytes prove format/identity only; execution still requires current admission. +//! A process-released filesystem lock bounds all final and pending files together. +//! Only synced, completely verified files are renamed into the visible namespace; +//! abandoned staging files are discarded under that same lock after restart. + +use async_trait::async_trait; +use manifest::plugin::{self, PluginPackageLimits}; +use server_api::{PluginArtifact, PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, File, OpenOptions}, + io::{self, Read}, + path::{Path, PathBuf}, + pin::Pin, + time::Duration, +}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; + +pub type ArtifactStream = Pin>; + +#[async_trait] +pub trait ArtifactSource: Send + Sync { + /// Stable authenticated Server authority, not a workspace or connection ID. + fn server_authority_identity(&self) -> &str; + async fn authorize( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> Result; + async fn fetch( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> Result; +} + +#[derive(Debug, thiserror::Error)] +pub enum DeliveryError { + #[error("artifact delivery denied by current owner")] + Denied, + #[error("artifact owner unavailable: {0}")] + Unavailable(String), + #[error("invalid artifact delivery request")] + InvalidRequest, + #[error("artifact metadata differs from trusted Host plan")] + MetadataMismatch, + #[error("artifact exceeds delivery bound or declared size")] + TooLarge, + #[error("invalid artifact: {0}")] + InvalidArtifact(String), + #[error("cache directory must be private and Runtime-owned")] + InvalidCacheRoot, + #[error("invalid artifact cache limits")] + InvalidLimits, + #[error("artifact delivery deadline exceeded")] + Timeout, + #[error("artifact cache I/O failed: {0}")] + Io(#[from] io::Error), +} + +/// Exact immutable descriptor selected by the Host, not metadata learned from a +/// delivery response. Keeping the complete descriptor also binds Tool schemas, +/// connection declarations, requested (not effective) grants and schema identity. +#[derive(Clone, Debug)] +pub struct ArtifactExpectation { + artifact: PluginArtifact, +} + +impl ArtifactExpectation { + pub fn new(host_plan_artifact: PluginArtifact) -> Self { + Self { + artifact: host_plan_artifact, + } + } + + pub fn descriptor(&self) -> &PluginArtifact { + &self.artifact + } +} + +/// Bytes have no public constructor or mutable accessor, and contain no cached +/// owner authorization, Worker configuration, effective grants or pin ownership. +pub struct VerifiedArtifact { + bytes: Vec, +} + +impl std::fmt::Debug for VerifiedArtifact { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("VerifiedArtifact") + .field("byte_length", &self.bytes.len()) + .finish_non_exhaustive() + } +} + +impl VerifiedArtifact { + pub fn bytes(&self) -> &[u8] { + &self.bytes + } +} + +#[derive(Clone, Debug)] +pub struct CacheLimits { + pub package: PluginPackageLimits, + /// Includes the current staging download, not only published files. + pub max_cache_bytes: u64, + pub max_entries: usize, + pub delivery_timeout: Duration, +} + +impl Default for CacheLimits { + fn default() -> Self { + Self { + package: PluginPackageLimits::default(), + max_cache_bytes: 256 * 1024 * 1024, + max_entries: 32, + delivery_timeout: Duration::from_secs(60), + } + } +} + +#[derive(Debug)] +pub struct ArtifactCache { + root: PathBuf, + limits: CacheLimits, +} + +impl ArtifactCache { + /// The root belongs exclusively to the Runtime cache. Existing roots must + /// already be private; this never fixes permissions on arbitrary directories. + pub fn new( + runtime_owned_root: impl AsRef, + limits: CacheLimits, + ) -> Result { + if limits.max_cache_bytes == 0 + || limits.max_entries == 0 + || limits.package.max_package_size_bytes == 0 + || limits.delivery_timeout.is_zero() + { + return Err(DeliveryError::InvalidLimits); + } + let root = runtime_owned_root.as_ref(); + let mut builder = fs::DirBuilder::new(); + builder.recursive(true); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + builder.mode(0o700); + } + builder.create(root)?; + let metadata = fs::symlink_metadata(root)?; + if !metadata.is_dir() || metadata.file_type().is_symlink() { + return Err(DeliveryError::InvalidCacheRoot); + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if metadata.permissions().mode() & 0o077 != 0 { + return Err(DeliveryError::InvalidCacheRoot); + } + } + Ok(Self { + root: fs::canonicalize(root)?, + limits, + }) + } + + pub async fn resolve( + &self, + source: &dyn ArtifactSource, + trusted_workspace_id: &str, + request: &PluginArtifactDeliveryRequest, + expected: &ArtifactExpectation, + ) -> Result { + tokio::time::timeout( + self.limits.delivery_timeout, + self.resolve_inner(source, trusted_workspace_id, request, expected), + ) + .await + .map_err(|_| DeliveryError::Timeout)? + } + + async fn resolve_inner( + &self, + source: &dyn ArtifactSource, + workspace: &str, + request: &PluginArtifactDeliveryRequest, + expected: &ArtifactExpectation, + ) -> Result { + let authority = source.server_authority_identity().to_owned(); + let descriptor = &expected.artifact; + if authority.is_empty() + || workspace.is_empty() + || request.plan_id.is_empty() + || request.plan_fingerprint.is_empty() + || request.installation_id.is_empty() + || !is_digest(&request.artifact_digest) + || request.artifact_digest != descriptor.artifact_digest + || !is_digest(&descriptor.package_digest) + { + return Err(DeliveryError::InvalidRequest); + } + let size = u64::from(descriptor.artifact_size); + if size == 0 + || size > self.limits.package.max_package_size_bytes + || size > self.limits.max_cache_bytes + { + return Err(DeliveryError::TooLarge); + } + self.authorize(source, &authority, workspace, request, expected) + .await?; + + // Lock ownership follows this future, including cancellation/deadline. + let _lock = lock_cache(self.root.join(".cache.lock")).await?; + let mut entries = self.inventory()?; + let path = self.root.join(format!( + "{}-{}.artifact", + hash(authority.as_bytes()), + &request.artifact_digest[7..] + )); + let bytes = match self.read_verified(&path, expected)? { + Some(bytes) => { + self.make_room(&mut entries, Some(&path), 0, 0)?; + bytes + } + None => { + // Discard a corrupt final file only while holding the cache lock; + // no competing publisher can have repaired it since the read. + if let Some(index) = entries.iter().position(|entry| entry.path == path) { + fs::remove_file(&path)?; + entries.remove(index); + } + self.make_room(&mut entries, None, size, 1)?; + let mut stream = source.fetch(request).await?; + if source.server_authority_identity() != authority { + return Err(DeliveryError::MetadataMismatch); + } + let pending = tempfile::Builder::new() + .prefix(".pending-") + .tempfile_in(&self.root)?; + let mut file = tokio::fs::File::from_std(pending.reopen()?); + let mut written = 0_u64; + let mut buffer = [0_u8; 64 * 1024]; + loop { + // Read at most one byte beyond the declaration, even if the + // peer streams forever. Excess bytes are never written. + let remaining = (size - written + 1).min(buffer.len() as u64) as usize; + let n = stream.read(&mut buffer[..remaining]).await?; + if n == 0 { + break; + } + written += n as u64; + if written > size { + return Err(DeliveryError::TooLarge); + } + file.write_all(&buffer[..n]).await?; + } + if written != size { + return Err(DeliveryError::InvalidArtifact( + "truncated artifact stream".into(), + )); + } + file.flush().await?; + file.sync_all().await?; + drop(file); + let bytes = self.read_exact_bounded(pending.path(), size)?; + self.verify(&bytes, expected)?; + // tempfile persist is an atomic same-filesystem rename. The global + // lock prevents overwrite races; no final name ever exposes a prefix. + pending.persist(&path).map_err(|error| error.error)?; + sync_directory(&self.root)?; + bytes + } + }; + // Neither disk presence nor authorization before transfer proves current + // owner policy after a slow download, eviction wait or cached verification. + self.authorize(source, &authority, workspace, request, expected) + .await?; + Ok(VerifiedArtifact { bytes }) + } + + async fn authorize( + &self, + source: &dyn ArtifactSource, + authority: &str, + workspace: &str, + request: &PluginArtifactDeliveryRequest, + expected: &ArtifactExpectation, + ) -> Result<(), DeliveryError> { + let metadata = source.authorize(request).await?; + if metadata.artifact.revoked || expected.artifact.revoked { + return Err(DeliveryError::Denied); + } + if source.server_authority_identity() != authority + || metadata.workspace_id != workspace + || metadata.request != *request + || metadata.artifact != expected.artifact + { + return Err(DeliveryError::MetadataMismatch); + } + Ok(()) + } + + fn read_verified( + &self, + path: &Path, + expected: &ArtifactExpectation, + ) -> Result>, DeliveryError> { + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + if !metadata.is_file() || metadata.file_type().is_symlink() { + return Err(DeliveryError::InvalidCacheRoot); + } + if metadata.len() != u64::from(expected.artifact.artifact_size) { + return Ok(None); + } + let bytes = self.read_exact_bounded(path, metadata.len())?; + match self.verify(&bytes, expected) { + Ok(()) => Ok(Some(bytes)), + Err(DeliveryError::InvalidArtifact(_)) => Ok(None), + Err(error) => Err(error), + } + } + + fn read_exact_bounded(&self, path: &Path, size: u64) -> Result, DeliveryError> { + let mut bytes = Vec::new(); + File::open(path)?.take(size + 1).read_to_end(&mut bytes)?; + if bytes.len() as u64 != size { + return Err(DeliveryError::InvalidArtifact( + "cached artifact size changed".into(), + )); + } + Ok(bytes) + } + + fn verify(&self, bytes: &[u8], expected: &ArtifactExpectation) -> Result<(), DeliveryError> { + let d = &expected.artifact; + let package = plugin::verify_plugin_artifact( + bytes, + &d.package_digest, + &d.artifact_digest, + u64::from(d.artifact_size), + &self.limits.package, + ) + .map_err(|error| DeliveryError::InvalidArtifact(error.message))?; + let m = &package.package.manifest; + let schema = plugin::plugin_config_schema_identity(m); + if d.package_id != m.id + || d.name != m.name + || d.version != m.version + || d.description != m.description + || d.schema_version != m.schema_version + || d.world != m.runtime.world + || d.required_imports != m.runtime.required_imports + || !same_json(&d.tools, &m.tools)? + || !same_json(&d.connections, &m.connections)? + || !same_json(&d.requested_grants, &m.requested_grants)? + || d.config_schema.dcdl != m.config_schema + || d.config_schema.digest != schema.digest + || d.config_schema.identity != schema.identity + || d.config_schema.version != schema.version + || d.config_schema.namespace != schema.namespace + { + return Err(DeliveryError::MetadataMismatch); + } + Ok(()) + } + + fn inventory(&self) -> Result, DeliveryError> { + let mut entries = Vec::new(); + for entry in fs::read_dir(&self.root)? { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_str().ok_or(DeliveryError::InvalidCacheRoot)?; + let metadata = fs::symlink_metadata(entry.path())?; + if !metadata.is_file() || metadata.file_type().is_symlink() { + return Err(DeliveryError::InvalidCacheRoot); + } + if name == ".cache.lock" { + continue; + } + if name.starts_with(".pending-") { + fs::remove_file(entry.path())?; + continue; + } + if !is_cache_filename(name) { + return Err(DeliveryError::InvalidCacheRoot); + } + entries.push(Entry { + path: entry.path(), + size: metadata.len(), + modified: metadata.modified()?, + }); + } + // Oldest publication first; hits do not need a second metadata ledger. + entries.sort_by(|a, b| a.modified.cmp(&b.modified).then(a.path.cmp(&b.path))); + Ok(entries) + } + + fn make_room( + &self, + entries: &mut Vec, + keep: Option<&Path>, + additional_bytes: u64, + additional_entries: usize, + ) -> Result<(), DeliveryError> { + let mut bytes = entries.iter().try_fold(additional_bytes, |sum, entry| { + sum.checked_add(entry.size).ok_or(DeliveryError::TooLarge) + })?; + let mut count = entries.len() + additional_entries; + for entry in entries.iter() { + if bytes <= self.limits.max_cache_bytes && count <= self.limits.max_entries { + break; + } + if keep == Some(entry.path.as_path()) { + continue; + } + fs::remove_file(&entry.path)?; + bytes -= entry.size; + count -= 1; + } + if bytes > self.limits.max_cache_bytes || count > self.limits.max_entries { + return Err(DeliveryError::TooLarge); + } + Ok(()) + } +} + +struct Entry { + path: PathBuf, + size: u64, + modified: std::time::SystemTime, +} + +async fn lock_cache(path: PathBuf) -> Result { + if let Ok(metadata) = fs::symlink_metadata(&path) + && (!metadata.is_file() || metadata.file_type().is_symlink()) + { + return Err(DeliveryError::InvalidCacheRoot); + } + let mut options = OpenOptions::new(); + options.create(true).truncate(false).read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let file = options.open(path)?; + loop { + if fs4::fs_std::FileExt::try_lock_exclusive(&file)? { + return Ok(file); + } + // OS locks provide crash recovery; nonblocking acquisition keeps cancelled + // waiters from occupying Tokio blocking threads after their caller exits. + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +fn sync_directory(root: &Path) -> Result<(), DeliveryError> { + // Windows does not expose directory fsync through std::fs::File. Atomic + // visibility still holds there; power-loss directory durability is Unix-only. + #[cfg(unix)] + File::open(root)?.sync_all()?; + #[cfg(not(unix))] + let _ = root; + Ok(()) +} + +fn hash(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn is_hex(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +fn is_digest(value: &str) -> bool { + value.strip_prefix("sha256:").is_some_and(is_hex) +} + +fn is_cache_filename(value: &str) -> bool { + value + .strip_suffix(".artifact") + .and_then(|value| value.split_once('-')) + .is_some_and(|(authority, digest)| is_hex(authority) && is_hex(digest)) +} + +fn same_json( + a: &A, + b: &B, +) -> Result { + let a = serde_json::to_value(a).map_err(|_| DeliveryError::MetadataMismatch)?; + let b = serde_json::to_value(b).map_err(|_| DeliveryError::MetadataMismatch)?; + Ok(a == b) +} + +#[cfg(test)] +mod tests; diff --git a/crates/plugin-runtime/src/delivery/tests.rs b/crates/plugin-runtime/src/delivery/tests.rs new file mode 100644 index 000000000..e6dc60aed --- /dev/null +++ b/crates/plugin-runtime/src/delivery/tests.rs @@ -0,0 +1,985 @@ +//! Cache/real filesystem/shared Component package integration. Only owner policy +//! and transport are fixtures; canonical verification, streaming limits, locking, +//! eviction and publication use production code. Subprocess actors prove lock +//! convergence and process-exit recovery, not arbitrary power-loss durability. + +use super::*; +use manifest::plugin::{CanonicalPluginArtifact, PluginSourceKind}; +use std::{ + io::{BufRead, BufReader, Write}, + process::{Child, Command, Stdio}, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, + task::{Context, Poll}, +}; +use tokio::{io::ReadBuf, sync::Semaphore}; + +const MANIFEST: &str = + include_str!("../../../../resources/plugin/templates/rust-component-tool/plugin.toml"); +const COMPONENT: &[u8] = + include_bytes!("../../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"); + +fn fixture() -> (CanonicalPluginArtifact, PluginArtifact) { + let directory = tempfile::tempdir().unwrap(); + fs::write(directory.path().join("plugin.toml"), MANIFEST).unwrap(); + fs::write(directory.path().join("plugin.component.wasm"), COMPONENT).unwrap(); + let limits = PluginPackageLimits::default(); + let package = + plugin::read_plugin_directory(directory.path(), PluginSourceKind::Explicit, &limits) + .unwrap(); + let artifact = plugin::pack_plugin_package(&package, &limits).unwrap(); + let m = package.package.manifest; + let schema_digest = format!("sha256:{}", hash(m.config_schema.as_bytes())); + let descriptor = PluginArtifact { + package_id: m.id.clone(), + name: m.name, + version: m.version.clone(), + description: m.description, + package_digest: artifact.package_digest.clone(), + artifact_digest: artifact.artifact_digest.clone(), + artifact_size: artifact.artifact_size.try_into().unwrap(), + schema_version: m.schema_version, + world: m.runtime.world, + required_imports: m.runtime.required_imports, + tools: serde_json::from_value(serde_json::to_value(m.tools).unwrap()).unwrap(), + connections: vec![], + requested_grants: vec![], + config_schema: server_api::PluginSchemaContribution { + identity: format!("{}/config", m.id), + version: m.version.clone(), + namespace: format!("{}/config/{}/{}", m.id, m.version, schema_digest), + digest: schema_digest, + dcdl: m.config_schema, + }, + revoked: false, + revocation_reason: None, + }; + (artifact, descriptor) +} + +#[derive(Clone, Copy)] +enum Transfer { + Complete, + Short, + Error, + Pause, + Crash, +} + +struct Source { + authority: String, + metadata: Mutex, + bytes: Vec, + fetches: AtomicUsize, + authorizations: AtomicUsize, + deny_on: AtomicUsize, + unavailable: AtomicUsize, + transfer: Transfer, + read_bytes: Arc, + paused: Arc, +} + +impl Source { + fn new(artifact: &CanonicalPluginArtifact, descriptor: PluginArtifact) -> Self { + Self { + authority: "authenticated-server-a".into(), + metadata: Mutex::new(PluginArtifactDeliveryMetadata { + workspace_id: "workspace-a".into(), + request: PluginArtifactDeliveryRequest { + plan_id: "plan-a".into(), + plan_fingerprint: "exact-plan-fingerprint".into(), + installation_id: "installation-a".into(), + artifact_digest: artifact.artifact_digest.clone(), + }, + artifact: descriptor, + }), + bytes: artifact.bytes.clone(), + fetches: AtomicUsize::new(0), + authorizations: AtomicUsize::new(0), + deny_on: AtomicUsize::new(0), + unavailable: AtomicUsize::new(0), + transfer: Transfer::Complete, + read_bytes: Arc::new(AtomicUsize::new(0)), + paused: Arc::new(Semaphore::new(0)), + } + } + + fn request(&self) -> PluginArtifactDeliveryRequest { + self.metadata.lock().unwrap().request.clone() + } + + fn expectation(&self) -> ArtifactExpectation { + ArtifactExpectation::new(self.metadata.lock().unwrap().artifact.clone()) + } + + async fn resolve(&self, cache: &ArtifactCache) -> Result { + cache + .resolve(self, "workspace-a", &self.request(), &self.expectation()) + .await + } +} + +#[async_trait] +impl ArtifactSource for Source { + fn server_authority_identity(&self) -> &str { + &self.authority + } + + async fn authorize( + &self, + _: &PluginArtifactDeliveryRequest, + ) -> Result { + let call = self.authorizations.fetch_add(1, Ordering::SeqCst) + 1; + if self.unavailable.load(Ordering::SeqCst) != 0 { + return Err(DeliveryError::Unavailable("offline fixture".into())); + } + if call == self.deny_on.load(Ordering::SeqCst) { + return Err(DeliveryError::Denied); + } + Ok(self.metadata.lock().unwrap().clone()) + } + + async fn fetch( + &self, + _: &PluginArtifactDeliveryRequest, + ) -> Result { + self.fetches.fetch_add(1, Ordering::SeqCst); + // A real bounded in-memory transport pipe fragments successful downloads. + // Fault streams replace only that transport, not the filesystem/cache. + if matches!(self.transfer, Transfer::Complete | Transfer::Short) { + let (reader, mut writer) = tokio::io::duplex(127); + let mut bytes = self.bytes.clone(); + if matches!(self.transfer, Transfer::Short) { + bytes.truncate(bytes.len() / 2); + } + tokio::spawn(async move { + for chunk in bytes.chunks(61) { + if writer.write_all(chunk).await.is_err() { + break; + } + } + }); + return Ok(Box::pin(CountingReader { + reader, + count: self.read_bytes.clone(), + })); + } + Ok(Box::pin(FaultReader { + bytes: self.bytes[..self.bytes.len() / 2].to_vec(), + position: 0, + transfer: self.transfer, + paused: self.paused.clone(), + })) + } +} + +struct CountingReader { + reader: R, + count: Arc, +} +impl AsyncRead for CountingReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + let before = buffer.filled().len(); + let result = Pin::new(&mut self.reader).poll_read(cx, buffer); + self.count + .fetch_add(buffer.filled().len() - before, Ordering::SeqCst); + result + } +} + +struct FaultReader { + bytes: Vec, + position: usize, + transfer: Transfer, + paused: Arc, +} +impl AsyncRead for FaultReader { + fn poll_read( + mut self: Pin<&mut Self>, + _: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + if self.position < self.bytes.len() { + let n = buffer.remaining().min(self.bytes.len() - self.position); + buffer.put_slice(&self.bytes[self.position..self.position + n]); + self.position += n; + return Poll::Ready(Ok(())); + } + match self.transfer { + Transfer::Error => Poll::Ready(Err(io::Error::new( + io::ErrorKind::ConnectionReset, + "interrupted fixture", + ))), + Transfer::Pause => { + // Reaching this poll proves the cache consumed/wrote the prefix + // and is awaiting further input, before any publication. + self.paused.add_permits(1); + Poll::Pending + } + Transfer::Crash => std::process::exit(73), + _ => unreachable!(), + } + } +} + +fn cache(directory: &tempfile::TempDir) -> ArtifactCache { + ArtifactCache::new(directory.path().join("cache"), CacheLimits::default()).unwrap() +} + +fn final_files(cache: &ArtifactCache) -> Vec { + fs::read_dir(&cache.root) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "artifact")) + .collect() +} + +fn pending_files(cache: &ArtifactCache) -> Vec { + fs::read_dir(&cache.root) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| { + path.file_name() + .unwrap() + .to_string_lossy() + .starts_with(".pending-") + }) + .collect() +} + +#[tokio::test] +async fn streamed_download_and_restart_hit_return_only_verified_bytes() { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + let verified = source.resolve(&cache).await.unwrap(); + assert_eq!(verified.bytes(), artifact.bytes); + assert_eq!(fs::read(&final_files(&cache)[0]).unwrap(), artifact.bytes); + let restarted = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + assert_eq!( + source.resolve(&restarted).await.unwrap().bytes(), + artifact.bytes + ); + assert_eq!(source.fetches.load(Ordering::SeqCst), 1); + assert_eq!(source.authorizations.load(Ordering::SeqCst), 4); + assert!(pending_files(&cache).is_empty()); +} + +#[tokio::test] +async fn cached_bytes_do_not_bypass_owner_outage_revocation_or_denial() { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + source.resolve(&cache).await.unwrap(); + source.unavailable.store(1, Ordering::SeqCst); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::Unavailable(_)) + )); + source.unavailable.store(0, Ordering::SeqCst); + source.deny_on.store(4, Ordering::SeqCst); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::Denied) + )); + source.metadata.lock().unwrap().artifact.revoked = true; + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::Denied) + )); + assert_eq!(source.fetches.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn final_authorization_denies_both_download_and_cache_hit() { + for hit in [false, true] { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + if hit { + source.resolve(&cache).await.unwrap(); + } + source + .deny_on + .store(if hit { 4 } else { 2 }, Ordering::SeqCst); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::Denied) + )); + assert_eq!(source.fetches.load(Ordering::SeqCst), 1); + // A verified blob can remain after denial, but is not usable authority. + assert_eq!(final_files(&cache).len(), 1); + } +} + +#[tokio::test] +async fn response_must_bind_trusted_workspace_and_every_exact_request_field() { + for field in [ + "workspace", + "plan", + "fingerprint", + "installation", + "artifact", + ] { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let request = source.request(); + let expected = source.expectation(); + { + let mut metadata = source.metadata.lock().unwrap(); + match field { + "workspace" => metadata.workspace_id = "another-workspace".into(), + "plan" => metadata.request.plan_id.push('x'), + "fingerprint" => metadata.request.plan_fingerprint.push('x'), + "installation" => metadata.request.installation_id.push('x'), + "artifact" => metadata.request.artifact_digest.push('x'), + _ => unreachable!(), + } + } + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + assert!( + matches!( + cache + .resolve(&source, "workspace-a", &request, &expected) + .await, + Err(DeliveryError::MetadataMismatch) + ), + "{field}" + ); + assert_eq!(source.fetches.load(Ordering::SeqCst), 0); + assert!(final_files(&cache).is_empty()); + } +} + +#[tokio::test] +async fn metadata_cannot_replace_the_host_plan_descriptor() { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let expected = source.expectation(); + source.metadata.lock().unwrap().artifact.version = "owner-selected-version".into(); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + assert!(matches!( + cache + .resolve(&source, "workspace-a", &source.request(), &expected) + .await, + Err(DeliveryError::MetadataMismatch) + )); + assert_eq!(source.fetches.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn canonical_package_must_match_complete_descriptor_not_only_digests() { + for field in [ + "id", + "name", + "version", + "description", + "api", + "world", + "imports", + "tools", + "connections", + "grants", + "schema_identity", + "schema_version", + "schema_digest", + "schema_namespace", + "schema_dcdl", + ] { + let (artifact, mut d) = fixture(); + match field { + "id" => d.package_id.push('x'), + "name" => d.name.push('x'), + "version" => d.version.push('x'), + "description" => d.description = Some("different".into()), + "api" => d.schema_version += 1, + "world" => d.world.push('x'), + "imports" => d.required_imports.push("yoi:host/logging@2.0.0".into()), + "tools" => d.tools[0].input_schema = serde_json::json!({"type":"string"}), + "connections" => d.connections.push(server_api::PluginConnectionMetadata { + name: "other".into(), + methods: vec![], + path_prefixes: vec![], + }), + "grants" => d + .requested_grants + .push(server_api::PluginGrantMetadata::Logging), + "schema_identity" => d.config_schema.identity.push('x'), + "schema_version" => d.config_schema.version.push('x'), + "schema_digest" => d.config_schema.digest.push('x'), + "schema_namespace" => d.config_schema.namespace.push('x'), + "schema_dcdl" => d.config_schema.dcdl = "{ different: String }".into(), + _ => unreachable!(), + } + let source = Source::new(&artifact, d); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + assert!( + matches!( + source.resolve(&cache).await, + Err(DeliveryError::MetadataMismatch) + ), + "{field}" + ); + assert!(final_files(&cache).is_empty(), "{field}"); + assert!(pending_files(&cache).is_empty(), "{field}"); + } +} + +#[tokio::test] +async fn corrupted_or_wrong_size_cache_file_is_refetched_and_verified() { + for wrong_size in [false, true] { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + source.resolve(&cache).await.unwrap(); + let path = final_files(&cache).pop().unwrap(); + let mut corrupt = artifact.bytes.clone(); + if wrong_size { + corrupt.truncate(7); + } else { + corrupt[0] ^= 1; + } + fs::write(&path, corrupt).unwrap(); + assert_eq!( + source.resolve(&cache).await.unwrap().bytes(), + artifact.bytes + ); + assert_eq!(source.fetches.load(Ordering::SeqCst), 2); + assert_eq!(fs::read(path).unwrap(), artifact.bytes); + } +} + +#[tokio::test] +async fn independent_package_and_artifact_digests_are_both_required() { + for package_digest in [false, true] { + let (artifact, mut descriptor) = fixture(); + if package_digest { + descriptor.package_digest = format!("sha256:{}", "0".repeat(64)); + } + let mut source = Source::new(&artifact, descriptor); + if !package_digest { + source.bytes[100] ^= 1; + } + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::InvalidArtifact(_)) + )); + assert!(final_files(&cache).is_empty()); + } +} + +#[tokio::test] +async fn declared_size_is_checked_before_fetch_and_stream_is_read_only_one_byte_past_bound() { + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let mut limits = CacheLimits::default(); + limits.package.max_package_size_bytes = artifact.artifact_size - 1; + let cache = ArtifactCache::new(directory.path().join("cache"), limits).unwrap(); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::TooLarge) + )); + assert_eq!(source.fetches.load(Ordering::SeqCst), 0); + source.bytes.extend_from_slice(&vec![0; 100_000]); + let cache = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::TooLarge) + )); + assert_eq!( + source.read_bytes.load(Ordering::SeqCst) as u64, + artifact.artifact_size + 1 + ); + assert!(final_files(&cache).is_empty()); + assert!(pending_files(&cache).is_empty()); +} + +#[tokio::test] +async fn interrupted_and_failed_streams_never_publish_and_restart_can_retry() { + for transfer in [Transfer::Short, Transfer::Error] { + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor); + source.transfer = transfer; + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + let error = source.resolve(&cache).await.unwrap_err(); + assert!(matches!( + error, + DeliveryError::InvalidArtifact(_) | DeliveryError::Io(_) + )); + assert!(final_files(&cache).is_empty()); + assert!(pending_files(&cache).is_empty()); + source.transfer = Transfer::Complete; + let restarted = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + assert_eq!( + source.resolve(&restarted).await.unwrap().bytes(), + artifact.bytes + ); + } +} + +#[tokio::test] +async fn cancellation_drops_staging_without_exposing_a_partial_final_file() { + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor); + source.transfer = Transfer::Pause; + let source = Arc::new(source); + let directory = tempfile::tempdir().unwrap(); + let cache = Arc::new(cache(&directory)); + let task = tokio::spawn({ + let source = source.clone(); + let cache = cache.clone(); + async move { source.resolve(&cache).await } + }); + tokio::time::timeout(Duration::from_secs(5), source.paused.acquire()) + .await + .unwrap() + .unwrap() + .forget(); + assert_eq!(pending_files(&cache).len(), 1); + assert!(final_files(&cache).is_empty()); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert!(pending_files(&cache).is_empty()); + let retry = Source::new(&artifact, source.expectation().artifact); + assert_eq!(retry.resolve(&cache).await.unwrap().bytes(), artifact.bytes); +} + +#[tokio::test] +async fn stalled_transfer_deadline_releases_lock_and_staging() { + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor); + source.transfer = Transfer::Pause; + let directory = tempfile::tempdir().unwrap(); + let limits = CacheLimits { + delivery_timeout: Duration::from_millis(100), + ..CacheLimits::default() + }; + let cache = ArtifactCache::new(directory.path().join("cache"), limits).unwrap(); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::Timeout) + )); + assert!(pending_files(&cache).is_empty()); + assert!(final_files(&cache).is_empty()); + source.transfer = Transfer::Complete; + let restarted = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + source.resolve(&restarted).await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn concurrent_instances_converge_on_one_verified_publication() { + let (artifact, descriptor) = fixture(); + let source = Arc::new(Source::new(&artifact, descriptor)); + let directory = tempfile::tempdir().unwrap(); + let first = cache(&directory); + let mut tasks = Vec::new(); + for _ in 0..8 { + let cache = ArtifactCache::new(&first.root, CacheLimits::default()).unwrap(); + let source = source.clone(); + tasks.push(tokio::spawn(async move { source.resolve(&cache).await })); + } + for task in tasks { + assert_eq!(task.await.unwrap().unwrap().bytes(), artifact.bytes); + } + assert_eq!(source.fetches.load(Ordering::SeqCst), 1); + assert_eq!(source.authorizations.load(Ordering::SeqCst), 16); + assert_eq!(final_files(&first).len(), 1); + assert!(pending_files(&first).is_empty()); +} + +#[tokio::test] +async fn restart_discards_stale_partial_even_when_it_contains_a_complete_valid_artifact() { + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + fs::write( + cache.root.join(".pending-crashed-complete"), + &artifact.bytes, + ) + .unwrap(); + fs::write( + cache.root.join(".pending-crashed-prefix"), + &artifact.bytes[..99], + ) + .unwrap(); + let restarted = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + assert_eq!( + source.resolve(&restarted).await.unwrap().bytes(), + artifact.bytes + ); + assert_eq!(source.fetches.load(Ordering::SeqCst), 1); + assert!(pending_files(&cache).is_empty()); +} + +#[tokio::test] +async fn eviction_bounds_bytes_and_entry_count_independently_without_invalidating_returned_bytes() { + let (artifact, descriptor) = fixture(); + for (max_cache_bytes, max_entries) in + [(artifact.artifact_size, 8), (artifact.artifact_size * 8, 1)] + { + let source = Source::new(&artifact, descriptor.clone()); + let directory = tempfile::tempdir().unwrap(); + let limits = CacheLimits { + max_cache_bytes, + max_entries, + ..CacheLimits::default() + }; + let cache = ArtifactCache::new(directory.path().join("cache"), limits).unwrap(); + let held = source.resolve(&cache).await.unwrap(); + let mut other = Source::new(&artifact, descriptor.clone()); + other.authority = "server-b".into(); + other.resolve(&cache).await.unwrap(); + assert_eq!(final_files(&cache).len(), 1); + assert_eq!( + fs::metadata(&final_files(&cache)[0]).unwrap().len(), + artifact.artifact_size + ); + assert_eq!(held.bytes(), artifact.bytes); + source.resolve(&cache).await.unwrap(); + assert_eq!(source.fetches.load(Ordering::SeqCst), 2); + assert_eq!(final_files(&cache).len(), 1); + } +} + +#[tokio::test] +async fn hashed_authority_separates_servers_while_fresh_workspace_and_worker_owners_are_required() { + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor.clone()); + source.authority = "../../outside/opaque:server".into(); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + source.resolve(&cache).await.unwrap(); + let mut other = Source::new(&artifact, descriptor.clone()); + other.authority = "server-b".into(); + other.resolve(&cache).await.unwrap(); + assert_eq!(final_files(&cache).len(), 2); + for path in final_files(&cache) { + assert_eq!(path.parent().unwrap(), cache.root); + assert!(is_cache_filename( + path.file_name().unwrap().to_str().unwrap() + )); + assert!( + !path + .file_name() + .unwrap() + .to_string_lossy() + .contains("outside") + ); + } + let mut workspace_b = Source::new(&artifact, descriptor.clone()); + workspace_b.authority = source.authority.clone(); + workspace_b.metadata.lock().unwrap().workspace_id = "workspace-b".into(); + let request = workspace_b.request(); + let expected = workspace_b.expectation(); + assert_eq!( + cache + .resolve(&workspace_b, "workspace-b", &request, &expected) + .await + .unwrap() + .bytes(), + artifact.bytes + ); + assert_eq!(workspace_b.fetches.load(Ordering::SeqCst), 0); + assert_eq!(workspace_b.authorizations.load(Ordering::SeqCst), 2); + let mut different_worker_owner = Source::new(&artifact, descriptor); + different_worker_owner.authority = source.authority; + different_worker_owner.deny_on.store(1, Ordering::SeqCst); + assert!(matches!( + different_worker_owner.resolve(&cache).await, + Err(DeliveryError::Denied) + )); +} + +#[cfg(unix)] +#[tokio::test] +async fn private_root_and_non_symlink_cache_entries_are_enforced() { + use std::os::unix::fs::{PermissionsExt, symlink}; + let directory = tempfile::tempdir().unwrap(); + let public = directory.path().join("public"); + fs::create_dir(&public).unwrap(); + fs::set_permissions(&public, fs::Permissions::from_mode(0o755)).unwrap(); + assert!(matches!( + ArtifactCache::new(&public, CacheLimits::default()), + Err(DeliveryError::InvalidCacheRoot) + )); + let link = directory.path().join("link"); + symlink(&public, &link).unwrap(); + assert!(matches!( + ArtifactCache::new(&link, CacheLimits::default()), + Err(DeliveryError::InvalidCacheRoot) + )); + let cache = cache(&directory); + assert_eq!( + fs::metadata(&cache.root).unwrap().permissions().mode() & 0o777, + 0o700 + ); + let sentinel = directory.path().join("sentinel"); + fs::write(&sentinel, b"do not change").unwrap(); + symlink(&sentinel, cache.root.join(".pending-symlink")).unwrap(); + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + assert!(matches!( + source.resolve(&cache).await, + Err(DeliveryError::InvalidCacheRoot) + )); + assert_eq!(fs::read(sentinel).unwrap(), b"do not change"); +} + +// Deliberately corrupt the shared package's API without breaking ZIP integrity, +// so rejection must come from the common manifest verifier, not a bad checksum. +fn unsupported_api_bytes(bytes: &mut [u8]) { + let header = bytes + .windows(4) + .position(|window| window == b"PK\x03\x04") + .unwrap(); + let mut offset = header; + let mut crc = None; + while &bytes[offset..offset + 4] == b"PK\x03\x04" { + let size = u32::from_le_bytes(bytes[offset + 18..offset + 22].try_into().unwrap()) as usize; + let name_size = + u16::from_le_bytes(bytes[offset + 26..offset + 28].try_into().unwrap()) as usize; + let extra = + u16::from_le_bytes(bytes[offset + 28..offset + 30].try_into().unwrap()) as usize; + let start = offset + 30 + name_size + extra; + if &bytes[offset + 30..offset + 30 + name_size] == b"plugin.toml" { + let payload = &mut bytes[start..start + size]; + let index = payload + .windows(b"schema_version = 2".len()) + .position(|window| window == b"schema_version = 2") + .unwrap(); + payload[index + b"schema_version = ".len()] = b'9'; + let mut checksum = !0_u32; + for byte in payload { + checksum ^= u32::from(*byte); + for _ in 0..8 { + checksum = (checksum >> 1) ^ (0xedb88320 & 0_u32.wrapping_sub(checksum & 1)); + } + } + let checksum = (!checksum).to_le_bytes(); + bytes[offset + 14..offset + 18].copy_from_slice(&checksum); + crc = Some(checksum); + } + offset = start + size; + } + while &bytes[offset..offset + 4] == b"PK\x01\x02" { + let name_size = + u16::from_le_bytes(bytes[offset + 28..offset + 30].try_into().unwrap()) as usize; + let extra = + u16::from_le_bytes(bytes[offset + 30..offset + 32].try_into().unwrap()) as usize; + let comment = + u16::from_le_bytes(bytes[offset + 32..offset + 34].try_into().unwrap()) as usize; + if &bytes[offset + 46..offset + 46 + name_size] == b"plugin.toml" { + bytes[offset + 16..offset + 20].copy_from_slice(&crc.unwrap()); + } + offset += 46 + name_size + extra + comment; + } +} + +#[tokio::test] +async fn unsupported_api_is_rejected_by_common_verifier_before_publication() { + let (mut artifact, mut descriptor) = fixture(); + unsupported_api_bytes(&mut artifact.bytes); + artifact.artifact_digest = format!("sha256:{}", hash(&artifact.bytes)); + descriptor.artifact_digest = artifact.artifact_digest.clone(); + descriptor.schema_version = 9; + let source = Source::new(&artifact, descriptor); + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + let error = source.resolve(&cache).await.unwrap_err(); + assert!( + matches!(error, DeliveryError::InvalidArtifact(ref message) if message.contains("unsupported schema")), + "{error:?}" + ); + assert!(final_files(&cache).is_empty()); +} + +struct Actor { + child: Child, + lines: std::sync::mpsc::Receiver>, + stdout_thread: Option>, + stderr_thread: Option>, + text: String, +} +impl Drop for Actor { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + if let Some(thread) = self.stdout_thread.take() { + let _ = thread.join(); + } + if let Some(thread) = self.stderr_thread.take() { + let _ = thread.join(); + } + } +} +impl Actor { + fn spawn(root: &Path, action: &str) -> Self { + let mut child = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "delivery::tests::process_cache_actor", + "--ignored", + "--nocapture", + "--test-threads=1", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + serde_json::to_writer(child.stdin.take().unwrap(), &(root, action)).unwrap(); + let stdout = child.stdout.take().unwrap(); + let mut stderr = child.stderr.take().unwrap(); + let (sender, lines) = std::sync::mpsc::channel(); + let stdout_thread = std::thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + if sender.send(Some(line.unwrap())).is_err() { + return; + } + } + let _ = sender.send(None); + }); + let stderr_thread = std::thread::spawn(move || { + let mut text = String::new(); + stderr.read_to_string(&mut text).unwrap(); + text + }); + Self { + child, + lines, + stdout_thread: Some(stdout_thread), + stderr_thread: Some(stderr_thread), + text: String::new(), + } + } + + fn line(&mut self) -> Option { + let line = self + .lines + .recv_timeout(Duration::from_secs(10)) + .unwrap_or_else(|error| panic!("child output deadline: {error}: {}", self.text)); + if let Some(line) = &line { + self.text.push_str(line); + self.text.push('\n'); + } + line + } + + fn ready(&mut self) { + loop { + let line = self.line().expect("child exited before readiness"); + if line.contains("cache-actor-ready") { + break; + } + } + } + + fn finish(&mut self) -> (std::process::ExitStatus, String) { + while self.line().is_some() {} + let status = self.child.wait().unwrap(); + self.text + .push_str(&self.stderr_thread.take().unwrap().join().unwrap()); + (status, std::mem::take(&mut self.text)) + } +} + +#[test] +#[ignore = "subprocess actor invoked by ordinary cache integration tests"] +fn process_cache_actor() { + let (root, action): (PathBuf, String) = serde_json::from_reader(io::stdin()).unwrap(); + let (artifact, descriptor) = fixture(); + let mut source = Source::new(&artifact, descriptor); + if action == "crash" { + source.transfer = Transfer::Crash; + } + let cache = ArtifactCache::new(root, CacheLimits::default()).unwrap(); + if action == "resolve" { + let probe = OpenOptions::new() + .read(true) + .write(true) + .open(cache.root.join(".cache.lock")) + .unwrap(); + assert!(!fs4::fs_std::FileExt::try_lock_exclusive(&probe).unwrap()); + } + println!("cache-actor-ready"); + io::stdout().flush().unwrap(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let verified = runtime.block_on(source.resolve(&cache)).unwrap(); + assert_eq!(verified.bytes(), artifact.bytes); + println!( + "cache-actor-fetches={}", + source.fetches.load(Ordering::SeqCst) + ); +} + +#[tokio::test] +async fn independent_processes_converge_after_contending_for_same_cache_lock() { + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + let lock = lock_cache(cache.root.join(".cache.lock")).await.unwrap(); + let mut first = Actor::spawn(&cache.root, "resolve"); + let mut second = Actor::spawn(&cache.root, "resolve"); + first.ready(); + second.ready(); + drop(lock); + let (a, a_text) = first.finish(); + let (b, b_text) = second.finish(); + assert!(a.success(), "{a_text}"); + assert!(b.success(), "{b_text}"); + assert_eq!( + format!("{a_text}{b_text}") + .matches("cache-actor-fetches=1") + .count(), + 1 + ); + assert_eq!( + format!("{a_text}{b_text}") + .matches("cache-actor-fetches=0") + .count(), + 1 + ); + assert_eq!(final_files(&cache).len(), 1); + assert!(pending_files(&cache).is_empty()); +} + +#[tokio::test] +async fn process_exit_before_publish_leaves_no_final_bytes_and_restart_discards_partial() { + let directory = tempfile::tempdir().unwrap(); + let cache = cache(&directory); + let mut actor = Actor::spawn(&cache.root, "crash"); + actor.ready(); + let (status, text) = actor.finish(); + assert_eq!(status.code(), Some(73), "{text}"); + assert!(final_files(&cache).is_empty()); + assert_eq!(pending_files(&cache).len(), 1); + let (artifact, descriptor) = fixture(); + let source = Source::new(&artifact, descriptor); + let restarted = ArtifactCache::new(&cache.root, CacheLimits::default()).unwrap(); + assert_eq!( + source.resolve(&restarted).await.unwrap().bytes(), + artifact.bytes + ); + assert!(pending_files(&cache).is_empty()); + assert_eq!(final_files(&cache).len(), 1); +} diff --git a/crates/plugin-runtime/src/lib.rs b/crates/plugin-runtime/src/lib.rs index d70bb5138..54a506cce 100644 --- a/crates/plugin-runtime/src/lib.rs +++ b/crates/plugin-runtime/src/lib.rs @@ -10,6 +10,7 @@ pub mod bindings { }); } +pub mod delivery; mod executor; pub mod host; pub mod owner; diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 4bf66dd01..f6580c3b9 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -224,6 +224,8 @@ impl_openapi_schema!( PluginInstallation, PluginListResponse, PluginOperationResponse, + PluginArtifactDeliveryRequest, + PluginArtifactDeliveryMetadata, SubjectPermissionsResponse, SubjectPermissionsMutationResponse, SubjectPermissionInterruption, @@ -3014,6 +3016,22 @@ pub trait ServerApi { #[path] operation_id: String, ) -> Result; + #[post("/api/w/{workspace_id}/workers/self/plugin-artifacts/authorize", status = 200, error_status = 400, additional_error_statuses = [401,403,404,409,413,503], openapi = false, normalize_body_errors = true)] + async fn plugin_artifact_authorize( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: PluginArtifactDeliveryRequest, + ) -> Result; + + #[post("/api/w/{workspace_id}/workers/self/plugin-artifacts/pull", status = 200, error_status = 400, additional_error_statuses = [401,403,404,409,413,503], openapi = false, normalize_body_errors = true)] + async fn plugin_artifact_pull( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: PluginArtifactDeliveryRequest, + ) -> Result; + // Ordinary operations accept member browser sessions or trusted Worker request // context. The service resolves Worker grants; no operation accepts an actor or // grant ID from the caller. Grant management is Workspace-owner-only. @@ -14943,6 +14961,8 @@ mod openapi_artifact_tests { // identity, or a one-use Reviewer capability. They remain generated Rust client/Axum // operations, but must not appear as unauthenticated operations in the public OpenAPI. const SIGNED_INTERNAL: &[&str] = &[ + "plugin_artifact_authorize", + "plugin_artifact_pull", "operation_authorize", "bound_operation_authorize", "current_worker_workspace_config_get", diff --git a/crates/server-api/src/plugin.rs b/crates/server-api/src/plugin.rs index 7fdf34c6f..418021947 100644 --- a/crates/server-api/src/plugin.rs +++ b/crates/server-api/src/plugin.rs @@ -182,3 +182,37 @@ pub struct PluginOperationResponse { pub kind: PluginOperationKind, pub installation: PluginInstallation, } + +/// Namespace a byte cache by the authenticated Server origin and signing key. +/// This is not a package hash or an authorization decision. Hosts derive the +/// bundle from their current registered authority, never from a delivery reply. +pub fn plugin_server_authority_identity(bundle: &crate::WorkspacePublicIdentityBundle) -> String { + serde_json::to_string(&( + bundle.backend_url.trim_end_matches('/'), + &bundle.public_key_fingerprint, + )) + .expect("string tuple is serializable") +} + +/// A selection within a persisted plan, never digest-only retrieval authority. +/// Workspace and Worker identities come from authenticated Host context. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct PluginArtifactDeliveryRequest { + pub plan_id: String, + pub plan_fingerprint: String, + pub installation_id: String, + pub artifact_digest: String, +} + +/// Fresh owner observation for one exact selection. Cache only bytes, not this +/// observation or the Workspace's configuration, grants or installation state. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct PluginArtifactDeliveryMetadata { + pub workspace_id: String, + pub request: PluginArtifactDeliveryRequest, + pub artifact: PluginArtifact, +} diff --git a/crates/server-api/src/plugin_types.rs b/crates/server-api/src/plugin_types.rs index a8983cf0b..775b4ac52 100644 --- a/crates/server-api/src/plugin_types.rs +++ b/crates/server-api/src/plugin_types.rs @@ -46,6 +46,8 @@ pub fn plugin_api_typescript() -> String { PluginListResponse::decl(&config), PluginOperationKind::decl(&config), PluginOperationResponse::decl(&config), + PluginArtifactDeliveryRequest::decl(&config), + PluginArtifactDeliveryMetadata::decl(&config), ]; format!( "// Generated from server-api. Do not edit by hand.\n// Regenerate: cargo run -q -p server-api --features typescript --example generate_plugin_api_types > web/workspace/src/lib/generated/plugin-api.ts\n\n{}\n", @@ -74,6 +76,7 @@ mod tests { assert!(text.contains("\"signed\", \"value\": string")); assert!(text.contains("\"unsigned\", \"value\": string")); } + use api_macros::ApiContract; #[test] fn plugin_routes_use_binary_upload_and_closed_management_contracts() { let document = crate::canonical_openapi_document().unwrap(); @@ -98,6 +101,56 @@ mod tests { assert!(value["paths"]["/api/w/{workspace_id}/plugins/{installation_id}/data"].is_null()); } + #[test] + fn artifact_delivery_is_runtime_signed_and_has_no_caller_authority_or_transport_inputs() { + for operation in ["plugin_artifact_authorize", "plugin_artifact_pull"] { + assert!( + crate::ServerApiMetadata::OPERATIONS + .iter() + .any(|entry| entry.operation_id == operation) + ); + } + let document = crate::canonical_openapi_document().unwrap(); + let value: serde_json::Value = serde_json::from_str(&document.to_json().unwrap()).unwrap(); + for suffix in ["authorize", "pull"] { + assert!( + value["paths"] + [format!("/api/w/{{workspace_id}}/workers/self/plugin-artifacts/{suffix}")] + .is_null() + ); + } + let selection = serde_json::json!({ + "plan_id":"plan", "plan_fingerprint":"fingerprint", + "installation_id":"installation", "artifact_digest":"digest" + }); + assert!( + serde_json::from_value::(selection.clone()) + .is_ok() + ); + for field in [ + "workspace_id", + "worker_id", + "principal_id", + "delegation_id", + "url", + "token", + "cache_path", + "bytes", + ] { + let mut invalid = selection.clone(); + invalid[field] = "caller-selected".into(); + assert!( + serde_json::from_value::(invalid).is_err() + ); + } + assert!( + serde_json::from_value::( + serde_json::json!({"artifact_digest":"digest"}) + ) + .is_err() + ); + } + #[cfg(feature = "typescript")] #[test] fn generated_plugin_types_are_current() { diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 8465242b9..89acce8bc 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -36,6 +36,7 @@ decodal.workspace = true flow = { path = "../flow" } fs4 = { workspace = true, optional = true } manifest.workspace = true +yoi-plugin-runtime = { path = "../plugin-runtime" } libc.workspace = true protocol.workspace = true runtime-api.workspace = true @@ -43,11 +44,12 @@ serde = { workspace = true, features = ["derive"] } session-store.workspace = true sha2.workspace = true serde_json.workspace = true -reqwest = { version = "0.13", optional = true, default-features = false, features = ["json", "rustls"] } +reqwest = { version = "0.13", optional = true, default-features = false, features = ["json", "rustls", "stream"] } ring.workspace = true tar.workspace = true thiserror = { workspace = true } tokio = { workspace = true, features = ["net", "process", "rt", "signal", "sync", "time"] } +tokio-util = { version = "0.7", features = ["io"] } tracing.workspace = true tracing-subscriber.workspace = true toml.workspace = true diff --git a/crates/worker-runtime/src/workspace_request.rs b/crates/worker-runtime/src/workspace_request.rs index f008640cd..4b55e457c 100644 --- a/crates/worker-runtime/src/workspace_request.rs +++ b/crates/worker-runtime/src/workspace_request.rs @@ -18,6 +18,8 @@ use crate::auth::{ RuntimeRequestSourceSigner, }; +pub mod plugin_delivery; + const DEFAULT_REQUEST_PROOF_TTL_SECONDS: u64 = 60; const RUNTIME_ID_HEADER: &str = "x-yoi-runtime-id"; const WORKER_ID_HEADER: &str = "x-yoi-worker-id"; diff --git a/crates/worker-runtime/src/workspace_request/plugin_delivery.rs b/crates/worker-runtime/src/workspace_request/plugin_delivery.rs new file mode 100644 index 000000000..29cc6cc36 --- /dev/null +++ b/crates/worker-runtime/src/workspace_request/plugin_delivery.rs @@ -0,0 +1,182 @@ +//! Pulls only from the Runtime's current registered Server. Credentials, URLs +//! and transport diagnostics stay in the Host, outside plan and model inputs. +//! Registration is reread for every operation, including cached-byte admission; +//! a captured registration is not a perpetual lease after trust revocation. +use super::*; +use crate::workspace_issuer::{WorkspaceIssuerTrustRecord, WorkspaceIssuerTrustState}; +use server_api::{PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest}; +use std::sync::Arc; +use yoi_plugin_runtime::delivery::{ArtifactSource, ArtifactStream, DeliveryError}; + +pub type CurrentArtifactServer = + dyn Fn() -> Result + Send + Sync; + +pub struct RemotePluginArtifactSource { + client: RuntimeWorkspaceRequestClient, + worker_id: String, + authority: String, + current_server: Arc, +} + +impl RemotePluginArtifactSource { + /// `current_server` reads Runtime-owned registration, never a model-provided + /// bundle or credentials. The request client must already have a Runtime key + /// and the registered Backend URL as its signing audience. + pub fn new( + client: RuntimeWorkspaceRequestClient, + worker_id: String, + current_server: Arc, + ) -> Result { + if worker_id.is_empty() { + return Err(DeliveryError::InvalidRequest); + } + let registration = current_server()?; + let authority = server_api::plugin_server_authority_identity(®istration.public_bundle()); + let source = Self { + client, + worker_id, + authority, + current_server, + }; + source.registration()?; + Ok(source) + } + + fn registration(&self) -> Result<(), DeliveryError> { + let registration = (self.current_server)()?; + if registration.state != WorkspaceIssuerTrustState::Active + || registration.workspace_id != self.client.workspace_id + || registration.backend_url.trim_end_matches('/') != self.client.base_url + || self.client.audience().map(|a| a.trim_end_matches('/')) + != Some(self.client.base_url.as_str()) + || server_api::plugin_server_authority_identity(®istration.public_bundle()) + != self.authority + { + return Err(DeliveryError::Denied); + } + Ok(()) + } + + fn request( + &self, + suffix: &str, + selection: &PluginArtifactDeliveryRequest, + ) -> Result { + self.registration()?; + let mut url = reqwest::Url::parse(&self.client.base_url) + .map_err(|_| DeliveryError::InvalidRequest)?; + url.path_segments_mut() + .map_err(|_| DeliveryError::InvalidRequest)? + .pop_if_empty() + .extend([ + "api", + "w", + self.client.workspace_id(), + "workers", + "self", + "plugin-artifacts", + suffix, + ]); + let path_and_query = url.path().to_owned(); + Ok(RuntimeWorkspaceRequest { + method: reqwest::Method::POST, + path_and_query, + body: serde_json::to_vec(selection).map_err(|_| DeliveryError::InvalidRequest)?, + headers: HeaderMap::from_iter([( + reqwest::header::CONTENT_TYPE, + reqwest::header::HeaderValue::from_static("application/json"), + )]), + permission: crate::auth::WORKSPACE_REQUEST_PERMISSION.into(), + worker_id: Some(self.worker_id.clone()), + timeout: Some(Duration::from_secs(30)), + max_response_bytes: manifest::plugin::PluginPackageLimits::default() + .max_package_size_bytes as usize, + }) + } + + async fn response( + &self, + request: RuntimeWorkspaceRequest, + ) -> Result { + let prepared = self.client.prepare(&request).map_err(|_| unavailable())?; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .no_proxy() + .timeout(Duration::from_secs(30)) + .build() + .map_err(|_| unavailable())?; + let response = client + .request(request.method, prepared.url) + .headers(request.headers) + .header(RUNTIME_ID_HEADER, &self.client.runtime_id) + .header(WORKER_ID_HEADER, &self.worker_id) + .header( + RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + prepared.proof.ok_or(DeliveryError::Denied)?, + ) + .body(request.body) + .send() + .await + .map_err(|_| unavailable())?; + if !response.status().is_success() { + return Err(if response.status().is_server_error() { + unavailable() + } else { + DeliveryError::Denied + }); + } + if response + .content_length() + .is_some_and(|size| size > request.max_response_bytes as u64) + { + return Err(DeliveryError::TooLarge); + } + self.registration()?; + Ok(response) + } +} + +fn unavailable() -> DeliveryError { + DeliveryError::Unavailable("registered owner unavailable".into()) +} + +#[async_trait::async_trait] +impl ArtifactSource for RemotePluginArtifactSource { + fn server_authority_identity(&self) -> &str { + &self.authority + } + + async fn authorize( + &self, + selection: &PluginArtifactDeliveryRequest, + ) -> Result { + let request = self.request("authorize", selection)?; + let response = self.response(request).await?; + // Static schema metadata is also bounded; never use response.json() on an + // unbounded peer response or reflect its body into diagnostics. + let mut bytes = Vec::new(); + let mut chunks = response.bytes_stream(); + while let Some(chunk) = chunks.next().await { + let chunk = chunk.map_err(|_| unavailable())?; + if bytes.len().saturating_add(chunk.len()) > 1024 * 1024 { + return Err(DeliveryError::TooLarge); + } + bytes.extend_from_slice(&chunk); + } + self.registration()?; + serde_json::from_slice(&bytes).map_err(|_| DeliveryError::MetadataMismatch) + } + + async fn fetch( + &self, + selection: &PluginArtifactDeliveryRequest, + ) -> Result { + let response = self.response(self.request("pull", selection)?).await?; + let chunks = response.bytes_stream().map(|chunk| { + chunk.map_err(|_| std::io::Error::other("artifact transport interrupted")) + }); + Ok(Box::pin(tokio_util::io::StreamReader::new(Box::pin( + chunks, + )))) + } +} diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 28de7544b..fd9096bc4 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -26,6 +26,7 @@ workspace-drive.workspace = true subjektiv.workspace = true flow = { path = "../flow" } manifest.workspace = true +yoi-plugin-runtime = { path = "../plugin-runtime" } job.workspace = true protocol = { workspace = true } runtime-api.workspace = true diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 54813cfcc..33e2a9190 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -16,6 +16,7 @@ pub mod plugin_binding; pub mod plugin_catalog; pub mod plugin_host; pub mod plugin_https; +pub mod plugin_delivery; pub mod plugin_plan; pub mod plugin_receipt; pub mod plugin_settings; diff --git a/crates/workspace-server/src/plugin_catalog.rs b/crates/workspace-server/src/plugin_catalog.rs index ff7c211b4..13646c015 100644 --- a/crates/workspace-server/src/plugin_catalog.rs +++ b/crates/workspace-server/src/plugin_catalog.rs @@ -170,7 +170,7 @@ impl Drop for PendingFile { let _ = fs::remove_file(&self.0); } } -fn read_blob(root: &Path, artifact: &str, size: u64) -> Result> { +pub(crate) fn read_blob(root: &Path, artifact: &str, size: u64) -> Result> { let path = blob_path(root, artifact)?; let metadata = storage(fs::symlink_metadata(&path))?; if !metadata.is_file() @@ -222,7 +222,7 @@ fn artifact_metadata( canonical: &plugin::CanonicalPluginArtifact, ) -> Result { let m = &package.package.manifest; - let schema_digest = format!("sha256:{}", hash(m.config_schema.as_bytes())); + let schema = plugin::plugin_config_schema_identity(m); Ok(PluginArtifact { package_id: m.id.clone(), name: m.name.clone(), @@ -250,10 +250,10 @@ fn artifact_metadata( connections: decode(&encode(&m.connections)?)?, requested_grants: decode(&encode(&m.requested_grants)?)?, config_schema: PluginSchemaContribution { - identity: format!("{}/config", m.id), - version: m.version.clone(), - namespace: format!("{}/config/{}/{}", m.id, m.version, schema_digest), - digest: schema_digest, + identity: schema.identity, + version: schema.version, + namespace: schema.namespace, + digest: schema.digest, dcdl: m.config_schema.clone(), }, revoked: false, diff --git a/crates/workspace-server/src/plugin_delivery.rs b/crates/workspace-server/src/plugin_delivery.rs new file mode 100644 index 000000000..006c619f4 --- /dev/null +++ b/crates/workspace-server/src/plugin_delivery.rs @@ -0,0 +1,146 @@ +//! Delivery admission is bound to an authenticated Host identity and a persisted +//! exact plan pin, not management permission or possession of a digest. Service +//! adapters must derive the identity and recheck live Host authority on every +//! call, including cache hits; these database checks are additional constraints. +//! +//! Reads never change plan retention. Revoked and released plans remain audit +//! evidence, but cannot authorize metadata or bytes. Bytes are read while holding +//! the catalog's SQLite write fence, preventing revoke/uninstall/GC interleaving. +use crate::{ + Result, + plugin_catalog::{self, PluginCatalogError}, + plugin_plan::{self, HostExecutionIdentity, PlanPinState}, + store::SqliteWorkspaceStore, +}; +use rusqlite::{Transaction, TransactionBehavior, params}; +use server_api::{PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest}; +use std::path::Path; + +/// Promote fresh descriptor metadata to a cache expectation only after binding +/// every immutable identity to the saved Host pin. This does not admit use; the +/// cache must still ask the current owner on each resolution. +pub fn artifact_expectation( + pin: &crate::plugin_plan::WorkerPluginPin, + artifact: &server_api::PluginArtifact, +) -> Result { + if !pin.matches_artifact(artifact)? { + return Err(PluginCatalogError::Conflict.into()); + } + Ok(yoi_plugin_runtime::delivery::ArtifactExpectation::new( + artifact.clone(), + )) +} + +/// The caller must hold an immediate transaction in the plan/catalog owner DB. +/// The current catalog operation is read inside this snapshot: unrelated version +/// additions must not invalidate the immutable pin's creation-time observation. +/// Current config is deliberately not resolved again, avoiding latest selection. +pub fn plugin_delivery_authorize( + tx: &Transaction<'_>, + workspace: &str, + execution: &HostExecutionIdentity, + request: &PluginArtifactDeliveryRequest, +) -> Result { + let saved = plugin_plan::plugin_plan_load(tx, workspace, &request.plan_id)?; + if saved.fingerprint != request.plan_fingerprint || saved.plan.execution() != execution { + return Err(PluginCatalogError::Conflict.into()); + } + let kind = match saved.pin_state { + PlanPinState::Pending => "pending", + PlanPinState::Active => "active", + PlanPinState::Restorable => "restorable", + PlanPinState::Released => return Err(PluginCatalogError::Conflict.into()), + }; + let pin = saved + .plan + .plugins() + .iter() + .find(|pin| { + pin.installation_id() == request.installation_id + && pin.artifact_digest() == request.artifact_digest + }) + .ok_or(PluginCatalogError::NotFound)?; + let retained: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM plugin_references WHERE workspace_id=?1 AND kind=?2 AND owner=?3 AND installation_id=?4 AND artifact_digest=?5)", + params![workspace, kind, request.plan_id, request.installation_id, request.artifact_digest], + |row| row.get(0), + )?; + if !retained { + return Err(PluginCatalogError::Conflict.into()); + } + let operation: String = tx.query_row( + "SELECT last_operation FROM plugin_installations WHERE workspace_id=?1 AND installation_id=?2", + params![workspace, request.installation_id], + |row| row.get(0), + )?; + let artifact = plugin_catalog::plugin_artifact_admit( + tx, + workspace, + &request.installation_id, + &operation, + pin.package_digest(), + &request.artifact_digest, + )?; + // Reconstruct using the *saved* projection and operation observation. This + // compares all immutable identities, size, Component world/imports and schema + // identity while retaining the original config/grants rather than live config. + if !pin.matches_artifact(&artifact)? { + return Err(PluginCatalogError::Conflict.into()); + } + Ok(PluginArtifactDeliveryMetadata { + workspace_id: workspace.into(), + request: request.clone(), + artifact, + }) +} + +/// Reauthorizes in the same owner transaction as the bounded immutable blob read. +/// A previous metadata response (including a cache descriptor) is not authority. +pub fn plugin_delivery_bytes( + tx: &Transaction<'_>, + root: &Path, + workspace: &str, + execution: &HostExecutionIdentity, + request: &PluginArtifactDeliveryRequest, +) -> Result> { + let admitted = plugin_delivery_authorize(tx, workspace, execution, request)?; + plugin_catalog::read_blob( + root, + &admitted.artifact.artifact_digest, + u64::from(admitted.artifact.artifact_size), + ) +} + +impl SqliteWorkspaceStore { + pub fn plugin_delivery_authorize( + &self, + workspace: &str, + execution: &HostExecutionIdentity, + request: &PluginArtifactDeliveryRequest, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let metadata = plugin_delivery_authorize(&tx, workspace, execution, request)?; + tx.commit()?; + Ok(metadata) + }) + } + + pub fn plugin_delivery_bytes( + &self, + root: &Path, + workspace: &str, + execution: &HostExecutionIdentity, + request: &PluginArtifactDeliveryRequest, + ) -> Result> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let bytes = plugin_delivery_bytes(&tx, root, workspace, execution, request)?; + tx.commit()?; + Ok(bytes) + }) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/workspace-server/src/plugin_delivery/tests.rs b/crates/workspace-server/src/plugin_delivery/tests.rs new file mode 100644 index 000000000..53b4fdf83 --- /dev/null +++ b/crates/workspace-server/src/plugin_delivery/tests.rs @@ -0,0 +1,591 @@ +//! Owner-DB integration: production upload/install, schema-checked config commit, +//! plan resolution/creation and file-backed SQLite. Host authentication/transport +//! are outside this suite; the supplied Host identity is the adapter's boundary. +use super::*; +use crate::{ + Error, + plugin_plan::{PluginPlanCreateReceipt, WorkerPluginExecutionPlan}, + store::{ControlPlaneStore, WorkspaceRecord}, +}; +use manifest::plugin; +use rusqlite::Connection; +use server_api::*; +use std::fs; +use tempfile::{TempDir, tempdir}; + +struct Fixture { + dir: TempDir, + store: SqliteWorkspaceStore, +} +impl Fixture { + async fn new() -> Self { + let dir = tempdir().unwrap(); + let store = SqliteWorkspaceStore::open_for_test(dir.path().join("server.db")).unwrap(); + for workspace in ["a", "b"] { + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: workspace.into(), + owner_account_id: "owner-account".into(), + display_name: workspace.into(), + state: "active".into(), + created_at: "2026-10-28T00:00:00Z".into(), + updated_at: "2026-10-28T00:00:00Z".into(), + }) + .await + .unwrap(); + } + Self { dir, store } + } + fn root(&self) -> std::path::PathBuf { + self.dir.path().join("artifacts") + } + fn install(&self, alias: &str) -> PluginOperationResponse { + let upload = self + .store + .plugin_upload(&self.root(), "a", "actor", &package("0.1.0")) + .unwrap(); + self.store + .plugin_install( + "a", + "actor", + &PluginInstallRequest { + upload_id: upload.upload_id, + alias: alias.into(), + operation_key: alias.into(), + }, + ) + .unwrap() + } + fn create(&self, installed: &PluginOperationResponse, key: &str) -> PluginPlanCreateReceipt { + let plan = self.configure(&[installed]); + self.store.plugin_plan_create("actor", key, &plan).unwrap() + } + fn authorize( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> Result { + self.store + .plugin_delivery_authorize("a", &execution(), request) + } + fn bytes(&self, request: &PluginArtifactDeliveryRequest) -> Result> { + self.store + .plugin_delivery_bytes(&self.root(), "a", &execution(), request) + } + fn retained(&self, plan_id: &str) -> Vec { + self.store.with_conn(|conn| Ok(conn.prepare("SELECT kind FROM plugin_references WHERE workspace_id='a' AND owner=?1 ORDER BY kind")? + .query_map([plan_id], |r| r.get(0))?.collect::>()?)).unwrap() + } + fn revoke(&self, installed: &PluginOperationResponse) { + self.store + .plugin_revoke( + "a", + "actor", + &installed.installation.installation_id, + &PluginRevokeRequest { + expected_last_operation: installed.operation_id.clone(), + operation_key: "revoke".into(), + artifact_digest: installed.installation.artifacts[0].artifact_digest.clone(), + reason: "owner revoked".into(), + }, + ) + .unwrap(); + } +} +fn execution() -> HostExecutionIdentity { + HostExecutionIdentity { + runtime_id: "runtime".into(), + worker_id: "worker".into(), + principal_id: "worker-principal".into(), + delegation_id: Some("delegation".into()), + } +} +fn request(saved: &PluginPlanCreateReceipt) -> PluginArtifactDeliveryRequest { + let pin = &saved.plan.plugins()[0]; + PluginArtifactDeliveryRequest { + plan_id: saved.plan_id.clone(), + plan_fingerprint: saved.fingerprint.clone(), + installation_id: pin.installation_id().into(), + artifact_digest: pin.artifact_digest().into(), + } +} +fn error(result: Result, expected: PluginCatalogError) { + assert!( + matches!(result, Err(Error::PluginCatalog(actual)) if actual == expected), + "expected {expected:?}, got {result:?}" + ); +} +fn package(version: &str) -> Vec { + let dir = tempdir().unwrap(); + fs::write( + dir.path().join("plugin.toml"), + include_str!("../../../../resources/plugin/templates/rust-component-tool/plugin.toml") + .replace("version = \"0.1.0\"", &format!("version = \"{version}\"")), + ) + .unwrap(); + fs::write( + dir.path().join("plugin.component.wasm"), + include_bytes!("../../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"), + ) + .unwrap(); + let package = plugin::read_plugin_directory( + dir.path(), + plugin::PluginSourceKind::Explicit, + &plugin::PluginPackageLimits::default(), + ) + .unwrap(); + plugin::pack_plugin_package(&package, &plugin::PluginPackageLimits::default()) + .unwrap() + .bytes +} +impl Fixture { + fn configure(&self, installed: &[&PluginOperationResponse]) -> WorkerPluginExecutionPlan { + use crate::config_source::{ConfigCommitRequest, WorkspaceConfigSchemaProvider}; + use config_source::{ConfigTreeChange, VirtualPath, WorkspaceConfigSchemaBundle}; + let bundle = WorkspaceConfigSchemaBundle::compose([ + crate::plugin_settings::PluginConfigSchemaProvider + .contribution() + .unwrap(), + crate::profile_settings::ProfileConfigSchemaProvider + .contribution() + .unwrap(), + ]) + .unwrap(); + let state = self + .store + .ensure_workspace_config_materialized_with_schema("a", "1", bundle.clone()) + .unwrap(); + let mut enabled = serde_json::Map::new(); + let mut selections = Vec::new(); + for (index, item) in installed.iter().enumerate() { + let a = &item.installation.artifacts[0]; + enabled.insert( + format!("plugin_{index}"), + serde_json::json!({ + "installation_id": item.installation.installation_id, + "expected_last_operation": item.operation_id, + "package_digest": a.package_digest, "artifact_digest": a.artifact_digest, + "tools": ["example_echo"], "config": {}, + }), + ); + selections.push(serde_json::json!({"installation_id":item.installation.installation_id,"tools":["example_echo"]})); + } + let value = serde_json::json!({"plugins":{"enabled": enabled}, "profile":{"entries":[{"selector":"project:tool","profile":{"plugins":{"selections":selections}}}]}}); + let path = VirtualPath::parse("main.dcdl").unwrap(); + let candidate = self + .store + .evaluate_workspace_config_candidate_with_schema( + "a", + &ConfigCommitRequest { + base_digest: state.snapshot.digest.clone(), + entrypoints: vec![path.clone()], + changes: vec![ConfigTreeChange::Update { + path: path.clone(), + expected_digest: state.snapshot.get(&path).unwrap().content_digest.clone(), + content: format!( + "{} as WorkspaceConfigSchema", + crate::plugin_settings::render_value(&value).unwrap() + ), + }], + }, + bundle, + ) + .unwrap(); + let state = self + .store + .commit_evaluated_workspace_config("a", &candidate) + .unwrap(); + self.store + .with_conn(|conn| { + let tx = conn.unchecked_transaction()?; + crate::plugin_settings::resolve_worker_plugin_plan( + &tx, + "a", + &state.snapshot.digest, + "project:tool", + execution(), + ) + }) + .unwrap() + } + fn disable(&self) { + self.configure(&[]); + } +} + +#[tokio::test] +async fn pending_active_restorable_reads_preserve_exact_pin_across_restart() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + let original = package("0.1.0"); + for (state, kind) in [ + (PlanPinState::Pending, "pending"), + (PlanPinState::Active, "active"), + (PlanPinState::Restorable, "restorable"), + ] { + let before = f.retained(&saved.plan_id); + assert_eq!(before, [kind]); + assert_eq!( + f.authorize(&request).unwrap(), + PluginArtifactDeliveryMetadata { + workspace_id: "a".into(), + request: request.clone(), + artifact: installed.installation.artifacts[0].clone(), + } + ); + assert_eq!(f.bytes(&request).unwrap(), original); + assert_eq!(f.retained(&saved.plan_id), before); + assert_eq!( + f.store + .plugin_plan_load("a", &saved.plan_id) + .unwrap() + .pin_state, + state + ); + let reopened = SqliteWorkspaceStore::open_for_test(f.dir.path().join("server.db")).unwrap(); + assert_eq!( + reopened + .plugin_delivery_bytes(&f.root(), "a", &execution(), &request) + .unwrap(), + original + ); + if state != PlanPinState::Restorable { + f.store + .plugin_plan_transition( + "a", + &saved.plan_id, + state, + if state == PlanPinState::Pending { + PlanPinState::Active + } else { + PlanPinState::Restorable + }, + ) + .unwrap(); + } + } +} + +#[tokio::test] +async fn foreign_workspace_host_identity_and_plan_coordinates_cannot_fetch_bytes() { + let f = Fixture::new().await; + let installed = f.install("one"); + let other = f.install("other"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + error( + f.store + .plugin_delivery_authorize("b", &execution(), &request), + PluginCatalogError::NotFound, + ); + error( + f.store + .plugin_delivery_bytes(&f.root(), "b", &execution(), &request), + PluginCatalogError::NotFound, + ); + let mut identities = Vec::new(); + let mut wrong = execution(); + wrong.runtime_id = "other-runtime".into(); + identities.push(wrong); + let mut wrong = execution(); + wrong.worker_id = "other-worker".into(); + identities.push(wrong); + let mut wrong = execution(); + wrong.principal_id = "other-principal".into(); + identities.push(wrong); + let mut wrong = execution(); + wrong.delegation_id = Some("other-delegation".into()); + identities.push(wrong); + let mut wrong = execution(); + wrong.delegation_id = None; + identities.push(wrong); + for wrong in identities { + error( + f.store.plugin_delivery_authorize("a", &wrong, &request), + PluginCatalogError::Conflict, + ); + error( + f.store + .plugin_delivery_bytes(&f.root(), "a", &wrong, &request), + PluginCatalogError::Conflict, + ); + } + let mut cases = Vec::new(); + let mut wrong = request.clone(); + wrong.plan_id = "missing-plan".into(); + cases.push((wrong, PluginCatalogError::NotFound)); + let mut wrong = request.clone(); + wrong.plan_fingerprint = format!("sha256:{}", "0".repeat(64)); + cases.push((wrong, PluginCatalogError::Conflict)); + let mut wrong = request.clone(); + wrong.installation_id = other.installation.installation_id.clone(); + cases.push((wrong, PluginCatalogError::NotFound)); + let mut wrong = request.clone(); + wrong.artifact_digest = format!("sha256:{}", "0".repeat(64)); + cases.push((wrong, PluginCatalogError::NotFound)); + // Another real persisted plan cannot lend its fingerprint to this plan. + let other_plan = f.create(&other, "other-launch"); + let mut wrong = request.clone(); + wrong.plan_id = other_plan.plan_id; + cases.push((wrong, PluginCatalogError::Conflict)); + for (wrong, expected) in cases { + error(f.authorize(&wrong), expected); + error(f.bytes(&wrong), expected); + } + assert_eq!(f.retained(&saved.plan_id), ["pending"]); +} + +#[tokio::test] +async fn revoked_artifact_denies_new_metadata_and_bytes_without_releasing_audit_pin() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + // Previously obtained metadata cannot stand in for current admission. + let metadata = f.authorize(&request).unwrap(); + f.revoke(&installed); + error(f.authorize(&metadata.request), PluginCatalogError::Conflict); + error(f.bytes(&metadata.request), PluginCatalogError::Conflict); + let stored = f.store.plugin_plan_load("a", &saved.plan_id).unwrap(); + assert_eq!(stored.plan, saved.plan); + assert_eq!(stored.pin_state, PlanPinState::Pending); + assert_eq!(f.retained(&saved.plan_id), ["pending"]); +} + +#[tokio::test] +async fn released_plan_and_missing_or_wrong_kind_reference_never_authorize_delivery() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + // Retention corruption is injected after valid owner creation, not by + // constructing or inserting a fake serialized plan. + f.store + .with_conn(|conn| { + conn.execute( + "UPDATE plugin_references SET kind='active' WHERE workspace_id='a' AND owner=?1", + [&saved.plan_id], + )?; + Ok(()) + }) + .unwrap(); + error(f.authorize(&request), PluginCatalogError::Conflict); + error(f.bytes(&request), PluginCatalogError::Conflict); + f.store + .with_conn(|conn| { + conn.execute( + "DELETE FROM plugin_references WHERE workspace_id='a' AND owner=?1", + [&saved.plan_id], + )?; + Ok(()) + }) + .unwrap(); + error(f.authorize(&request), PluginCatalogError::Conflict); + error(f.bytes(&request), PluginCatalogError::Conflict); + let another = f.create(&installed, "second-launch"); + let another_request = self::request(&another); + f.store + .plugin_plan_transition( + "a", + &another.plan_id, + PlanPinState::Pending, + PlanPinState::Released, + ) + .unwrap(); + error(f.authorize(&another_request), PluginCatalogError::Conflict); + error(f.bytes(&another_request), PluginCatalogError::Conflict); + assert!(f.retained(&another.plan_id).is_empty()); + assert_eq!( + f.store + .plugin_plan_load("a", &another.plan_id) + .unwrap() + .plan, + another.plan + ); +} + +#[tokio::test] +async fn new_catalog_version_and_current_config_change_do_not_reselect_saved_pin() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + let original = f.bytes(&request).unwrap(); + let upload = f + .store + .plugin_upload(&f.root(), "a", "actor", &package("0.2.0")) + .unwrap(); + let added = f + .store + .plugin_add_artifact( + "a", + "actor", + &installed.installation.installation_id, + &PluginAddArtifactRequest { + upload_id: upload.upload_id, + expected_last_operation: installed.operation_id.clone(), + operation_key: "add-version".into(), + }, + ) + .unwrap(); + assert_ne!(added.operation_id, installed.operation_id); + f.disable(); + assert_eq!( + f.authorize(&request).unwrap().artifact, + installed.installation.artifacts[0] + ); + assert_eq!(f.bytes(&request).unwrap(), original); + assert_eq!(f.retained(&saved.plan_id), ["pending"]); + let newer = added + .installation + .artifacts + .iter() + .find(|a| a.version == "0.2.0") + .unwrap(); + let mut outside = request.clone(); + outside.artifact_digest = newer.artifact_digest.clone(); + error(f.authorize(&outside), PluginCatalogError::NotFound); + error(f.bytes(&outside), PluginCatalogError::NotFound); +} + +#[tokio::test] +async fn release_then_uninstall_keeps_plan_for_audit_but_denies_delivery() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + f.store + .plugin_plan_transition( + "a", + &saved.plan_id, + PlanPinState::Pending, + PlanPinState::Released, + ) + .unwrap(); + f.disable(); + f.store + .plugin_uninstall( + "a", + "actor", + &installed.installation.installation_id, + &PluginUninstallRequest { + expected_last_operation: installed.operation_id.clone(), + operation_key: "uninstall".into(), + }, + ) + .unwrap(); + error(f.authorize(&request), PluginCatalogError::Conflict); + error(f.bytes(&request), PluginCatalogError::Conflict); + assert_eq!( + f.store.plugin_plan_load("a", &saved.plan_id).unwrap().plan, + saved.plan + ); +} + +#[tokio::test] +async fn artifact_metadata_identity_corruption_cannot_override_persisted_pin() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + let original = serde_json::to_value(&installed.installation.artifacts[0]).unwrap(); + let mut corruptions = Vec::new(); + for (name, value) in [ + ( + "artifact_digest", + serde_json::json!(format!("sha256:{}", "0".repeat(64))), + ), + ( + "package_digest", + serde_json::json!(format!("sha256:{}", "0".repeat(64))), + ), + ("package_id", serde_json::json!("other-package")), + ("version", serde_json::json!("9.0.0")), + ("artifact_size", serde_json::json!(1)), + ("schema_version", serde_json::json!(999)), + ("world", serde_json::json!("unsupported-world")), + ("required_imports", serde_json::json!(["other-import"])), + ] { + let mut value_ = original.clone(); + value_[name] = value; + corruptions.push(value_); + } + for name in ["identity", "version", "namespace", "digest"] { + let mut changed = original.clone(); + changed["config_schema"][name] = if name == "digest" { + serde_json::json!(format!("sha256:{}", "0".repeat(64))) + } else { + serde_json::json!("other-schema") + }; + corruptions.push(changed); + } + for metadata in corruptions { + f.store.with_conn(|conn| { + conn.execute("UPDATE plugin_artifacts SET metadata_json=?1 WHERE workspace_id='a' AND installation_id=?2 AND artifact_digest=?3", params![metadata.to_string(), request.installation_id, request.artifact_digest])?; + Ok(()) + }).unwrap(); + assert!( + f.authorize(&request).is_err(), + "admitted changed {metadata}" + ); + assert!(f.bytes(&request).is_err(), "fetched changed {metadata}"); + } + assert_eq!(f.retained(&saved.plan_id), ["pending"]); +} + +#[tokio::test] +async fn bounded_blob_read_rejects_missing_truncated_oversize_and_digest_mismatch() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + let path = f + .root() + .join("blobs") + .join(request.artifact_digest.strip_prefix("sha256:").unwrap()); + let original = fs::read(&path).unwrap(); + fs::remove_file(&path).unwrap(); + error(f.bytes(&request), PluginCatalogError::Unavailable); + for bad in [vec![], vec![0; original.len() + 1], vec![0; original.len()]] { + fs::write(&path, bad).unwrap(); + error(f.bytes(&request), PluginCatalogError::Unavailable); + } + fs::write(&path, &original).unwrap(); + assert_eq!(f.bytes(&request).unwrap(), original); + assert_eq!(f.retained(&saved.plan_id), ["pending"]); +} + +#[tokio::test] +async fn owner_transaction_fences_concurrent_catalog_mutation_until_bytes_are_read() { + let f = Fixture::new().await; + let installed = f.install("one"); + let saved = f.create(&installed, "launch"); + let request = request(&saved); + let mut other = Connection::open(f.dir.path().join("server.db")).unwrap(); + other.busy_timeout(std::time::Duration::ZERO).unwrap(); + f.store + .with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + plugin_delivery_authorize(&tx, "a", &execution(), &request)?; + assert!( + other + .transaction_with_behavior(TransactionBehavior::Immediate) + .is_err() + ); + assert_eq!( + plugin_delivery_bytes(&tx, &f.root(), "a", &execution(), &request)?, + package("0.1.0") + ); + tx.commit()?; + Ok(()) + }) + .unwrap(); + other + .transaction_with_behavior(TransactionBehavior::Immediate) + .unwrap() + .rollback() + .unwrap(); + f.revoke(&installed); + error(f.bytes(&request), PluginCatalogError::Conflict); +} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 175fd01bf..a502d7b62 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -50,6 +50,9 @@ mod drive_grants; mod plugin; #[path = "server_plugin_host.rs"] mod plugin_host; +#[path = "server_plugin_delivery.rs"] +mod plugin_delivery; +pub use plugin_delivery::InProcessPluginArtifactSource; #[path = "server_workspace_config.rs"] mod workspace_config; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -7653,6 +7656,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou )) .merge(plugin::router(service.clone())) .merge(plugin_host::router(service.clone())) + .merge(plugin_delivery::router(service.clone())) .merge(server_api::server_api_axum::current_worker_workspace_config_get(service.clone())) .merge(server_api::server_api_axum::current_worker_workspace_config_attach(service.clone())) .merge( @@ -13511,6 +13515,27 @@ impl server_api::ServerApi for ServerApiContractService { plugin_host::execute(Arc::new(api.clone()), context, workspace_id, request).await } + async fn plugin_artifact_authorize( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::PluginArtifactDeliveryRequest, + ) -> std::result::Result< + server_api::PluginArtifactDeliveryMetadata, + server_api::RepositoryApiError, + > { + plugin_delivery::authorize(self.workspace_api()?, context, workspace_id, request).await + } + + async fn plugin_artifact_pull( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::PluginArtifactDeliveryRequest, + ) -> std::result::Result { + plugin_delivery::pull(self.workspace_api()?, context, workspace_id, request).await + } + async fn plugin_upload( &self, actor: RequestActor, @@ -38663,6 +38688,10 @@ mod tests { include!("server_ticket_case_signed_tests.rs"); } + mod plugin_delivery_tests { + include!("server_plugin_delivery_tests.rs"); + } + mod plugin_tests { include!("server_plugin_tests.rs"); } diff --git a/crates/workspace-server/src/server_plugin.rs b/crates/workspace-server/src/server_plugin.rs index 40be94954..e70a6ea6e 100644 --- a/crates/workspace-server/src/server_plugin.rs +++ b/crates/workspace-server/src/server_plugin.rs @@ -24,7 +24,7 @@ pub(super) fn router(service: Arc) -> Router { .layer(middleware::from_fn(normalize_rejections)) } -async fn normalize_rejections(request: Request, next: Next) -> Response { +pub(super) async fn normalize_rejections(request: Request, next: Next) -> Response { let response = next.run(request).await; // The management contract classifies malformed DTOs/queries as invalid=400, // including Axum's JSON data/media rejections. Preserve the common envelope @@ -40,7 +40,7 @@ async fn normalize_rejections(request: Request, next: Next) -> Response { Json(RepositoryApiError::new( 400, "plugin_invalid", - "Invalid Plugin catalog request", + "Invalid Plugin request", vec![], )), ) diff --git a/crates/workspace-server/src/server_plugin_delivery.rs b/crates/workspace-server/src/server_plugin_delivery.rs new file mode 100644 index 000000000..b8622a214 --- /dev/null +++ b/crates/workspace-server/src/server_plugin_delivery.rs @@ -0,0 +1,215 @@ +//! Authenticated byte delivery and the in-process adapter share current Worker +//! authority and the persisted plan owner. Neither a cache hit nor an adapter +//! handle retains permission. No destination policy or credential reaches a guest. +use super::*; +use server_api::{PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest}; +use yoi_plugin_runtime::delivery::{ArtifactSource, ArtifactStream, DeliveryError}; + +pub(super) fn router(service: Arc) -> Router { + Router::new() + .merge(server_api::server_api_axum::plugin_artifact_authorize( + service.clone(), + )) + .merge(server_api::server_api_axum::plugin_artifact_pull(service)) + .layer(middleware::from_fn(plugin::normalize_rejections)) +} + +fn failure(error: impl Into) -> server_api::RepositoryApiError { + let error = error.into(); + let status = match &error.error { + Error::PluginCatalog(error) => error.status(), + _ => api_error_status(&error.error).as_u16(), + }; + let (status, code) = match status { + 400 => (400, "plugin_delivery_invalid"), + 401 | 403 => (403, "plugin_delivery_denied"), + 404 => (404, "plugin_delivery_not_found"), + 409 => (409, "plugin_delivery_conflict"), + 413 => (413, "plugin_delivery_limit"), + _ => (503, "plugin_delivery_unavailable"), + }; + server_api::RepositoryApiError::new(status, code, "Plugin artifact delivery refused", vec![]) +} + +fn delivery_error(error: server_api::RepositoryApiError) -> DeliveryError { + if api_macros::HttpError::status_code(&error) >= 500 { + DeliveryError::Unavailable("owner unavailable".into()) + } else { + DeliveryError::Denied + } +} + +async fn worker_from_context( + api: &WorkspaceApi, + workspace: &str, + context: &server_api::ServerRequestContext, +) -> std::result::Result { + validate_workspace_scope(api, workspace).map_err(failure)?; + let headers = current_worker_contract_headers(context)?; + authenticate_worker_mutation_source_owned(api, workspace, &headers) + .await + .map_err(failure) +} + +impl WorkspaceApi { + /// A Host-only adapter bound to this Server and Worker, not a loopback client. + /// Construction resolves the active Server key; every operation checks it + /// again, together with Worker membership, live grants, lineage and pin state. + pub async fn plugin_artifact_source( + &self, + worker: RuntimeWorkerRef, + ) -> crate::Result { + let authority = authority_identity(self)?; + self.plugin_execution_identity(&worker).await?; + Ok(InProcessPluginArtifactSource { + api: self.clone(), + worker, + authority, + }) + } +} + +fn authority_identity(api: &WorkspaceApi) -> crate::Result { + let identity = api + .store + .get_workspace_signing_identity(api.workspace_id())? + .ok_or_else(|| { + Error::WorkspacePermissionDenied("Plugin Server identity unavailable".into()) + })?; + let bundle = project_workspace_signing_identity(api, identity)? + .public_bundle + .ok_or_else(|| { + Error::WorkspacePermissionDenied("Plugin Server identity inactive".into()) + })?; + Ok(server_api::plugin_server_authority_identity(&bundle)) +} + +async fn current_execution( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + request: &PluginArtifactDeliveryRequest, +) -> std::result::Result +{ + let execution = api + .plugin_execution_identity(worker) + .await + .map_err(failure)?; + let stage_api = api.clone(); + let stage_worker = worker.clone(); + let installation_id = request.installation_id.clone(); + api.host_executors + .host() + .map_err(failure)? + .run(move || { + subject_authorization::authorize( + &stage_api, + &stage_worker, + &server_api::authorization::OperationAuthorizationRequest { + operation: "plugin.artifact.read".into(), + resource: server_api::authorization::PermissionResource::PluginInstallation { + workspace_id: stage_api.workspace_id().into(), + installation_id, + }, + }, + ) + .map_err(|error| error.error) + }) + .await + .map_err(failure)?; + Ok(execution) +} + +async fn metadata_for_worker( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + request: PluginArtifactDeliveryRequest, +) -> std::result::Result { + let execution = current_execution(api, worker, &request).await?; + let store = api.config_store.clone(); + let workspace = api.workspace_id().to_owned(); + store + .db_executor() + .run(move || store.plugin_delivery_authorize(&workspace, &execution, &request)) + .await + .map_err(failure) +} + +async fn bytes_for_worker( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + request: PluginArtifactDeliveryRequest, +) -> std::result::Result, server_api::RepositoryApiError> { + let execution = current_execution(api, worker, &request).await?; + let store = api.config_store.clone(); + let workspace = api.workspace_id().to_owned(); + let root = + ServerConfig::server_data_root_for_data_dir(&api.config.data_root).join("plugin-artifacts"); + let before_read = request.clone(); + let bytes = store + .db_executor() + .run(move || store.plugin_delivery_bytes(&root, &workspace, &execution, &request)) + .await + .map_err(failure)?; + // The storage read is not publication authority. Recheck security narrowing + // and the exact persisted pin after I/O; Runtime also checks before returning. + metadata_for_worker(api, worker, before_read).await?; + Ok(bytes) +} + +pub(super) async fn authorize( + api: &WorkspaceApi, + context: server_api::ServerRequestContext, + workspace: String, + request: PluginArtifactDeliveryRequest, +) -> std::result::Result { + let worker = worker_from_context(api, &workspace, &context).await?; + metadata_for_worker(api, &worker, request).await +} + +pub(super) async fn pull( + api: &WorkspaceApi, + context: server_api::ServerRequestContext, + workspace: String, + request: PluginArtifactDeliveryRequest, +) -> std::result::Result { + let worker = worker_from_context(api, &workspace, &context).await?; + bytes_for_worker(api, &worker, request) + .await + .map(server_api::BinaryBody::from) +} + +pub struct InProcessPluginArtifactSource { + api: WorkspaceApi, + worker: RuntimeWorkerRef, + authority: String, +} + +#[async_trait::async_trait] +impl ArtifactSource for InProcessPluginArtifactSource { + fn server_authority_identity(&self) -> &str { + &self.authority + } + + async fn authorize( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> std::result::Result { + if authority_identity(&self.api).map_err(|_| DeliveryError::Denied)? != self.authority { + return Err(DeliveryError::Denied); + } + metadata_for_worker(&self.api, &self.worker, request.clone()) + .await + .map_err(delivery_error) + } + + async fn fetch( + &self, + request: &PluginArtifactDeliveryRequest, + ) -> std::result::Result { + self.authorize(request).await?; + let bytes = bytes_for_worker(&self.api, &self.worker, request.clone()) + .await + .map_err(delivery_error)?; + Ok(Box::pin(std::io::Cursor::new(bytes))) + } +} diff --git a/crates/workspace-server/src/server_plugin_delivery_tests.rs b/crates/workspace-server/src/server_plugin_delivery_tests.rs new file mode 100644 index 000000000..a522baaf4 --- /dev/null +++ b/crates/workspace-server/src/server_plugin_delivery_tests.rs @@ -0,0 +1,1153 @@ +// Server library integration tests (`plugin_delivery_tests` filter). +// Real generated HTTP routes, Runtime signatures/trust/JTI verification, Subject +// policy storage, catalog files, config commits and persisted plans are connected. +// Worker execution uses WorkdirlessFixtureRuntime; the interrupted-pull case +// injects a body-stream failure only after the real owner succeeds. No model or +// HTTPS destination is contacted. Each fixture owns its loopback listener and DB. +use super::*; +use crate::plugin_plan::{HostExecutionIdentity, PlanPinState, PluginPlanCreateReceipt}; +use server_api::{ + PluginArtifactDeliveryMetadata, PluginArtifactDeliveryRequest, PluginInstallRequest, + PluginOperationResponse, PluginRevokeRequest, + authorization::{PermissionResource, ScopedPermission, SubjectPermissionsUpdate}, +}; +use std::sync::Mutex; +use tokio::io::AsyncReadExt; +use worker_runtime::{ + workspace_issuer::{WorkspaceIssuerTrustRecord, WorkspaceIssuerTrustState}, + workspace_request::{ + RuntimeWorkspaceRequestClient, plugin_delivery::RemotePluginArtifactSource, + }, +}; +use yoi_plugin_runtime::delivery::{ + ArtifactCache, ArtifactExpectation, ArtifactSource, CacheLimits, DeliveryError, +}; + +struct LoopbackServer(Option>); +impl LoopbackServer { + async fn stop(&mut self) { + if let Some(task) = self.0.take() { + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + } + } +} +impl Drop for LoopbackServer { + fn drop(&mut self) { + if let Some(task) = &self.0 { + task.abort(); + } + } +} + +struct DeliveryFixture { + api: WorkspaceApi, + identity: RuntimeIdentityMaterial, + worker: RuntimeWorkerRef, + subject: Option, + installed: PluginOperationResponse, + saved: PluginPlanCreateReceipt, + request: PluginArtifactDeliveryRequest, + bytes: Vec, + registration: Arc>, + remote: RemotePluginArtifactSource, + interrupt_pull: Arc, + server: LoopbackServer, + url: String, + dir: tempfile::TempDir, +} + +fn artifact_package() -> Vec { + let dir = tempfile::tempdir().unwrap(); + fs::write( + dir.path().join("plugin.toml"), + include_str!("../../../resources/plugin/templates/rust-component-tool/plugin.toml"), + ) + .unwrap(); + fs::write( + dir.path().join("plugin.component.wasm"), + include_bytes!("../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"), + ) + .unwrap(); + let limits = manifest::plugin::PluginPackageLimits::default(); + let package = manifest::plugin::read_plugin_directory( + dir.path(), + manifest::plugin::PluginSourceKind::Explicit, + &limits, + ) + .unwrap(); + manifest::plugin::pack_plugin_package(&package, &limits) + .unwrap() + .bytes +} + +async fn delivery_worker(api: &WorkspaceApi, subject: Option<&str>) -> RuntimeWorkerRef { + let Json(worker) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(), + display_name: "Artifact consumer".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: subject.map(|subject| server_api::SubjektivWorkerConnectionRequest { + subject_id: subject.into(), + }), + }, + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap(); + RuntimeWorkerRef::new(worker.runtime_id, worker.worker_id) +} + +fn configure_delivery_plan( + api: &WorkspaceApi, + installed: &PluginOperationResponse, + execution: HostExecutionIdentity, + key: &str, +) -> PluginPlanCreateReceipt { + use config_source::{ConfigTreeChange, VirtualPath}; + let bundle = api.config_schema_registry.compose().unwrap(); + let state = api + .config_store + .ensure_workspace_config_materialized_with_schema(TEST_WORKSPACE_ID, "1", bundle.clone()) + .unwrap(); + let artifact = &installed.installation.artifacts[0]; + let value = json!({ + "plugins":{"enabled":{"delivery":{ + "installation_id": installed.installation.installation_id, + "expected_last_operation": installed.operation_id, + "package_digest": artifact.package_digest, + "artifact_digest": artifact.artifact_digest, + "tools":["example_echo"], "config":{} + }}}, + "profile":{"entries":[{"selector":"project:delivery","profile":{ + "plugins":{"selections":[{"installation_id":installed.installation.installation_id,"tools":["example_echo"]}]} + }}]} + }); + let path = VirtualPath::parse("main.dcdl").unwrap(); + let candidate = api + .config_store + .evaluate_workspace_config_candidate_with_schema( + TEST_WORKSPACE_ID, + &ConfigCommitRequest { + base_digest: state.snapshot.digest.clone(), + entrypoints: vec![path.clone()], + changes: vec![ConfigTreeChange::Update { + path: path.clone(), + expected_digest: state.snapshot.get(&path).unwrap().content_digest.clone(), + content: format!( + "{} as WorkspaceConfigSchema", + crate::plugin_settings::render_value(&value).unwrap() + ), + }], + }, + bundle, + ) + .unwrap(); + let committed = api + .config_store + .commit_evaluated_workspace_config(TEST_WORKSPACE_ID, &candidate) + .unwrap(); + let plan = api + .config_store + .with_conn(|conn| { + let tx = conn.unchecked_transaction()?; + crate::plugin_settings::resolve_worker_plugin_plan( + &tx, + TEST_WORKSPACE_ID, + &committed.snapshot.digest, + "project:delivery", + execution, + ) + }) + .unwrap(); + api.config_store + .plugin_plan_create("delivery-host", key, &plan) + .unwrap() +} + +impl DeliveryFixture { + async fn new(subject_bound: bool) -> Self { + let dir = tempfile::tempdir().unwrap(); + let mut api = test_api(dir.path()).await; + register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; + api.runtime + .register_or_replace(WorkdirlessFixtureRuntime::default()); + let subject = subject_bound.then(|| { + open_subjektiv_store(&api) + .unwrap() + .create_subject(crate::subjektiv::SubjectRole::new("artifact-consumer").unwrap()) + .unwrap() + .id + }); + let worker = delivery_worker(&api, subject.as_deref()).await; + let identity = RuntimeIdentityMaterial::generate(&worker.runtime_id).unwrap(); + configure_runtime_request_auth(&mut api, &identity, &worker.runtime_id); + seed_worker_source_member(&api, &worker.runtime_id, &worker.worker_id); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + api.config.backend_base_url = Some(url.clone()); + let signing = api + .signing_identities + .provision_existing(TEST_WORKSPACE_ID, &test_owner_actor().account_id) + .unwrap(); + let bundle = project_workspace_signing_identity(&api, signing) + .unwrap() + .public_bundle + .unwrap(); + let (_, registered) = worker_runtime::workspace_issuer::add_workspace_issuer_trust( + &mut Vec::new(), + bundle, + worker_runtime::auth::unix_now_seconds() as i64, + ) + .unwrap(); + let registration = Arc::new(Mutex::new(registered)); + let bytes = artifact_package(); + let root = ServerConfig::server_data_root_for_data_dir(&api.config.data_root) + .join("plugin-artifacts"); + let uploaded = api + .config_store + .plugin_upload(&root, TEST_WORKSPACE_ID, "delivery-host", &bytes) + .unwrap(); + let installed = api + .config_store + .plugin_install( + TEST_WORKSPACE_ID, + "delivery-host", + &PluginInstallRequest { + upload_id: uploaded.upload_id, + alias: "delivery".into(), + operation_key: "install-delivery".into(), + }, + ) + .unwrap(); + let execution = api.plugin_execution_identity(&worker).await.unwrap(); + let saved = configure_delivery_plan(&api, &installed, execution, "delivery-launch"); + let pin = &saved.plan.plugins()[0]; + let request = PluginArtifactDeliveryRequest { + plan_id: saved.plan_id.clone(), + plan_fingerprint: saved.fingerprint.clone(), + installation_id: pin.installation_id().into(), + artifact_digest: pin.artifact_digest().into(), + }; + let client = + RuntimeWorkspaceRequestClient::new(TEST_WORKSPACE_ID, &url, &worker.runtime_id) + .with_runtime_request_source(&identity, &url); + let current = registration.clone(); + let remote = RemotePluginArtifactSource::new( + client, + worker.worker_id.clone(), + Arc::new(move || Ok(current.lock().unwrap().clone())), + ) + .unwrap(); + let interrupt_pull = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let fault = interrupt_pull.clone(); + // Fault injection is outside the authenticated generated routes and owner: + // successful pull must complete those checks before its HTTP stream fails. + let app = build_router(api.clone()).layer(axum::middleware::from_fn( + move |request: Request, next: axum::middleware::Next| { + let fault = fault.clone(); + async move { + let pull = request.uri().path().ends_with("/plugin-artifacts/pull"); + let response = next.run(request).await; + if !pull + || !response.status().is_success() + || !fault.load(std::sync::atomic::Ordering::SeqCst) + { + return response; + } + let (mut parts, body) = response.into_parts(); + let bytes = axum::body::to_bytes(body, 16 * 1024 * 1024).await.unwrap(); + parts.headers.remove(axum::http::header::CONTENT_LENGTH); + let prefix = bytes.slice(..bytes.len() / 2); + let stream = + futures::stream::once(async move { Ok::<_, std::io::Error>(prefix) }) + .chain(futures::stream::once(async { + tokio::task::yield_now().await; + Err::(std::io::Error::other( + "injected delivery disconnect", + )) + })); + Response::from_parts(parts, Body::from_stream(stream)) + } + }, + )); + let server = LoopbackServer(Some(tokio::spawn(async move { + axum::serve(listener, app).await.unwrap() + }))); + Self { + api, + identity, + worker, + subject, + installed, + saved, + request, + bytes, + registration, + remote, + interrupt_pull, + server, + url, + dir, + } + } + fn expected(&self) -> ArtifactExpectation { + ArtifactExpectation::new(self.installed.installation.artifacts[0].clone()) + } + fn cache(&self, name: &str) -> ArtifactCache { + ArtifactCache::new(self.dir.path().join(name), CacheLimits::default()).unwrap() + } + fn path(&self, suffix: &str) -> String { + format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/plugin-artifacts/{suffix}") + } + fn proof( + &self, + workspace: &str, + worker: &str, + method: &str, + path: &str, + body: &[u8], + ) -> String { + worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&self.identity) + .issue( + &self.url, + workspace, + Some(worker), + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + method, + path, + body, + worker_runtime::auth::unix_now_seconds() as i64, + 30, + ) + .unwrap() + } + async fn post(&self, path: &str, body: Vec, proof: Option<&str>) -> reqwest::Response { + self.post_as(&self.worker.worker_id, path, body, proof) + .await + } + async fn post_as( + &self, + worker_id: &str, + path: &str, + body: Vec, + proof: Option<&str>, + ) -> reqwest::Response { + let client = reqwest::Client::builder() + .no_proxy() + .timeout(std::time::Duration::from_secs(10)) + .build() + .unwrap(); + let mut request = client + .post(format!("{}{path}", self.url)) + .header(CONTENT_TYPE.as_str(), "application/json") + .header("x-yoi-runtime-id", &self.worker.runtime_id) + .header("x-yoi-worker-id", worker_id) + .body(body); + if let Some(proof) = proof { + request = request.header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ); + } + request.send().await.unwrap() + } + async fn signed_post(&self, suffix: &str, value: &impl serde::Serialize) -> reqwest::Response { + let path = self.path(suffix); + let body = serde_json::to_vec(value).unwrap(); + let proof = self.proof( + TEST_WORKSPACE_ID, + &self.worker.worker_id, + "POST", + &path, + &body, + ); + self.post(&path, body, Some(&proof)).await + } + fn artifact_permission(&self) -> ScopedPermission { + ScopedPermission { + operation: "plugin.artifact.read".into(), + resource: PermissionResource::PluginInstallation { + workspace_id: TEST_WORKSPACE_ID.into(), + installation_id: self.request.installation_id.clone(), + }, + } + } + fn policy(&self, execution: Vec, delegation: Vec) { + let subject = self.subject.as_ref().unwrap(); + let store = open_subjektiv_store(&self.api).unwrap(); + let previous = store.subject_permissions(subject).unwrap(); + store + .set_subject_permissions( + subject, + &SubjectPermissionsUpdate { + operation_id: uuid::Uuid::now_v7().to_string(), + expected_change_id: previous.map(|p| p.change_id), + execution, + delegation, + }, + ) + .unwrap(); + } + fn assert_pending_pin(&self) { + let stored = self + .api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, &self.saved.plan_id) + .unwrap(); + assert_eq!(stored.plan, self.saved.plan); + assert_eq!(stored.pin_state, PlanPinState::Pending); + let references: Vec = self + .api + .config_store + .with_conn(|conn| { + Ok(conn + .prepare( + "SELECT kind FROM plugin_references WHERE workspace_id=?1 AND owner=?2", + )? + .query_map( + rusqlite::params![TEST_WORKSPACE_ID, self.saved.plan_id], + |row| row.get(0), + )? + .collect::>()?) + }) + .unwrap(); + assert_eq!(references, ["pending"]); + } +} + +async fn source_bytes( + source: &dyn ArtifactSource, + request: &PluginArtifactDeliveryRequest, +) -> Vec { + let mut bytes = vec![]; + source + .fetch(request) + .await + .unwrap() + .read_to_end(&mut bytes) + .await + .unwrap(); + bytes +} + +#[tokio::test] +async fn same_persisted_plan_remote_and_inprocess_return_exact_archive_and_digests() { + use sha2::{Digest, Sha256}; + let f = DeliveryFixture::new(false).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + assert_eq!( + f.remote.server_authority_identity(), + local.server_authority_identity() + ); + let remote_metadata = f.remote.authorize(&f.request).await.unwrap(); + assert_eq!(remote_metadata, local.authorize(&f.request).await.unwrap()); + assert_eq!(remote_metadata.workspace_id, TEST_WORKSPACE_ID); + assert_eq!(remote_metadata.request, f.request); + assert_eq!( + remote_metadata.artifact, + f.installed.installation.artifacts[0] + ); + let remote = source_bytes(&f.remote, &f.request).await; + assert_eq!(remote, source_bytes(&local, &f.request).await); + assert_eq!(remote, f.bytes); + assert_eq!( + format!("sha256:{}", Sha256::digest(&remote).iter().map(|byte| format!("{byte:02x}")).collect::()), + f.request.artifact_digest + ); + let unpacked = manifest::plugin::read_plugin_package_bytes( + &remote, + manifest::plugin::PluginSourceKind::Explicit, + &manifest::plugin::PluginPackageLimits::default(), + ) + .unwrap(); + assert_eq!( + unpacked.package.package_digest, + remote_metadata.artifact.package_digest + ); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert_eq!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + } + f.assert_pending_pin(); +} + +#[tokio::test] +async fn signed_foreign_plan_pin_and_digest_only_requests_never_deliver_bytes() { + let f = DeliveryFixture::new(false).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let root = ServerConfig::server_data_root_for_data_dir(&f.api.config.data_root) + .join("plugin-artifacts"); + let uploaded = f + .api + .config_store + .plugin_upload(&root, TEST_WORKSPACE_ID, "delivery-host", &f.bytes) + .unwrap(); + let outside = f + .api + .config_store + .plugin_install( + TEST_WORKSPACE_ID, + "delivery-host", + &PluginInstallRequest { + upload_id: uploaded.upload_id, + alias: "outside-plan".into(), + operation_key: "install-outside-plan".into(), + }, + ) + .unwrap(); + let other_plan = configure_delivery_plan( + &f.api, + &outside, + f.api.plugin_execution_identity(&f.worker).await.unwrap(), + "outside-plan-launch", + ); + let mut cases = vec![]; + let mut wrong = f.request.clone(); + wrong.plan_id = other_plan.plan_id; + cases.push((wrong, 409)); + let mut wrong = f.request.clone(); + wrong.plan_id = "missing-plan".into(); + cases.push((wrong, 404)); + let mut wrong = f.request.clone(); + wrong.plan_fingerprint = format!("sha256:{}", "0".repeat(64)); + cases.push((wrong, 409)); + let mut wrong = f.request.clone(); + wrong.installation_id = outside.installation.installation_id; + cases.push((wrong, 404)); + let mut wrong = f.request.clone(); + wrong.artifact_digest = format!("sha256:{}", "0".repeat(64)); + cases.push((wrong, 404)); + for (request, status) in cases { + for suffix in ["authorize", "pull"] { + let response = f.signed_post(suffix, &request).await; + assert_eq!(response.status().as_u16(), status, "{suffix}: {request:?}"); + let error: server_api::RepositoryApiError = response.json().await.unwrap(); + assert!(error.error.starts_with("plugin_delivery_")); + } + assert!(matches!( + f.remote.authorize(&request).await, + Err(DeliveryError::Denied) + )); + assert!(local.fetch(&request).await.is_err()); + } + for suffix in ["authorize", "pull"] { + let response = f + .signed_post( + suffix, + &json!({"artifact_digest":f.request.artifact_digest}), + ) + .await; + assert_eq!(response.status().as_u16(), 400); + } + // A valid, registered second Worker is authenticated but cannot borrow the plan. + // Its builtin Profile selects no Plugins; it still cannot borrow this plan. + let other = delivery_worker(&f.api, None).await; + let path = f.path("pull"); + let body = serde_json::to_vec(&f.request).unwrap(); + let proof = f.proof(TEST_WORKSPACE_ID, &other.worker_id, "POST", &path, &body); + let response = f + .post_as(&other.worker_id, &path, body.clone(), Some(&proof)) + .await; + assert_eq!(response.status().as_u16(), 409); + let path = f + .path("pull") + .replace(TEST_WORKSPACE_ID, "foreign-workspace"); + let proof = f.proof( + "foreign-workspace", + &f.worker.worker_id, + "POST", + &path, + &body, + ); + assert_eq!( + f.post(&path, body, Some(&proof)).await.status().as_u16(), + 401 + ); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn delivery_requires_signature_bound_to_actual_body_method_path_and_single_use_jti() { + let f = DeliveryFixture::new(false).await; + let body = serde_json::to_vec(&f.request).unwrap(); + for suffix in ["authorize", "pull"] { + let path = f.path(suffix); + assert_eq!( + f.post(&path, body.clone(), None).await.status().as_u16(), + 401 + ); + let proof = f.proof(TEST_WORKSPACE_ID, &f.worker.worker_id, "POST", &path, &body); + let mut tampered = proof.clone().into_bytes(); + let index = tampered.len() - 8; + tampered[index] = if tampered[index] == b'A' { b'B' } else { b'A' }; + assert_eq!( + f.post( + &path, + body.clone(), + Some(std::str::from_utf8(&tampered).unwrap()) + ) + .await + .status() + .as_u16(), + 401 + ); + let mut changed = body.clone(); + changed.push(b' '); + assert_eq!( + f.post(&path, changed, Some(&proof)).await.status().as_u16(), + 401 + ); + for (method, signed_path) in [("GET", path.clone()), ("POST", format!("{path}?changed=1"))] + { + let proof = f.proof( + TEST_WORKSPACE_ID, + &f.worker.worker_id, + method, + &signed_path, + &body, + ); + assert_eq!( + f.post(&path, body.clone(), Some(&proof)) + .await + .status() + .as_u16(), + 401 + ); + } + let proof = f.proof(TEST_WORKSPACE_ID, &f.worker.worker_id, "POST", &path, &body); + let response = f.post(&path, body.clone(), Some(&proof)).await; + assert_eq!(response.status().as_u16(), 200); + if suffix == "pull" { + assert_eq!(response.bytes().await.unwrap().as_ref(), f.bytes); + } else { + assert_eq!( + response + .json::() + .await + .unwrap() + .request, + f.request + ); + } + let response = f.post(&path, body.clone(), Some(&proof)).await; + assert_eq!(response.status().as_u16(), 401); + let error: server_api::RepositoryApiError = response.json().await.unwrap(); + assert_eq!(error.error, "Unauthorized"); + } + f.assert_pending_pin(); +} + +#[tokio::test] +async fn populated_cache_cannot_bypass_artifact_revocation_or_release_pins() { + let f = DeliveryFixture::new(false).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + f.api + .config_store + .plugin_revoke( + TEST_WORKSPACE_ID, + "delivery-host", + &f.request.installation_id, + &PluginRevokeRequest { + expected_last_operation: f.installed.operation_id.clone(), + operation_key: "revoke-artifact".into(), + artifact_digest: f.request.artifact_digest.clone(), + reason: "withdraw executable".into(), + }, + ) + .unwrap(); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert!(matches!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + assert!(source.fetch(&f.request).await.is_err()); + } + f.assert_pending_pin(); +} + +#[tokio::test] +async fn unavailable_owner_denies_previously_verified_remote_cache_hit() { + let mut f = DeliveryFixture::new(false).await; + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + f.server.stop().await; + assert!(matches!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Unavailable(_)) + )); + // The same owner remains available through the non-HTTP adapter. + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + assert_eq!( + cache + .resolve(&local, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn released_persisted_plan_denies_remote_and_inprocess_warm_cache() { + let f = DeliveryFixture::new(false).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap(); + f.api + .config_store + .plugin_plan_transition( + TEST_WORKSPACE_ID, + &f.saved.plan_id, + PlanPinState::Pending, + PlanPinState::Released, + ) + .unwrap(); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert!(matches!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + } + assert_eq!( + f.api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, &f.saved.plan_id) + .unwrap() + .pin_state, + PlanPinState::Released + ); +} + +fn published_cache_files(root: &std::path::Path) -> Vec { + fs::read_dir(root) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "artifact")) + .collect() +} + +#[tokio::test] +async fn real_remote_cache_converges_repairs_corruption_and_refetches_after_eviction() { + let f = DeliveryFixture::new(false).await; + let root = f.dir.path().join("runtime-cache"); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + let (first, second) = tokio::join!( + cache.resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected), + cache.resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected), + ); + assert_eq!(first.unwrap().bytes(), f.bytes); + assert_eq!(second.unwrap().bytes(), f.bytes); + let files = published_cache_files(&root); + assert_eq!(files.len(), 1); + fs::write(&files[0], vec![0; f.bytes.len()]).unwrap(); + drop(cache); + // Reopening the Runtime-owned cache must not trust a corrupt published file. + let cache = f.cache("runtime-cache"); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + assert_eq!(fs::read(&files[0]).unwrap(), f.bytes); + fs::remove_file(&files[0]).unwrap(); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + assert_eq!(published_cache_files(&root), files); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn interrupted_real_http_pull_does_not_publish_partial_cache_and_retry_recovers() { + let f = DeliveryFixture::new(false).await; + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + f.interrupt_pull + .store(true, std::sync::atomic::Ordering::SeqCst); + assert!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .is_err() + ); + assert!(published_cache_files(&f.dir.path().join("runtime-cache")).is_empty()); + // Reopen after failure: neither abandoned staging bytes nor a previous owner + // observation can be used as a verified executable on the next attempt. + drop(cache); + let cache = f.cache("runtime-cache"); + f.interrupt_pull + .store(false, std::sync::atomic::Ordering::SeqCst); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + assert_eq!( + published_cache_files(&f.dir.path().join("runtime-cache")).len(), + 1 + ); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn missing_truncated_oversize_or_mismatched_owner_blob_never_publishes_remote_cache_entry() { + let f = DeliveryFixture::new(false).await; + let root = ServerConfig::server_data_root_for_data_dir(&f.api.config.data_root) + .join("plugin-artifacts"); + let blob = root + .join("blobs") + .join(f.request.artifact_digest.strip_prefix("sha256:").unwrap()); + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + fs::remove_file(&blob).unwrap(); + assert!(matches!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Unavailable(_)) + )); + for bad in [vec![], vec![0; f.bytes.len() + 1], vec![0; f.bytes.len()]] { + fs::write(&blob, bad).unwrap(); + assert!(matches!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Unavailable(_)) + )); + assert!(local.fetch(&f.request).await.is_err()); + assert!(published_cache_files(&f.dir.path().join("runtime-cache")).is_empty()); + } + fs::write(&blob, &f.bytes).unwrap(); + assert_eq!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn foreign_server_registration_cannot_lend_its_metadata_to_existing_source() { + let f = DeliveryFixture::new(false).await; + let foreign = DeliveryFixture::new(false).await; + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap(); + assert_ne!( + f.remote.server_authority_identity(), + foreign.remote.server_authority_identity() + ); + // Even a valid registration for another live Server cannot retarget a source + // or carry verified metadata across the original authority's cache key. + *f.registration.lock().unwrap() = foreign.registration.lock().unwrap().clone(); + assert!(matches!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + assert!(matches!( + foreign.remote.authorize(&f.request).await, + Err(DeliveryError::Denied) + )); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn revoked_server_runtime_binding_denies_both_adapters_even_on_cache_hit() { + let f = DeliveryFixture::new(false).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap(); + let mut binding = f + .api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, &f.worker.runtime_id) + .await + .unwrap() + .unwrap(); + binding.revoked_at = Some(now_registry_timestamp()); + f.api + .config_store + .upsert_workspace_runtime_binding(binding, true) + .unwrap(); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert!(matches!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + } + f.assert_pending_pin(); +} + +#[tokio::test] +async fn subject_delegation_replacement_denies_child_warm_cache_and_preserves_saved_plan() { + let f = DeliveryFixture::new(true).await; + let permission = f.artifact_permission(); + f.policy(vec![], vec![permission.clone()]); + // Use the signed, replay-safe Worker control route to issue the child grant. + let path = format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"); + let body = serde_json::to_vec(&json!({ + "runtime_id": f.worker.runtime_id, "display_name":"Artifact delegated consumer", + "profile":"builtin:coder", "initial_submit":[], "workdir_attachments":[], "feature_connections":{}, + "control_operation_id":"delivery-delegated-child", + "execution_grant":{"permissions":[permission], "redelegation":[]} + })).unwrap(); + let proof = f.proof(TEST_WORKSPACE_ID, &f.worker.worker_id, "POST", &path, &body); + let response = f.post(&path, body, Some(&proof)).await; + let status = response.status(); + let body: Value = response.json().await.unwrap(); + assert_eq!(status.as_u16(), 200, "{body}"); + let child = RuntimeWorkerRef::new( + body["runtime_id"].as_str().unwrap(), + body["worker_id"].as_str().unwrap(), + ); + let execution = f.api.plugin_execution_identity(&child).await.unwrap(); + assert_eq!( + execution.principal_id, + format!("subject:{}", f.subject.as_ref().unwrap()) + ); + assert!(execution.delegation_id.is_some()); + let saved = configure_delivery_plan(&f.api, &f.installed, execution, "delegated-launch"); + let request = PluginArtifactDeliveryRequest { + plan_id: saved.plan_id.clone(), + plan_fingerprint: saved.fingerprint.clone(), + ..f.request.clone() + }; + let client = RuntimeWorkspaceRequestClient::new(TEST_WORKSPACE_ID, &f.url, &child.runtime_id) + .with_runtime_request_source(&f.identity, &f.url); + let current = f.registration.clone(); + let remote = RemotePluginArtifactSource::new( + client, + child.worker_id.clone(), + Arc::new(move || Ok(current.lock().unwrap().clone())), + ) + .unwrap(); + let local = f.api.plugin_artifact_source(child.clone()).await.unwrap(); + let cache = f.cache("child-runtime-cache"); + let expected = f.expected(); + for source in [&remote as &dyn ArtifactSource, &local] { + assert_eq!( + cache + .resolve(source, TEST_WORKSPACE_ID, &request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + } + // Replacement fences the saved source_change_id, even with the same resource. + f.policy(vec![], vec![]); + assert!(f.api.plugin_execution_identity(&child).await.is_err()); + for source in [&remote as &dyn ArtifactSource, &local] { + assert!(matches!( + cache + .resolve(source, TEST_WORKSPACE_ID, &request, &expected) + .await, + Err(DeliveryError::Denied) + )); + assert!(source.fetch(&request).await.is_err()); + } + let retained = f + .api + .config_store + .plugin_plan_load(TEST_WORKSPACE_ID, &saved.plan_id) + .unwrap(); + assert_eq!(retained.plan, saved.plan); + assert_eq!(retained.pin_state, PlanPinState::Pending); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn runtime_registration_revocation_cannot_reuse_remote_verified_cache() { + let f = DeliveryFixture::new(false).await; + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap(); + { + let mut registration = f.registration.lock().unwrap(); + worker_runtime::workspace_issuer::revoke_workspace_issuer_trust( + std::slice::from_mut(&mut *registration), + TEST_WORKSPACE_ID, + worker_runtime::auth::unix_now_seconds() as i64, + ) + .unwrap(); + assert_eq!(registration.state, WorkspaceIssuerTrustState::Revoked); + } + assert!(matches!( + cache + .resolve(&f.remote, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + f.assert_pending_pin(); +} + +#[tokio::test] +async fn subject_exact_artifact_permission_is_live_and_does_not_grant_https() { + let f = DeliveryFixture::new(true).await; + let local = f + .api + .plugin_artifact_source(f.worker.clone()) + .await + .unwrap(); + let cache = f.cache("runtime-cache"); + let expected = f.expected(); + assert!(matches!( + f.remote.authorize(&f.request).await, + Err(DeliveryError::Denied) + )); + let permission = f.artifact_permission(); + let mut wrong_operation = permission.clone(); + wrong_operation.operation = "plugin".into(); + let mut wrong_resource = permission.clone(); + wrong_resource.resource = PermissionResource::Workspace { + workspace_id: TEST_WORKSPACE_ID.into(), + }; + let mut wrong_installation = permission.clone(); + wrong_installation.resource = PermissionResource::PluginInstallation { + workspace_id: TEST_WORKSPACE_ID.into(), + installation_id: "other-installation".into(), + }; + for wrong in [wrong_operation, wrong_resource, wrong_installation] { + f.policy(vec![wrong], vec![]); + assert!(matches!( + f.remote.authorize(&f.request).await, + Err(DeliveryError::Denied) + )); + assert!(local.fetch(&f.request).await.is_err()); + } + f.policy(vec![permission], vec![]); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert_eq!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await + .unwrap() + .bytes(), + f.bytes + ); + } + // Only a policy decision is requested; this test never contacts a destination. + let path = format!("/api/w/{TEST_WORKSPACE_ID}/authorization/check"); + let body = serde_json::to_vec(&json!({"operation":"plugin.https.request","resource":{ + "kind":"plugin_installation","workspace_id":TEST_WORKSPACE_ID,"installation_id":f.request.installation_id + }})).unwrap(); + let proof = f.proof(TEST_WORKSPACE_ID, &f.worker.worker_id, "POST", &path, &body); + let response = f.post(&path, body, Some(&proof)).await; + assert_eq!(response.status().as_u16(), 403); + f.policy(vec![], vec![]); + for source in [&f.remote as &dyn ArtifactSource, &local] { + assert!(matches!( + cache + .resolve(source, TEST_WORKSPACE_ID, &f.request, &expected) + .await, + Err(DeliveryError::Denied) + )); + assert!(source.fetch(&f.request).await.is_err()); + } + f.assert_pending_pin(); +} diff --git a/web/workspace/src/lib/generated/plugin-api.ts b/web/workspace/src/lib/generated/plugin-api.ts index b57411c3d..d1178a014 100644 --- a/web/workspace/src/lib/generated/plugin-api.ts +++ b/web/workspace/src/lib/generated/plugin-api.ts @@ -34,3 +34,7 @@ export type PluginListResponse = { installations: Array