From 0cd481173952c3d029c43c41ae6ce37420e4736a Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 04:40:40 +0900 Subject: [PATCH] fix: supply receipt-free Plugin preparation and authenticated Host entry --- crates/plugin-runtime/src/owner.rs | 22 + crates/server-api/src/plugin_host.rs | 6 + crates/workspace-server/src/plugin_host.rs | 33 +- .../src/server_plugin_host.rs | 224 +++++++- .../src/server_plugin_host_tests.rs | 526 ++++++++++++++++-- openapi/server-api.json | 31 +- .../src/lib/generated/companion-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 2 +- web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/workdir-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 2 +- 12 files changed, 780 insertions(+), 74 deletions(-) diff --git a/crates/plugin-runtime/src/owner.rs b/crates/plugin-runtime/src/owner.rs index db79fd6ef..cf74bc53a 100644 --- a/crates/plugin-runtime/src/owner.rs +++ b/crates/plugin-runtime/src/owner.rs @@ -21,6 +21,27 @@ use tokio::time::Instant; pub trait PluginHostOwner: Send + Sync { async fn call(&self, request: PluginHostRequest) -> Result; + + /// Recheck export/restore admission without inventing or consuming a Tool call. + /// Success is a current observation, never a reusable execution capability. + async fn prepare( + &self, + plan_id: String, + installation_id: String, + tools: Vec, + ) -> Result<(), PluginHostError> { + match self + .call(PluginHostRequest::Prepare { + plan_id, + installation_id, + tools, + }) + .await? + { + PluginHostResponse::Authorized => Ok(()), + _ => Err(protocol_error(false)), + } + } } /// Uses an injected, already configured generated client (including its existing @@ -537,6 +558,7 @@ mod tests { }); } Ok(match request { + PluginHostRequest::Prepare { .. } => PluginHostResponse::Authorized, PluginHostRequest::Begin { .. } => PluginHostResponse::Invocation { invocation_id: "owner-issued".into(), expires_at_ms: i64::try_from( diff --git a/crates/server-api/src/plugin_host.rs b/crates/server-api/src/plugin_host.rs index 9933facd8..8739880d8 100644 --- a/crates/server-api/src/plugin_host.rs +++ b/crates/server-api/src/plugin_host.rs @@ -7,6 +7,12 @@ use serde::{Deserialize, Serialize}; #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] pub enum PluginHostRequest { + /// Fresh export/restore admission only; issues no invocation or effect identity. + Prepare { + plan_id: String, + installation_id: String, + tools: Vec, + }, Begin { plan_id: String, installation_id: String, diff --git a/crates/workspace-server/src/plugin_host.rs b/crates/workspace-server/src/plugin_host.rs index 2f9bc1d3d..84d1e80af 100644 --- a/crates/workspace-server/src/plugin_host.rs +++ b/crates/workspace-server/src/plugin_host.rs @@ -35,7 +35,7 @@ pub fn authorize( execution: &HostExecutionIdentity, plan_id: &str, installation: &str, - consumer: &PluginConsumer, + tools: &[String], operation: &PluginHostOperation, ) -> Result { let plan = plugin_plan_load(tx, workspace, plan_id)?; @@ -49,22 +49,23 @@ pub fn authorize( .find(|pin| pin.installation_id() == installation) .cloned() .ok_or_else(|| failure(PluginHostErrorCode::Denied))?; - let tool_name = match consumer { - PluginConsumer::ToolCall { - tool_call_id, - tool_name, - } if !tool_call_id.is_empty() - && tool_call_id.len() <= 512 - && !tool_call_id.chars().any(char::is_control) => - { - tool_name - } - // The scope is reserved for future authenticated consumers. This initial - // Tool implementation cannot manufacture Service/event admission. - _ => return Err(failure(PluginHostErrorCode::Unsupported)), - }; + if tools.is_empty() + || tools.len() > 256 + || tools + .iter() + .any(|tool| tool.is_empty() || tool.len() > 512 || tool.chars().any(char::is_control)) + || tools + .iter() + .collect::>() + .len() + != tools.len() + { + return Err(failure(PluginHostErrorCode::InvalidInput)); + } if !pin.projection().surfaces.contains(&PluginSurface::Tool) - || !pin.projection().tools.contains(tool_name) + || tools + .iter() + .any(|tool| !pin.projection().tools.contains(tool)) { return Err(failure(PluginHostErrorCode::Denied)); } diff --git a/crates/workspace-server/src/server_plugin_host.rs b/crates/workspace-server/src/server_plugin_host.rs index efea27c46..bbf8ce442 100644 --- a/crates/workspace-server/src/server_plugin_host.rs +++ b/crates/workspace-server/src/server_plugin_host.rs @@ -3,7 +3,75 @@ //! JSON. Network dispatch is single-attempt and follows durable effect admission. use super::*; use crate::{plugin_host as policy, plugin_https as https, plugin_receipt as receipt}; +use rusqlite::OptionalExtension; use server_api::*; + +impl WorkspaceApi { + /// Host plan creation after the canonical reservation and grant/Job binding, + /// before Runtime dispatch. The reservation is re-read, never trusted as a + /// caller-issued principal. This derives identity only, not execution rights. + pub(crate) fn plugin_reserved_execution_identity( + &self, + runtime_id: &str, + reservation: &crate::store::WorkerCreateReservation, + expected_grant: Option<&server_api::authorization::WorkerExecutionGrant>, + ) -> crate::Result { + let worker = RuntimeWorkerRef::new(runtime_id, reservation.worker_id.to_string()); + self.plugin_reserved_identity( + &worker, + Some(&reservation.create_fingerprint), + expected_grant, + ) + } + + fn plugin_reserved_identity( + &self, + worker: &RuntimeWorkerRef, + expected_fingerprint: Option<&str>, + expected_grant: Option<&server_api::authorization::WorkerExecutionGrant>, + ) -> crate::Result { + let refused = || policy::failure(PluginHostErrorCode::Denied); + let (fingerprint, allocation) = self.config_store.with_conn(|conn| { + Ok(conn.query_row("SELECT create_fingerprint,allocation_key FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='reserved'", + rusqlite::params![self.workspace_id(), worker.runtime_id, worker.worker_id], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))).optional()?) + })?.ok_or_else(refused)?; + // A Job which has not yet acquired its canonical attempt/Worker binding + // must not accidentally become a legacy ordinary Worker principal. + if allocation.starts_with("backend-job:") + && self + .store + .worker_registry_projection(self.workspace_id(), worker)? + .and_then(|projection| projection.job) + .is_none() + { + return Err(refused()); + } + if expected_fingerprint.is_some_and(|expected| expected != fingerprint) { + return Err(refused()); + } + self.config_store + .get_workspace_runtime_binding(self.workspace_id(), &worker.runtime_id)? + .filter(|binding| binding.revoked_at.is_none()) + .ok_or_else(refused)?; + self.store + .require_current_worker_singleton_owner(self.workspace_id(), worker)?; + if let Some(expected) = expected_grant { + let mut expected = expected.clone(); + expected.worker_id = worker.worker_id.clone(); + let actual = open_subjektiv_store(self) + .map_err(|e| e.error)? + .worker_execution_grant(&worker.runtime_id, &worker.worker_id) + .map_err(|_| refused())?; + if actual.as_ref() != Some(&expected) { + return Err(refused()); + } + } + // This is the same owner as live identity. Do not manufacture a separate + // principal/delegation spelling or copy policy into the reservation. + subject_authorization::plugin_execution_identity(self, worker).map_err(|e| e.error) + } +} + use tokio_util::sync::CancellationToken; const RECEIPT_RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1000; @@ -36,14 +104,62 @@ async fn check( installation: String, consumer: PluginConsumer, operation: PluginHostOperation, +) -> std::result::Result { + let tools = match consumer { + PluginConsumer::ToolCall { + tool_call_id, + tool_name, + } if !tool_call_id.is_empty() + && tool_call_id.len() <= 512 + && !tool_call_id.chars().any(char::is_control) => + { + vec![tool_name] + } + _ => { + return Err(policy::map_error(policy::failure( + PluginHostErrorCode::Unsupported, + ))); + } + }; + check_tools( + api, + context, + workspace, + plan_id, + installation, + tools, + operation, + false, + ) + .await +} + +async fn check_tools( + api: Arc, + context: ServerRequestContext, + workspace: String, + plan_id: String, + installation: String, + tools: Vec, + operation: PluginHostOperation, + preparing: bool, ) -> std::result::Result { validate_workspace_scope(&api, &workspace).map_err(|e| policy::map_error(e.error))?; let worker = subject_authorization::source_from_context(&context) .map_err(|e| policy::map_error(e.error))?; - let execution = api - .plugin_execution_identity(&worker) - .await - .map_err(policy::map_error)?; + let execution = if preparing + && api + .store + .has_active_worker_create_reservation(api.workspace_id(), &worker) + .map_err(policy::map_error)? + { + api.plugin_reserved_identity(&worker, None, None) + .map_err(policy::map_error)? + } else { + api.plugin_execution_identity(&worker) + .await + .map_err(policy::map_error)? + }; let executor = api.config_store.db_executor(); executor .run(move || { @@ -66,7 +182,7 @@ async fn check( &execution, &plan_id, &installation, - &consumer, + &tools, &operation, ) }) @@ -129,11 +245,36 @@ async fn invocation( #[derive(Clone)] pub struct EmbeddedPluginHostOwner { api: Arc, - context: ServerRequestContext, + source: EmbeddedPluginSource, workspace: String, transport_override: Option>, } + +#[derive(Clone)] +enum EmbeddedPluginSource { + Signed(Arc), + #[cfg(test)] + Verified(ServerRequestContext), +} + impl EmbeddedPluginHostOwner { + /// Uses the Runtime's existing signer. Every request is verified by the same + /// proof owner as Remote; this factory neither creates Worker claims nor + /// borrows human credentials. No HTTP loopback or standalone fallback. + pub(crate) fn authenticated( + api: Arc, + workspace: String, + authorizer: Arc, + ) -> Self { + Self { + api, + workspace, + source: EmbeddedPluginSource::Signed(authorizer), + transport_override: None, + } + } + + #[cfg(test)] pub(crate) fn new( api: Arc, context: ServerRequestContext, @@ -141,7 +282,7 @@ impl EmbeddedPluginHostOwner { ) -> Self { Self { api, - context, + source: EmbeddedPluginSource::Verified(context), workspace, transport_override: None, } @@ -158,9 +299,58 @@ impl yoi_plugin_runtime::owner::PluginHostOwner for EmbeddedPluginHostOwner { &self, request: PluginHostRequest, ) -> std::result::Result { + let context = match &self.source { + #[cfg(test)] + EmbeddedPluginSource::Verified(context) => context.clone(), + EmbeddedPluginSource::Signed(authorizer) => { + let body = serde_json::to_vec(&request).map_err(|_| denied())?; + let mut url = + url::Url::parse("https://plugin-owner.invalid/").map_err(|_| denied())?; + url.path_segments_mut().map_err(|_| denied())?.extend([ + "api", + "w", + &self.workspace, + "plugin-host", + ]); + let path = url.path(); + let headers = authorizer + .authorize(server_api::client_support::AuthorizerRequest { + method: &reqwest::Method::POST, + path_and_query: path, + body: &body, + }) + .map_err(|_| denied())?; + let proof = headers + .get(worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(denied)?; + let source = crate::worker_source::verify_runtime_request_source_proof( + &self.api, + proof, + &self.workspace, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + "POST", + path, + &worker_runtime::auth::request_body_digest(&body), + ) + .await + .map_err(|_| denied())?; + server_request_context( + None, + None, + Some(ServerRuntimeSource { + authentication_binding_id: source.authentication_binding_id, + runtime_id: source.runtime_id, + worker_id: source.worker_id, + }), + None, + &headers, + ) + } + }; execute_with_transport( self.api.clone(), - self.context.clone(), + context, self.workspace.clone(), request, self.transport_override.clone(), @@ -192,6 +382,24 @@ async fn execute_with_transport( // Human management and a Runtime-only (no Worker) source both fail closed. subject_authorization::source_from_context(&context).map_err(|_| denied())?; match request { + PluginHostRequest::Prepare { + plan_id, + installation_id, + tools, + } => { + check_tools( + api, + context, + workspace, + plan_id, + installation_id, + tools, + PluginHostOperation::Admit, + true, + ) + .await?; + Ok(PluginHostResponse::Authorized) + } PluginHostRequest::Begin { plan_id, installation_id, diff --git a/crates/workspace-server/src/server_plugin_host_tests.rs b/crates/workspace-server/src/server_plugin_host_tests.rs index 27cc1ee72..913f59aa3 100644 --- a/crates/workspace-server/src/server_plugin_host_tests.rs +++ b/crates/workspace-server/src/server_plugin_host_tests.rs @@ -33,6 +33,7 @@ const METHODS: &[PluginHttpMethod] = &[ struct Fixture { api: WorkspaceApi, + runtime: WorkdirlessFixtureRuntime, identity: RuntimeIdentityMaterial, source: RuntimeWorkerRef, subject: String, @@ -51,8 +52,8 @@ impl Fixture { let directory = tempfile::tempdir().unwrap(); let mut api = test_api(directory.path()).await; register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; - api.runtime - .register_or_replace(WorkdirlessFixtureRuntime::default()); + let runtime = WorkdirlessFixtureRuntime::default(); + api.runtime.register_or_replace(runtime.clone()); let subject = open_subjektiv_store(&api) .unwrap() .create_subject(crate::subjektiv::SubjectRole::new("plugin-host-integration").unwrap()) @@ -175,6 +176,7 @@ impl Fixture { .unwrap(); Self { api, + runtime, identity, source, subject, @@ -254,10 +256,15 @@ impl Fixture { } async fn owner(&self, transport: Option) -> Arc { - let owner = plugin_host::EmbeddedPluginHostOwner::new( + let owner = plugin_host::EmbeddedPluginHostOwner::authenticated( Arc::new(self.api.clone()), - self.verified_context().await, TEST_WORKSPACE_ID.into(), + Arc::new(SignedAuthorizer { + signer: worker_runtime::auth::RuntimeRequestSourceSigner::from_identity( + &self.identity, + ), + worker_id: self.source.worker_id.clone(), + }), ); Arc::new(match transport { None => owner, @@ -729,33 +736,82 @@ async fn all_methods_lost_dispatch_acknowledgement_replays_indeterminate_without // A new real receipt connection and owner have no in-memory replay state. // This models lost owner completion, not an arbitrary OS crash. let mut recovered_api = f.api.clone(); - recovered_api.config_store = Arc::new(SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap()); - let recovered = Arc::new(plugin_host::EmbeddedPluginHostOwner::new( - Arc::new(recovered_api.clone()), f.verified_context().await, TEST_WORKSPACE_ID.into(), - ).with_transport(peer.transport())); + recovered_api.config_store = + Arc::new(SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap()); + let recovered = Arc::new( + plugin_host::EmbeddedPluginHostOwner::new( + Arc::new(recovered_api.clone()), + f.verified_context().await, + TEST_WORKSPACE_ID.into(), + ) + .with_transport(peer.transport()), + ); let restored = f.host(recovered.clone(), "lost-acknowledgements").await; - assert_eq!(restored.context().invocation_id, host.context().invocation_id); + assert_eq!( + restored.context().invocation_id, + host.context().invocation_id + ); for (index, method) in METHODS.iter().enumerate() { - let error = recovered.call(send(&restored.context().invocation_id, method.clone(), &format!("lost-{index}"))).await.unwrap_err(); + let error = recovered + .call(send( + &restored.context().invocation_id, + method.clone(), + &format!("lost-{index}"), + )) + .await + .unwrap_err(); assert_eq!(error.code, PluginHostErrorCode::Indeterminate); } - let expiry = recovered_api.config_store.with_immediate_transaction(|tx| { - let record = crate::plugin_receipt::lookup(tx, TEST_WORKSPACE_ID, &restored.context().invocation_id)?; - crate::plugin_receipt::prune(tx, TEST_WORKSPACE_ID, record.expires_at_ms) - }).unwrap(); + let expiry = recovered_api + .config_store + .with_immediate_transaction(|tx| { + let record = crate::plugin_receipt::lookup( + tx, + TEST_WORKSPACE_ID, + &restored.context().invocation_id, + )?; + crate::plugin_receipt::prune(tx, TEST_WORKSPACE_ID, record.expires_at_ms) + }) + .unwrap(); assert_eq!(expiry, METHODS.len()); for key in ["lost-0", "unused-after-retention"] { - assert_eq!(recovered.call(send(&restored.context().invocation_id, PluginHttpMethod::Get, key)).await.unwrap_err().code, - PluginHostErrorCode::ReceiptExpired); + assert_eq!( + recovered + .call(send( + &restored.context().invocation_id, + PluginHttpMethod::Get, + key + )) + .await + .unwrap_err() + .code, + PluginHostErrorCode::ReceiptExpired + ); } assert_eq!(peer.connection_count(), METHODS.len()); - assert_eq!(recovered.call(PluginHostRequest::Authorize { - invocation_id: restored.context().invocation_id.clone(), operation: PluginHostOperation::Publish, - }).await.unwrap_err().code, PluginHostErrorCode::ReceiptExpired); - assert_eq!(recovered.call(PluginHostRequest::Begin { - plan_id: f.plan.plan_id.clone(), installation_id: f.installed.installation.installation_id.clone(), - consumer: f.consumer("lost-acknowledgements"), - }).await.unwrap_err().code, PluginHostErrorCode::ReceiptExpired); + assert_eq!( + recovered + .call(PluginHostRequest::Authorize { + invocation_id: restored.context().invocation_id.clone(), + operation: PluginHostOperation::Publish, + }) + .await + .unwrap_err() + .code, + PluginHostErrorCode::ReceiptExpired + ); + assert_eq!( + recovered + .call(PluginHostRequest::Begin { + plan_id: f.plan.plan_id.clone(), + installation_id: f.installed.installation.installation_id.clone(), + consumer: f.consumer("lost-acknowledgements"), + }) + .await + .unwrap_err() + .code, + PluginHostErrorCode::ReceiptExpired + ); f.shutdown().await; } @@ -1041,7 +1097,11 @@ async fn revocation_or_cancellation_during_dns_prevents_dispatch_and_recovery() result = &mut call => panic!("send finished before DNS gate: {result:?}"), result = started => result.unwrap(), } - assert_eq!(f.count("plugin_effect_receipts"), 1, "admission committed before DNS"); + assert_eq!( + f.count("plugin_effect_receipts"), + 1, + "admission committed before DNS" + ); if cancel { host.cancel().await.unwrap(); host.cancel().await.unwrap(); @@ -1050,16 +1110,34 @@ async fn revocation_or_cancellation_during_dns_prevents_dispatch_and_recovery() } release.send(vec![peer.public_address()]).unwrap(); let error = call.await.unwrap_err(); - assert_eq!(error.code, if cancel { PluginHostErrorCode::Cancelled } else { PluginHostErrorCode::Denied }); - assert!(owner.call(send(&id, PluginHttpMethod::Post, "key")).await.is_err()); - assert!(owner.call(send(&id, PluginHttpMethod::Get, "unused-key")).await.is_err()); + assert_eq!( + error.code, + if cancel { + PluginHostErrorCode::Cancelled + } else { + PluginHostErrorCode::Denied + } + ); + assert!( + owner + .call(send(&id, PluginHttpMethod::Post, "key")) + .await + .is_err() + ); + assert!( + owner + .call(send(&id, PluginHttpMethod::Get, "unused-key")) + .await + .is_err() + ); assert_eq!(peer.connection_count(), 0); f.shutdown().await; } } #[tokio::test] -async fn authority_removed_after_dispatch_withholds_response_and_restored_read_replays_without_send() { +async fn authority_removed_after_dispatch_withholds_response_and_restored_read_replays_without_send() + { use crate::plugin_https::test_support::Reply; let (release, waiting) = tokio::sync::oneshot::channel(); let mut peer = Peer::new(HOST, vec![Reply::Gated(waiting)]).await; @@ -1074,18 +1152,31 @@ async fn authority_removed_after_dispatch_withholds_response_and_restored_read_r _ = peer.observed() => {}, } f.deny(); - release.send(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndata".to_vec()).unwrap(); + release + .send(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndata".to_vec()) + .unwrap(); assert_denied(call.await); assert_denied(owner.call(send(&id, PluginHttpMethod::Post, "key")).await); - set_policy(&f.api, &f.subject, &f.installed.installation.installation_id, true); - let replay = owner.call(send(&id, PluginHttpMethod::Post, "key")).await.unwrap(); - assert!(matches!(replay, PluginHostResponse::Response { response } if response.body == b"data")); + set_policy( + &f.api, + &f.subject, + &f.installed.installation.installation_id, + true, + ); + let replay = owner + .call(send(&id, PluginHttpMethod::Post, "key")) + .await + .unwrap(); + assert!( + matches!(replay, PluginHostResponse::Response { response } if response.body == b"data") + ); assert_eq!(peer.connection_count(), 1); f.shutdown().await; } #[tokio::test] -async fn generated_remote_owner_component_https_and_embedded_replay_share_receipt_and_live_denial() { +async fn generated_remote_owner_component_https_and_embedded_replay_share_receipt_and_live_denial() +{ use yoi_plugin_runtime::owner::RemotePluginHostOwner; let mut peer = Peer::new(HOST, vec![response(503, &[], b"peer data")]).await; let mut f = Fixture::new(Some(peer_binding(&peer))).await; @@ -1093,27 +1184,376 @@ async fn generated_remote_owner_component_https_and_embedded_replay_share_receip let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let router = build_router(f.api.clone()); - let mut server = ServerTask(tokio::spawn(async move { axum::serve(listener, router).await.unwrap(); })); + let mut server = ServerTask(tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + })); let client = server_api::ServerApiClient::builder(format!("http://{address}")) - .unwrap().client(reqwest::Client::builder().no_proxy().timeout(std::time::Duration::from_secs(10)).build().unwrap()) + .unwrap() + .client( + reqwest::Client::builder() + .no_proxy() + .timeout(std::time::Duration::from_secs(10)) + .build() + .unwrap(), + ) .authorizer(SignedAuthorizer { signer: worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&f.identity), worker_id: f.source.worker_id.clone(), - }).build().unwrap(); - let owner = Arc::new(RemotePluginHostOwner::new(Arc::new(client), TEST_WORKSPACE_ID.into())); + }) + .build() + .unwrap(); + let owner = Arc::new(RemotePluginHostOwner::new( + Arc::new(client), + TEST_WORKSPACE_ID.into(), + )); let host = f.host(owner.clone(), "remote-https").await; let id = host.context().invocation_id.clone(); let result = f.invoke(host).await.unwrap(); assert_eq!(result.output.summary, "HTTP response observed"); - assert!(String::from_utf8(peer.observed().await).unwrap().starts_with("GET /status HTTP/1.1\r\n")); - let remote = owner.call(send(&id, PluginHttpMethod::Get, "status-v1")).await.unwrap(); + assert!( + String::from_utf8(peer.observed().await) + .unwrap() + .starts_with("GET /status HTTP/1.1\r\n") + ); + let remote = owner + .call(send(&id, PluginHttpMethod::Get, "status-v1")) + .await + .unwrap(); let embedded = f.owner(Some(peer.transport())).await; - assert_eq!(embedded.call(send(&id, PluginHttpMethod::Get, "status-v1")).await.unwrap(), remote); + assert_eq!( + embedded + .call(send(&id, PluginHttpMethod::Get, "status-v1")) + .await + .unwrap(), + remote + ); assert_eq!(f.count("plugin_effect_receipts"), 1); f.deny(); - assert_denied(owner.call(send(&id, PluginHttpMethod::Get, "status-v1")).await); - assert_denied(embedded.call(send(&id, PluginHttpMethod::Get, "status-v1")).await); + assert_denied( + owner + .call(send(&id, PluginHttpMethod::Get, "status-v1")) + .await, + ); + assert_denied( + embedded + .call(send(&id, PluginHttpMethod::Get, "status-v1")) + .await, + ); assert_eq!(peer.connection_count(), 1); - server.0.abort(); let _ = (&mut server.0).await; + server.0.abort(); + let _ = (&mut server.0).await; + f.shutdown().await; +} + +async fn remote_owner_for( + f: &Fixture, + worker_id: &str, + workspace: &str, +) -> (Arc, ServerTask) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let router = build_router(f.api.clone()); + let task = ServerTask(tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + })); + let client = server_api::ServerApiClient::builder(format!("http://{address}")) + .unwrap() + .client( + reqwest::Client::builder() + .no_proxy() + .timeout(std::time::Duration::from_secs(10)) + .build() + .unwrap(), + ) + .authorizer(SignedAuthorizer { + signer: worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&f.identity), + worker_id: worker_id.into(), + }) + .build() + .unwrap(); + ( + Arc::new(yoi_plugin_runtime::owner::RemotePluginHostOwner::new( + Arc::new(client), + workspace.into(), + )), + task, + ) +} + +fn prepare(f: &Fixture) -> PluginHostRequest { + PluginHostRequest::Prepare { + plan_id: f.plan.plan_id.clone(), + installation_id: f.installed.installation.installation_id.clone(), + tools: vec![f.tool.into()], + } +} + +#[tokio::test] +async fn pure_and_https_restore_prepare_rechecks_live_owner_without_consuming_calls_or_receipts() { + for https in [false, true] { + let peer = Peer::new(HOST, vec![]).await; + let f = Fixture::new(https.then(|| peer_binding(&peer))).await; + let embedded = f.owner(Some(peer.transport())).await; + let (remote, mut server) = + remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + for owner in [&embedded, &remote] { + assert_eq!( + owner.call(prepare(&f)).await.unwrap(), + PluginHostResponse::Authorized + ); + owner + .prepare( + f.plan.plan_id.clone(), + f.installed.installation.installation_id.clone(), + vec![f.tool.into()], + ) + .await + .unwrap(); + let mut expanded = prepare(&f); + if let PluginHostRequest::Prepare { tools, .. } = &mut expanded { + tools.push("not-selected".into()); + } + assert_denied(owner.call(expanded).await); + } + f.api + .config_store + .plugin_plan_transition( + TEST_WORKSPACE_ID, + &f.plan.plan_id, + PlanPinState::Active, + PlanPinState::Restorable, + ) + .unwrap(); + for owner in [&embedded, &remote] { + assert_eq!( + owner.call(prepare(&f)).await.unwrap(), + PluginHostResponse::Authorized + ); + } + f.deny(); + for owner in [&embedded, &remote] { + assert_denied(owner.call(prepare(&f)).await); + } + set_policy( + &f.api, + &f.subject, + &f.installed.installation.installation_id, + true, + ); + f.revoke(); + for owner in [&embedded, &remote] { + assert_denied(owner.call(prepare(&f)).await); + } + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + assert_eq!(peer.connection_count(), 0); + server.0.abort(); + let _ = (&mut server.0).await; + f.shutdown().await; + } +} + +#[tokio::test] +async fn prepare_unavailable_owner_foreign_workspace_and_unsigned_embedded_never_admit_or_consume() +{ + let f = Fixture::new(None).await; + let unsigned = plugin_host::EmbeddedPluginHostOwner::authenticated( + Arc::new(f.api.clone()), + TEST_WORKSPACE_ID.into(), + Arc::new(server_api::client_support::NoAuthorizer), + ); + assert_denied(unsigned.call(prepare(&f)).await); + let (foreign, mut foreign_server) = + remote_owner_for(&f, &f.source.worker_id, "foreign-workspace").await; + assert!(foreign.call(prepare(&f)).await.is_err()); + let embedded = f.owner(None).await; + let (remote, mut server) = remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + f.api + .config_store + .with_conn(|conn| { + conn.execute_batch("DROP TABLE plugin_execution_plans")?; + Ok(()) + }) + .unwrap(); + for owner in [&embedded, &remote] { + assert_eq!( + owner.call(prepare(&f)).await.unwrap_err().code, + PluginHostErrorCode::Unavailable + ); + } + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + server.0.abort(); + let _ = (&mut server.0).await; + foreign_server.0.abort(); + let _ = (&mut foreign_server.0).await; + f.shutdown().await; +} + +#[tokio::test] +async fn canonical_reserved_child_identity_and_prepare_match_live_owner_and_fence_lost_delegation() +{ + use server_api::authorization::WorkerDelegationRequest; + let f = Fixture::new(None).await; + let store = open_subjektiv_store(&f.api).unwrap(); + let previous = store.subject_permissions(&f.subject).unwrap().unwrap(); + store + .set_subject_permissions( + &f.subject, + &SubjectPermissionsUpdate { + operation_id: "delegate-plugin".into(), + expected_change_id: Some(previous.change_id), + execution: previous.execution.clone(), + delegation: previous.execution.clone(), + }, + ) + .unwrap(); + let request = WorkerDelegationRequest { + permissions: previous.execution, + redelegation: vec![], + }; + let grant = subject_authorization::prepare_delegation( + &f.api, + &f.source, + Some(&request), + &f.source.runtime_id, + PROFILE, + "profile-digest", + None, + false, + ) + .unwrap() + .unwrap(); + let memory = f + .api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let reserved = f + .api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + &f.source.runtime_id, + "prepare-child", + "sha256:reserved-child", + None, + &memory, + ) + .unwrap(); + assert!( + f.api + .plugin_reserved_execution_identity(&f.source.runtime_id, &reserved, Some(&grant)) + .is_err(), + "unbound delegation must not become an ordinary Worker" + ); + subject_authorization::bind_reserved_grant(&f.api, &grant, &reserved.worker_id.to_string()) + .unwrap(); + let child = RuntimeWorkerRef::new(&f.source.runtime_id, reserved.worker_id.to_string()); + assert!(f.api.runtime.worker(&child).is_err()); + let identity = f + .api + .plugin_reserved_execution_identity(&child.runtime_id, &reserved, Some(&grant)) + .unwrap(); + let mut forged = reserved.clone(); + forged.create_fingerprint.push_str("different"); + assert!( + f.api + .plugin_reserved_execution_identity(&child.runtime_id, &forged, Some(&grant)) + .is_err() + ); + let state = f + .api + .config_store + .load_workspace_config(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let plan = f + .api + .config_store + .with_immediate_transaction(|tx| { + let plan = crate::plugin_settings::resolve_worker_plugin_plan( + tx, + TEST_WORKSPACE_ID, + &state.snapshot.digest, + PROFILE, + identity.clone(), + )?; + crate::plugin_plan::plugin_plan_create(tx, "fixture-host", "child-plan", &plan) + }) + .unwrap(); + let embedded = Arc::new(plugin_host::EmbeddedPluginHostOwner::authenticated( + Arc::new(f.api.clone()), + TEST_WORKSPACE_ID.into(), + Arc::new(SignedAuthorizer { + signer: worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&f.identity), + worker_id: child.worker_id.clone(), + }), + )) as Arc; + let (remote, mut server) = remote_owner_for(&f, &child.worker_id, TEST_WORKSPACE_ID).await; + let preflight = PluginHostRequest::Prepare { + plan_id: plan.plan_id, + installation_id: f.installed.installation.installation_id.clone(), + tools: vec![f.tool.into()], + }; + for owner in [&embedded, &remote] { + assert_eq!( + owner.call(preflight.clone()).await.unwrap(), + PluginHostResponse::Authorized + ); + } + // Only unrelated Worker execution/summary is substituted by this fixture. + // The identity, reservation, signed proof and delegation checks stay real. + let mut summary = f.api.runtime.worker(&f.source).unwrap(); + summary.worker.worker_id = child.worker_id.clone(); + summary.singleton_key = None; + f.runtime.workers.lock().unwrap().push(summary); + assert_eq!( + f.api.plugin_execution_identity(&child).await.unwrap(), + identity + ); + f.deny(); + assert!( + f.api + .plugin_reserved_execution_identity(&child.runtime_id, &reserved, Some(&grant)) + .is_err() + ); + for owner in [&embedded, &remote] { + assert_denied(owner.call(preflight.clone()).await); + } + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); + server.0.abort(); + let _ = (&mut server.0).await; + f.shutdown().await; +} + +#[tokio::test] +async fn reserved_job_without_canonical_job_binding_cannot_fall_back_to_ordinary_identity() { + let f = Fixture::new(None).await; + let memory = f + .api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let reserved = f + .api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + &f.source.runtime_id, + &crate::backend_job::allocation_key("not-bound-attempt"), + "sha256:unbound-job", + None, + &memory, + ) + .unwrap(); + assert!( + f.api + .plugin_reserved_execution_identity(&f.source.runtime_id, &reserved, None) + .is_err() + ); + assert_eq!(f.count("plugin_invocations"), 0); + assert_eq!(f.count("plugin_effect_receipts"), 0); f.shutdown().await; } diff --git a/openapi/server-api.json b/openapi/server-api.json index c94b247d8..914cb5a28 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -7763,6 +7763,35 @@ }, "PluginHostRequest": { "oneOf": [ + { + "additionalProperties": false, + "description": "Fresh export/restore admission only; issues no invocation or effect identity.", + "properties": { + "installation_id": { + "type": "string" + }, + "kind": { + "const": "prepare", + "type": "string" + }, + "plan_id": { + "type": "string" + }, + "tools": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "required": [ + "kind", + "plan_id", + "installation_id", + "tools" + ], + "type": "object" + }, { "additionalProperties": false, "properties": { @@ -19818,7 +19847,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55" + "x-yoi-source-digest": "sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index cbb64ae82..e9fec1102 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 00fef6b0c..37d4c5a9a 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 6f08b8bd3..3a5a024cb 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 7a5eefb9d..3b3f30934 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/workdir-api.ts b/web/workspace/src/lib/generated/workdir-api.ts index ed1a8e853..5fe02d27e 100644 --- a/web/workspace/src/lib/generated/workdir-api.ts +++ b/web/workspace/src/lib/generated/workdir-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-workdir-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/workdir-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index f7b406e6e..d17adc8e8 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:d0830733128babb97634d7a55273591fd4b3b5afa5f038a983999e9192c1af55) +// Input contract: 0.1.0 (sha256:63f6a6713e8a5d5e9aaf9e75482ec9271ffedff394ca1f32d19210929723ea80) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check