Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: deliver exact Plugin artifacts through current Host authority

Browse Source 
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.
This commit is contained in:
Hare
2026-10-30 04:09:12 +09:00
parent 732fe5c4c1
commit 3a13301f75
23 changed files with 3971 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+4
View File
@@ -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",
crates/manifest/src/plugin.rs
+23
View File
@@ -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)))
}
crates/plugin-runtime/Cargo.toml
+4 -1
View File
@@ -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"
crates/plugin-runtime/README.md
+3 -1
View File
@@ -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.
crates/plugin-runtime/src/delivery.rs
+509
View File
@@ -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<Box<dyn AsyncRead + Send + Unpin>>;
#[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<PluginArtifactDeliveryMetadata, DeliveryError>;
async fn fetch(
&self,
request: &PluginArtifactDeliveryRequest,
) -> Result<ArtifactStream, DeliveryError>;
}
#[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<u8>,
}
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<Path>,
limits: CacheLimits,
) -> Result<Self, DeliveryError> {
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<VerifiedArtifact, DeliveryError> {
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<VerifiedArtifact, DeliveryError> {
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<Option<Vec<u8>>, 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<Vec<u8>, 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<Vec<Entry>, 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<Entry>,
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<File, DeliveryError> {
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: serde::Serialize, B: serde::Serialize>(
a: &A,
b: &B,
) -> Result<bool, DeliveryError> {
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;
crates/plugin-runtime/src/delivery/tests.rs
+985
View File
@@ -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<PluginArtifactDeliveryMetadata>,
bytes: Vec<u8>,
fetches: AtomicUsize,
authorizations: AtomicUsize,
deny_on: AtomicUsize,
unavailable: AtomicUsize,
transfer: Transfer,
read_bytes: Arc<AtomicUsize>,
paused: Arc<Semaphore>,
}
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<VerifiedArtifact, DeliveryError> {
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<PluginArtifactDeliveryMetadata, DeliveryError> {
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<ArtifactStream, DeliveryError> {
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<R> {
reader: R,
count: Arc<AtomicUsize>,
}
impl<R: AsyncRead + Unpin> AsyncRead for CountingReader<R> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buffer: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
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<u8>,
position: usize,
transfer: Transfer,
paused: Arc<Semaphore>,
}
impl AsyncRead for FaultReader {
fn poll_read(
mut self: Pin<&mut Self>,
_: &mut Context<'_>,
buffer: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
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<PathBuf> {
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<PathBuf> {
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<Option<String>>,
stdout_thread: Option<std::thread::JoinHandle<()>>,
stderr_thread: Option<std::thread::JoinHandle<String>>,
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<String> {
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);
}
crates/plugin-runtime/src/lib.rs
+1
View File
@@ -10,6 +10,7 @@ pub mod bindings {
});
}
pub mod delivery;
mod executor;
pub mod host;
pub mod owner;
crates/server-api/src/lib.rs
+20
View File
@@ -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<PluginOperationResponse, RepositoryApiError>;
#[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<PluginArtifactDeliveryMetadata, RepositoryApiError>;
#[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<BinaryBody, RepositoryApiError>;
// 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",
crates/server-api/src/plugin.rs
+34
View File
@@ -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,
}
crates/server-api/src/plugin_types.rs
+53
View File
@@ -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::<crate::PluginArtifactDeliveryRequest>(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::<crate::PluginArtifactDeliveryRequest>(invalid).is_err()
);
}
assert!(
serde_json::from_value::<crate::PluginArtifactDeliveryRequest>(
serde_json::json!({"artifact_digest":"digest"})
)
.is_err()
);
}
#[cfg(feature = "typescript")]
#[test]
fn generated_plugin_types_are_current() {
crates/worker-runtime/Cargo.toml
+3 -1
View File
@@ -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
crates/worker-runtime/src/workspace_request.rs
+2
View File
@@ -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";
crates/worker-runtime/src/workspace_request/plugin_delivery.rs
+182
View File
@@ -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<WorkspaceIssuerTrustRecord, DeliveryError> + Send + Sync;
pub struct RemotePluginArtifactSource {
client: RuntimeWorkspaceRequestClient,
worker_id: String,
authority: String,
current_server: Arc<CurrentArtifactServer>,
}
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<CurrentArtifactServer>,
) -> Result<Self, DeliveryError> {
if worker_id.is_empty() {
return Err(DeliveryError::InvalidRequest);
}
let registration = current_server()?;
let authority = server_api::plugin_server_authority_identity(&registration.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(&registration.public_bundle())
!= self.authority
{
return Err(DeliveryError::Denied);
}
Ok(())
}
fn request(
&self,
suffix: &str,
selection: &PluginArtifactDeliveryRequest,
) -> Result<RuntimeWorkspaceRequest, DeliveryError> {
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<reqwest::Response, DeliveryError> {
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<PluginArtifactDeliveryMetadata, DeliveryError> {
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<ArtifactStream, DeliveryError> {
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,
))))
}
}
crates/workspace-server/Cargo.toml
+1
View File
@@ -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
crates/workspace-server/src/lib.rs
+1
View File
@@ -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;
crates/workspace-server/src/plugin_catalog.rs
+6 -6
View File
@@ -170,7 +170,7 @@ impl Drop for PendingFile {
let _ = fs::remove_file(&self.0);
}
}
fn read_blob(root: &Path, artifact: &str, size: u64) -> Result<Vec<u8>> {
pub(crate) fn read_blob(root: &Path, artifact: &str, size: u64) -> Result<Vec<u8>> {
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<PluginArtifact> {
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,
crates/workspace-server/src/plugin_delivery.rs
+146
View File
@@ -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<yoi_plugin_runtime::delivery::ArtifactExpectation> {
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<PluginArtifactDeliveryMetadata> {
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<Vec<u8>> {
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<PluginArtifactDeliveryMetadata> {
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<Vec<u8>> {
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;
crates/workspace-server/src/plugin_delivery/tests.rs
+591
View File
@@ -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<PluginArtifactDeliveryMetadata> {
self.store
.plugin_delivery_authorize("a", &execution(), request)
}
fn bytes(&self, request: &PluginArtifactDeliveryRequest) -> Result<Vec<u8>> {
self.store
.plugin_delivery_bytes(&self.root(), "a", &execution(), request)
}
fn retained(&self, plan_id: &str) -> Vec<String> {
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::<rusqlite::Result<_>>()?)).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<T: std::fmt::Debug>(result: Result<T>, expected: PluginCatalogError) {
assert!(
matches!(result, Err(Error::PluginCatalog(actual)) if actual == expected),
"expected {expected:?}, got {result:?}"
);
}
fn package(version: &str) -> Vec<u8> {
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);
}
crates/workspace-server/src/server.rs
+29
View File
@@ -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<server_api::BinaryBody, server_api::RepositoryApiError> {
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");
}
crates/workspace-server/src/server_plugin.rs
+2 -2
View File
@@ -24,7 +24,7 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> 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![],
)),
)
crates/workspace-server/src/server_plugin_delivery.rs
+215
View File
@@ -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<ServerApiContractService>) -> 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<ApiError>) -> 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<RuntimeWorkerRef, server_api::RepositoryApiError> {
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<InProcessPluginArtifactSource> {
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<String> {
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<crate::plugin_plan::HostExecutionIdentity, server_api::RepositoryApiError>
{
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<PluginArtifactDeliveryMetadata, server_api::RepositoryApiError> {
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<Vec<u8>, 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<PluginArtifactDeliveryMetadata, server_api::RepositoryApiError> {
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<server_api::BinaryBody, server_api::RepositoryApiError> {
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<PluginArtifactDeliveryMetadata, DeliveryError> {
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<ArtifactStream, DeliveryError> {
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)))
}
}
crates/workspace-server/src/server_plugin_delivery_tests.rs
+1153
View File
File diff suppressed because it is too large Load Diff
web/workspace/src/lib/generated/plugin-api.ts
+4
View File
@@ -34,3 +34,7 @@ export type PluginListResponse = { installations: Array<PluginInstallationSummar
export type PluginOperationKind = "install" | "add_artifact" | "revoke" | "uninstall";
export type PluginOperationResponse = { operation_id: string, kind: PluginOperationKind, installation: PluginInstallation, };
export type PluginArtifactDeliveryRequest = { plan_id: string, plan_fingerprint: string, installation_id: string, artifact_digest: string, };
export type PluginArtifactDeliveryMetadata = { workspace_id: string, request: PluginArtifactDeliveryRequest, artifact: PluginArtifact, };