diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 6e4243cc8..12c56a6b1 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -447,6 +447,9 @@ pub struct CreateWorkerRequest { pub workdir_attachment_requests: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub workdir_attachments: Vec, + /// Enables the optional Runtime-local peer observation provider, not operation + /// authority. Scoped Backend observation follows the Worker Feature and live + /// operation/control checks, including when this saved flag is false. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub worker_observation_enabled: bool, #[serde(default, skip_serializing_if = "Vec::is_empty")] diff --git a/crates/server-api/src/authorization.rs b/crates/server-api/src/authorization.rs index bfc418c1e..ed7fa5fee 100644 --- a/crates/server-api/src/authorization.rs +++ b/crates/server-api/src/authorization.rs @@ -42,8 +42,10 @@ //! terminal confirmation and does not undo completed effects. //! //! A child receives no Subject body identity, singleton, whole Memory or public -//! conversation-post authority. `worker.result` is explicit notify authority to a -//! named origin Worker, not arbitrary control or publication. Chat delivery and +//! conversation-post authority. `worker.messaging` permits the messaging operation; +//! the Worker Feature's Backend control grant separately limits its target and direction. +//! Stored `worker.result` and `worker.notify` names are read compatibility only, not +//! result publication, arbitrary control or current request names. Chat delivery and //! conversation correlation belong to the caller integration, not this contract. //! Unset existing Subject policies deny new Subject execution/delegation; ordinary //! non-Subject Workers explicitly retain existing Backend/provider contracts. @@ -115,6 +117,11 @@ pub enum PermissionResource { path: String, recursive: bool, }, + /// Authorizes operation use, not a target Worker or a Workspace collection. + /// The Worker Feature must still enforce its directional target control grant. + WorkerOperation { + workspace_id: String, + }, Worker { runtime_id: String, worker_id: String, @@ -186,6 +193,9 @@ impl<'de> Deserialize<'de> for PermissionResource { path: String, recursive: bool, }, + WorkerOperation { + workspace_id: String, + }, Worker { runtime_id: String, worker_id: String, @@ -285,6 +295,7 @@ impl<'de> Deserialize<'de> for PermissionResource { path, recursive, }, + Wire::WorkerOperation { workspace_id } => Self::WorkerOperation { workspace_id }, Wire::Worker { runtime_id, worker_id, @@ -310,6 +321,7 @@ impl PermissionResource { | Self::Repository { workspace_id, .. } | Self::WorkdirAttachment { workspace_id, .. } | Self::WorkspaceConfig { workspace_id, .. } + | Self::WorkerOperation { workspace_id } | Self::Workspace { workspace_id } => Some(workspace_id), _ => None, } @@ -414,7 +426,9 @@ impl PermissionResource { validate_authorization_identifier(name) } Self::Feature => Ok(()), - Self::Workspace { workspace_id } => validate_authorization_identifier(workspace_id), + Self::Workspace { workspace_id } | Self::WorkerOperation { workspace_id } => { + validate_authorization_identifier(workspace_id) + } Self::Workdir { working_directory_id, path, @@ -567,6 +581,10 @@ impl PermissionResource { } (Self::Feature, Self::Feature) => true, (Self::Workspace { workspace_id: a }, Self::Workspace { workspace_id: b }) => a == b, + ( + Self::WorkerOperation { workspace_id: a }, + Self::WorkerOperation { workspace_id: b }, + ) => a == b, ( Self::Worker { runtime_id: ar, @@ -605,6 +623,14 @@ impl PermissionResource { } } +/// Stored scopes retain their original resources and exact operation names. +/// +/// A legacy Worker read adapter may use [`canonical_worker_operation`] to project +/// operation use, but must retain the original target separately for the Worker +/// Feature's control-grant check, including direction, revocation and delegation +/// limits. Replacing an old Worker resource with WorkerOperation and discarding +/// that target would broaden authority. Neither validation nor `covers` performs +/// that conversion. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(deny_unknown_fields)] @@ -620,6 +646,14 @@ impl ScopedPermission { if matches!(self.resource, PermissionResource::Feature) && self.operation.contains('.') { return Err("Feature resources authorize activation names only, not operations".into()); } + if matches!(self.resource, PermissionResource::WorkerOperation { .. }) + && canonical_worker_operation(&self.operation) != Some(self.operation.as_str()) + { + return Err( + "WorkerOperation resources authorize only exact current Worker control operations" + .into(), + ); + } Ok(()) } @@ -691,6 +725,18 @@ pub struct OperationAuthorizationRequest { impl OperationAuthorizationRequest { pub fn validate(&self) -> Result<(), String> { + if let Some(operation) = canonical_worker_operation(&self.operation) { + if operation != self.operation { + return Err( + "legacy Worker message names are storage-only; request worker.messaging".into(), + ); + } + if !matches!(self.resource, PermissionResource::WorkerOperation { .. }) { + return Err( + "current Worker control requests require a WorkerOperation resource".into(), + ); + } + } ScopedPermission { operation: self.operation.clone(), resource: self.resource.clone(), @@ -736,6 +782,26 @@ pub struct BoundOperationAuthorizationRequest { pub target: BoundAuthorizationTarget, } +/// Maps exact runtime Worker control names to the current operation-use contract. +/// +/// `worker.notify` and `worker.result` are accepted here only to read saved +/// policy/grants; current requests must use `worker.messaging`. This helper does +/// not grant authority or convert resources: adapters must preserve legacy Worker +/// target restrictions separately. Spawn, discovery/list and Internal SubWorker +/// operations are not part of this contract. Unknown names return `None`. +pub fn canonical_worker_operation(operation: &str) -> Option<&'static str> { + match operation { + "worker.messaging" | "worker.notify" | "worker.result" => Some("worker.messaging"), + "worker.send_input" => Some("worker.send_input"), + "worker.cancel" => Some("worker.cancel"), + "worker.stop" => Some("worker.stop"), + "worker.restore" => Some("worker.restore"), + "worker.remove" => Some("worker.remove"), + "worker.observe" => Some("worker.observe"), + _ => None, + } +} + pub fn validate_operation_name(value: &str) -> Result<(), String> { if value.is_empty() || value.len() > 256 @@ -898,6 +964,220 @@ mod tests { ); } + #[test] + fn worker_operation_wire_and_schema_require_only_workspace_identity() { + let wire = serde_json::json!({"kind":"worker_operation","workspace_id":"ws"}); + let resource: PermissionResource = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(serde_json::to_value(&resource).unwrap(), wire); + assert_eq!(resource.workspace_id(), Some("ws")); + assert!(resource.validate().is_ok()); + for invalid in [ + serde_json::json!({"kind":"worker_operation"}), + serde_json::json!({"kind":"worker_operation","workspace_id":"ws","worker_id":"w"}), + serde_json::json!({"kind":"worker_operation","workspace_id":"ws","runtime_id":"r"}), + ] { + assert!(serde_json::from_value::(invalid).is_err()); + } + for workspace_id in ["", " ws", "ws\n", "*", &"x".repeat(257)] { + let resource: PermissionResource = serde_json::from_value( + serde_json::json!({"kind":"worker_operation","workspace_id":workspace_id}), + ) + .unwrap(); + assert!(resource.validate().is_err(), "{workspace_id:?}"); + assert!(!resource.covers(&resource)); + } + let schema = serde_json::to_value(schemars::schema_for!(PermissionResource)).unwrap(); + let variant = schema["oneOf"] + .as_array() + .unwrap() + .iter() + .find(|variant| variant["properties"]["kind"]["const"] == "worker_operation") + .unwrap(); + assert_eq!(variant["properties"]["workspace_id"]["type"], "string"); + assert_eq!( + variant["required"], + serde_json::json!(["kind", "workspace_id"]) + ); + assert_eq!(variant["additionalProperties"], false); + } + + #[test] + fn worker_operation_containment_never_crosses_workspace_or_resource_kind() { + let resource = PermissionResource::WorkerOperation { + workspace_id: "ws".into(), + }; + assert!(resource.covers(&resource)); + for other in [ + PermissionResource::WorkerOperation { + workspace_id: "other".into(), + }, + PermissionResource::Workspace { + workspace_id: "ws".into(), + }, + PermissionResource::Worker { + runtime_id: "r".into(), + worker_id: "w".into(), + }, + PermissionResource::SubWorker { + runtime_id: "r".into(), + worker_id: "w".into(), + name: "child".into(), + }, + ] { + assert!(!resource.covers(&other), "{other:?}"); + assert!(!other.covers(&resource), "{other:?}"); + } + } + + #[test] + fn worker_operation_scopes_accept_only_seven_exact_current_operations() { + let resource = PermissionResource::WorkerOperation { + workspace_id: "ws".into(), + }; + for operation in [ + "worker.messaging", + "worker.send_input", + "worker.cancel", + "worker.stop", + "worker.restore", + "worker.remove", + "worker.observe", + ] { + assert_eq!(canonical_worker_operation(operation), Some(operation)); + let permission = ScopedPermission { + operation: operation.into(), + resource: resource.clone(), + }; + assert!(permission.validate().is_ok(), "{operation}"); + assert!(permission.covers(&permission)); + assert!( + OperationAuthorizationRequest { + operation: operation.into(), + resource: resource.clone(), + } + .validate() + .is_ok() + ); + for old_resource in [ + PermissionResource::Worker { + runtime_id: "r".into(), + worker_id: "w".into(), + }, + PermissionResource::Workspace { + workspace_id: "ws".into(), + }, + PermissionResource::SubWorker { + runtime_id: "r".into(), + worker_id: "w".into(), + name: "child".into(), + }, + ] { + assert!( + OperationAuthorizationRequest { + operation: operation.into(), + resource: old_resource, + } + .validate() + .is_err() + ); + } + let descendant = format!("{operation}.extra"); + assert_eq!(canonical_worker_operation(&descendant), None); + assert!( + ScopedPermission { + operation: descendant, + resource: resource.clone(), + } + .validate() + .is_err() + ); + } + for operation in [ + "worker", + "worker.spawn", + "worker.lifecycle", + "worker.list", + "worker.input", + "subworker.stop", + "sub_worker.stop", + "ticket.read", + "worker.notify", + "worker.result", + ] { + assert!( + ScopedPermission { + operation: operation.into(), + resource: resource.clone(), + } + .validate() + .is_err(), + "{operation}" + ); + if !matches!(operation, "worker.notify" | "worker.result") { + assert_eq!(canonical_worker_operation(operation), None); + } + } + } + + #[test] + fn legacy_worker_message_storage_stays_exact_and_cannot_authorize_current_requests() { + let current = ScopedPermission { + operation: "worker.messaging".into(), + resource: PermissionResource::WorkerOperation { + workspace_id: "ws".into(), + }, + }; + for operation in ["worker.notify", "worker.result"] { + assert_eq!( + canonical_worker_operation(operation), + Some("worker.messaging") + ); + let wire = serde_json::json!({ + "operation":operation, + "resource":{"kind":"worker","runtime_id":"r","worker_id":"w"} + }); + let stored: ScopedPermission = serde_json::from_value(wire.clone()).unwrap(); + assert!(stored.validate().is_ok()); + assert_eq!(serde_json::to_value(&stored).unwrap(), wire); + assert!(stored.covers(&stored)); + assert!(!stored.covers(¤t)); + assert!(!current.covers(&stored)); + let renamed = ScopedPermission { + operation: "worker.messaging".into(), + resource: stored.resource.clone(), + }; + assert!(!stored.covers(&renamed)); + assert!(!renamed.covers(&stored)); + for resource in [ + stored.resource, + current.resource.clone(), + PermissionResource::Workspace { + workspace_id: "ws".into(), + }, + ] { + assert!( + OperationAuthorizationRequest { + operation: operation.into(), + resource, + } + .validate() + .is_err() + ); + } + } + // Creation and discovery remain outside runtime control operation-use scopes. + assert!( + OperationAuthorizationRequest { + operation: "worker.lifecycle".into(), + resource: PermissionResource::Workspace { + workspace_id: "ws".into() + }, + } + .validate() + .is_ok() + ); + } + #[test] fn authorization_response_carries_matching_scope_and_defaults_legacy_absence() { let legacy: OperationAuthorizationResponse = diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 01cca342f..5c95047bb 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -1781,6 +1781,21 @@ pub trait ServerApi { #[extension] context: ServerRequestContext, #[path] workspace_id: String, ) -> Result; + #[get( + "/api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/permissions/{permission}", + status = 204, + error_status = 400, + additional_error_statuses = [401, 403, 404, 500], + openapi = false, + )] + async fn worker_control_authorize( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[path] runtime_id: String, + #[path] worker_id: String, + #[path] permission: String, + ) -> Result<(), RepositoryApiError>; #[post( "/api/w/{workspace_id}/worker-control/workers", status = 200, @@ -14914,6 +14929,7 @@ mod openapi_artifact_tests { "merge_request_review_submit", "merge_request_reviewer_child_register", "runtime_resource_fetch", + "worker_control_authorize", "worker_control_cancel", "worker_control_input", "worker_control_list", diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 5e9c883ec..f6bb365bb 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -284,8 +284,9 @@ pub struct CreateWorkerRequest { pub workdir_attachment_requests: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub workdir_attachments: Vec, - /// Backend-only feature enablement. Grants still define local Runtime peers; - /// the Workspace provider reauthorizes its dynamic set per operation. + /// Backend-only enablement of the optional Runtime-local peer provider. + /// Scoped Backend observation is composed by the Worker Feature, independently + /// of this saved flag, and rechecks operation use plus current target control. #[serde(default, skip_serializing_if = "is_false")] pub worker_observation_enabled: bool, /// Backend-authored, bounded peer session grants. Runtime revalidates each diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index ea0a56011..df85e3602 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -143,9 +143,6 @@ use workdir::{ #[cfg(test)] use worker::WorkerController; use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession; -use worker::feature::builtin::{ - CompositeWorkerObservationProvider, WorkspaceClientWorkerObservationProvider, -}; use worker::feature::contracts::{ WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject, WorkerObservationSubjectRef, WorkerSessionCapture, @@ -2161,20 +2158,15 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { if let (Some(runtime_id), Some(workspace_id)) = (observation_runtime_id, observation_workspace_id.clone()) && observation_enabled + && !observation_grants.is_empty() { - let mut providers: Vec> = vec![Arc::new( - WorkspaceClientWorkerObservationProvider::new(worker.workspace_client_handle()), - )]; - if !observation_grants.is_empty() { - providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider { + worker.bind_worker_observation_provider(Some(Arc::new( + RuntimeGrantedWorkerObservationProvider { runtime_id, workspace_id, grants: observation_grants.into_iter().take(100).collect(), hub: self.observation_hub.clone(), - })); - } - worker.bind_worker_observation_provider(Some(Arc::new( - CompositeWorkerObservationProvider::new(providers), + }, ))); } if flow_transition_enabled { @@ -2877,30 +2869,24 @@ impl ProfileRuntimeWorkerFactory { .expect("retained restore owner") .sessions .push(workdir_sessions.clone()); - let mut providers: Vec> = vec![Arc::new( - WorkspaceClientWorkerObservationProvider::new(workspace_client.clone()), - )]; let observation_provider = if let (Some(runtime_id), Some(api)) = ( self.runtime_id.clone(), request.request.workspace_api.as_ref(), ) && request.request.worker_observation_enabled + && !request.request.worker_observation_grants.is_empty() { - if !request.request.worker_observation_grants.is_empty() { - providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider { - runtime_id, - workspace_id: api.workspace_id.clone(), - grants: request - .request - .worker_observation_grants - .iter() - .take(100) - .cloned() - .collect(), - hub: self.observation_hub.clone(), - })); - } - Some(Arc::new(CompositeWorkerObservationProvider::new(providers)) - as Arc) + Some(Arc::new(RuntimeGrantedWorkerObservationProvider { + runtime_id, + workspace_id: api.workspace_id.clone(), + grants: request + .request + .worker_observation_grants + .iter() + .take(100) + .cloned() + .collect(), + hub: self.observation_hub.clone(), + }) as Arc) } else { None }; diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index 57f2e1f00..ec81e3748 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -705,6 +705,20 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { client.worker_control_list(workspace_id).await, ) } + WorkspaceServerOperation::WorkerControlAuthorize { + runtime_id, + worker_id, + permission, + } => generated_workspace_response( + client + .worker_control_authorize( + workspace_id, + runtime_id, + worker_id, + permission, + ) + .await, + ), WorkspaceServerOperation::WorkerControlSpawn(request) => { generated_workspace_response( client.worker_control_spawn(workspace_id, request).await, diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 3bc7c180d..dfc9fd7cb 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2045,7 +2045,7 @@ where crate::feature::builtin::objective::composition::compose( &mut feature_registry, feature_config.objective.enabled, - workspace_client, + workspace_client.clone(), objective_namespace.map(|route| (route, wip_permissions.clone())), )?; @@ -2066,6 +2066,15 @@ where ) { observation_providers.push(provider); } + // A scoped Backend provider belongs to the Worker Feature's composition, + // including ordinary spawn and restore. Runtime's optional local peer + // provider is not its enablement or authority: every actual capture still + // requires current operation use and the target's Backend control grant. + if worker_enabled { + observation_providers.push(Arc::new( + crate::feature::builtin::worker_observation::WorkspaceClientWorkerObservationProvider::new(workspace_client), + )); + } if let Some(provider) = host_worker_observation_provider { observation_providers.push(provider); } @@ -4141,6 +4150,82 @@ mod tests { } } + #[tokio::test] + async fn scoped_observation_is_composed_only_with_runtime_worker_service_without_startup_access() + { + for enabled in [false, true] { + let root = TempDir::new().unwrap(); + let mut manifest = manifest::WorkerManifest::from_toml( + "[worker]\nname = 'observation-wiring'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n" + ).unwrap(); + manifest.feature.worker.enabled = enabled; + manifest.feature.sub_worker.enabled = false; + manifest.feature.workdir_catalog.enabled = false; + let model = ScriptedClient::new(vec![]); + let engine = + Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated( + model, + ); + let store = session_store::CombinedStore::new( + session_store::FsStore::new(root.path().join("sessions")).unwrap(), + session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(), + ); + let operations = Arc::new(std::sync::Mutex::new(Vec::new())); + let host = Arc::new(DriveWorkspaceHost { + available: true, + workspace: Some("workspace-1"), + granted: false, + operations: operations.clone(), + }); + let mut worker = Worker::new( + manifest, + engine, + store, + crate::worker::WorkerWorkspaceContext::with_client( + Some(crate::worker::WorkspaceId::new("workspace-1").unwrap()), + host, + ), + crate::worker::WorkerFilesystemAuthority::None, + manifest::Scope::empty(), + ) + .await + .unwrap(); + worker.set_system_prompt_template( + crate::SystemPromptTemplate::parse( + "default", + crate::PromptCatalogSource::builtins_only(), + ) + .unwrap(), + ); + let prepared = WorkerController::initialize(worker, root.path(), None) + .await + .unwrap(); + let Event::Snapshot { greeting, .. } = + crate::ipc::protocol_session::subscribe_worker_protocol_session( + &prepared.handle, + ) + .snapshot_event + else { + panic!("initial committed greeting"); + }; + for name in [ + "ViewSessionOverview", + "SearchSessionEntries", + "ReadSessionEntry", + ] { + assert_eq!( + greeting.tools.iter().any(|tool| tool == name), + enabled, + "Worker service configuration must determine observation publication, not client availability alone" + ); + } + assert!( + operations.lock().unwrap().is_empty(), + "provider composition may not authorize or read protected resources" + ); + } + } + #[tokio::test] async fn drive_controller_registers_without_authorization_and_checks_each_access_in_both_modes() { diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 078e91741..3b5ba6c51 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -41,22 +41,18 @@ struct WorkspaceWorkerControlService { } impl WorkspaceWorkerControlService { - fn authorize_runtime_worker( - &self, - operation: &str, - runtime_id: &str, - worker_id: &str, - ) -> Result<(), WorkspaceClientError> { + fn authorize_runtime_operation(&self, operation: &str) -> Result<(), WorkspaceClientError> { if !self.runtime_worker_control { return Err(WorkspaceClientError::Unavailable( "Runtime Worker control is not enabled for this Worker".into(), )); } + // Common authorization permits using the operation. The Backend separately + // checks the authenticated caller's control grant for the actual target. self.client.authorize_operation( operation, - &PermissionResource::Worker { - runtime_id: runtime_id.into(), - worker_id: worker_id.into(), + &PermissionResource::WorkerOperation { + workspace_id: self.workspace_id.clone(), }, ) } @@ -201,14 +197,10 @@ impl WorkerControlService for WorkspaceWorkerControlService { runtime_id, worker_id, } => { - self.authorize_runtime_worker( - match kind { - WorkerControlInputKind::User => "worker.send_input", - WorkerControlInputKind::Notify => "worker.notify", - }, - &runtime_id, - &worker_id, - )?; + self.authorize_runtime_operation(match kind { + WorkerControlInputKind::User => "worker.send_input", + WorkerControlInputKind::Notify => "worker.messaging", + })?; let kind = match kind { WorkerControlInputKind::User => "user", WorkerControlInputKind::Notify => "notify", @@ -272,14 +264,10 @@ impl WorkerControlService for WorkspaceWorkerControlService { runtime_id, worker_id, } => { - self.authorize_runtime_worker( - match kind { - WorkerControlStopKind::Cancel => "worker.cancel", - WorkerControlStopKind::Stop => "worker.stop", - }, - &runtime_id, - &worker_id, - )?; + self.authorize_runtime_operation(match kind { + WorkerControlStopKind::Cancel => "worker.cancel", + WorkerControlStopKind::Stop => "worker.stop", + })?; let operation = match kind { WorkerControlStopKind::Cancel => { crate::worker::WorkspaceServerOperation::WorkerControlCancel { @@ -360,7 +348,7 @@ impl WorkerControlService for WorkspaceWorkerControlService { "Runtime Worker control is not enabled for this Worker".to_string(), )); } - self.authorize_runtime_worker("worker.restore", &runtime_id, &worker_id)?; + self.authorize_runtime_operation("worker.restore")?; self.client.execute_server_operation( crate::worker::WorkspaceServerOperation::WorkerControlRestore { runtime_id, @@ -381,7 +369,7 @@ impl WorkerControlService for WorkspaceWorkerControlService { "Runtime Worker control is not enabled for this Worker".to_string(), )); } - self.authorize_runtime_worker("worker.remove", runtime_id, worker_id)?; + self.authorize_runtime_operation("worker.remove")?; self.client .execute_worker_remove(runtime_id, worker_id, reason) } @@ -415,50 +403,26 @@ impl WorkerControlService for WorkspaceWorkerControlService { "Runtime Worker control is not enabled for this Worker".to_string(), )); } - self.authorize_runtime_worker( - &format!("worker.{permission}"), - runtime_id, - worker_id, - )?; + let operation = if permission == "notify" { + "worker.messaging".to_owned() + } else { + format!("worker.{permission}") + }; + self.authorize_runtime_operation(&operation)?; let response = self.client.execute_server_operation( - crate::worker::WorkspaceServerOperation::WorkerControlList, + crate::worker::WorkspaceServerOperation::WorkerControlAuthorize { + runtime_id: runtime_id.clone(), + worker_id: worker_id.clone(), + permission: permission.into(), + }, )?; - if !response.is_success() { - return Err(WorkspaceClientError::Request(format!( - "Workspace control request returned {}: {}", - response.status, response.body - ))); - } - let body: serde_json::Value = - serde_json::from_str(&response.body).map_err(|error| { - WorkspaceClientError::Request(format!( - "invalid Workspace control response: {error}" - )) - })?; - let granted = body - .get("items") - .and_then(serde_json::Value::as_array) - .into_iter() - .flatten() - .any(|item| { - item.get("subject") - .and_then(|value| value.get("runtime_id")) - == Some(&serde_json::Value::String(runtime_id.clone())) - && item.get("subject").and_then(|value| value.get("worker_id")) - == Some(&serde_json::Value::String(worker_id.clone())) - && item - .get("permissions") - .and_then(serde_json::Value::as_array) - .is_some_and(|permissions| { - permissions.iter().any(|candidate| candidate == permission) - }) - }); - if granted { + if response.is_success() { Ok(()) } else { - Err(WorkspaceClientError::Request( - "unknown Worker or permission not granted".to_string(), - )) + Err(WorkspaceClientError::Request(format!( + "Worker target permission check returned {}: {}", + response.status, response.body + ))) } } } @@ -1105,6 +1069,32 @@ fn tool_output( response.status, response.body ))); } + if operation == WorkerOperation::Notify { + let notified: server_api::RuntimeWorkerInputResult = serde_json::from_str(&response.body) + .map_err(|error| { + ToolError::ExecutionFailed(format!("invalid WorkerNotify response: {error}")) + })?; + // HTTP transport success is not Runtime acceptance. Preserve uncertain + // diagnostics in the error; callers must inspect history, not resend or + // implicitly Restore a stopped destination. + if notified.state != server_api::WorkerOperationState::Accepted + || notified + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "worker_input_outcome_unknown") + { + return Err(ToolError::ExecutionFailed(format!( + "WorkerNotify was not confirmed accepted (no automatic retry or Restore): {}", + response.body + ))); + } + return Ok(ToolOutput { + summary: "WorkerNotify accepted; model processing and work completion are unconfirmed" + .into(), + content: Some(response.body), + attachments: Vec::new(), + }); + } if operation == WorkerOperation::Restore { let restored: server_api::WorkerRestoreResponse = serde_json::from_str(&response.body) .map_err(|error| { @@ -1287,6 +1277,8 @@ mod tests { struct RecordingWorkspaceClient { requests: Mutex>, authorizations: Mutex>, + bound_authorizations: + Mutex>, denied: std::sync::atomic::AtomicBool, removals: Mutex>, } @@ -1320,7 +1312,16 @@ mod tests { body: r#"{"scope":null}"#.into(), }); } + if request.path.ends_with("/authorization/bound-check") { + let auth = serde_json::from_str(request.body.as_deref().unwrap()).unwrap(); + self.bound_authorizations.lock().unwrap().push(auth); + return Ok(WorkspaceResponse { + status: 200, + body: r#"{"scope":null}"#.into(), + }); + } let is_worker_list = request.path.ends_with("/worker-control/workers"); + let is_worker_input = request.path.ends_with("/input"); self.requests.lock().unwrap().push(request); Ok(WorkspaceResponse { status: 200, @@ -1332,6 +1333,7 @@ mod tests { "runtime_id": "runtime-a", "worker_id": "worker-a" }, + "permissions": ["observe"], "summary": { "status": "idle" } }, { @@ -1339,11 +1341,14 @@ mod tests { "runtime_id": "runtime-a", "worker_id": "worker-a" }, + "permissions": ["observe"], "summary": { "status": "idle" } } ] }) .to_string() + } else if is_worker_input { + serde_json::json!({"state":"accepted","runtime_id":"runtime-a","worker_id":"worker-a","diagnostics":[]}).to_string() } else { "{}".to_string() }, @@ -1475,12 +1480,48 @@ mod tests { fn test_control(client: Arc) -> Arc { Arc::new(WorkspaceWorkerControlService { client, - workspace_id: "workspace%2Ftest".to_string(), + workspace_id: "workspace/test".to_string(), registry: None, runtime_worker_control: true, }) } + #[test] + fn notify_transport_success_requires_runtime_acceptance_and_preserves_unknown_diagnostics() { + for body in [ + serde_json::json!({"state":"rejected","runtime_id":"r","worker_id":"w","diagnostics":[]}), + serde_json::json!({"state":"unsupported","runtime_id":"r","worker_id":"w","diagnostics":[]}), + serde_json::json!({"state":"rejected","runtime_id":"r","worker_id":"w","diagnostics":[{"code":"worker_input_outcome_unknown","severity":"warning","message":"inspect current session before retrying"}]}), + ] { + let error = tool_output( + WorkerOperation::Notify, + WorkspaceResponse { + status: 200, + body: body.to_string(), + }, + ) + .unwrap_err() + .to_string(); + assert!(error.contains("not confirmed accepted"), "{error}"); + assert!(error.contains("no automatic retry or Restore"), "{error}"); + if !body["diagnostics"].as_array().unwrap().is_empty() { + assert!(error.contains("worker_input_outcome_unknown")); + } + } + assert!( + tool_output( + WorkerOperation::Notify, + WorkspaceResponse { + status: 200, + body: "{}".into() + } + ) + .is_err() + ); + let accepted = tool_output(WorkerOperation::Notify,WorkspaceResponse{status:200,body:serde_json::json!({"state":"accepted","runtime_id":"r","worker_id":"w","diagnostics":[]}).to_string()}).unwrap(); + assert!(accepted.summary.contains("unconfirmed")); + } + #[tokio::test] async fn worker_restore_forwards_exact_observation_and_request_identity() { let client = Arc::new(RecordingWorkspaceClient::default()); @@ -1678,7 +1719,7 @@ mod tests { let client = Arc::new(RecordingWorkspaceClient::default()); let control = WorkspaceWorkerControlService { client: client.clone(), - workspace_id: "workspace%2Ftest".to_string(), + workspace_id: "workspace/test".to_string(), registry: None, runtime_worker_control: true, }; @@ -1714,7 +1755,7 @@ mod tests { let client = Arc::new(RecordingWorkspaceClient::default()); let control = WorkspaceWorkerControlService { client: client.clone(), - workspace_id: "workspace%2Ftest".to_string(), + workspace_id: "workspace/test".to_string(), registry: None, runtime_worker_control: true, }; @@ -1763,7 +1804,7 @@ mod tests { let registry = SpawnedWorkerRegistry::new(runtime_dir); let control: Arc = Arc::new(WorkspaceWorkerControlService { client: client.clone(), - workspace_id: "workspace%2Ftest".to_string(), + workspace_id: "workspace/test".to_string(), registry: Some(registry), runtime_worker_control: false, }); @@ -1831,7 +1872,7 @@ mod tests { let client = Arc::new(RecordingWorkspaceClient::default()); let control: Arc = Arc::new(WorkspaceWorkerControlService { client: client.clone(), - workspace_id: "workspace%2Ftest".to_string(), + workspace_id: "workspace/test".to_string(), registry: None, runtime_worker_control: false, }); @@ -2137,6 +2178,28 @@ mod tests { ) .await .unwrap(); + let check = client + .authorizations + .lock() + .unwrap() + .last() + .cloned() + .unwrap(); + assert_eq!( + check.operation, + match operation { + WorkerOperation::SendInput => "worker.send_input", + WorkerOperation::Notify => "worker.messaging", + WorkerOperation::Cancel => "worker.cancel", + _ => unreachable!(), + } + ); + assert_eq!( + check.resource, + PermissionResource::WorkerOperation { + workspace_id: "workspace/test".into(), + } + ); let request = client.requests.lock().unwrap().last().cloned().unwrap(); assert!(request.path.ends_with(expected_suffix)); assert!(request.path.contains("/worker-control/workers/")); @@ -2325,12 +2388,12 @@ mod tests { }; match op { "worker.list" => control.list_workers().await, - "worker.send_input" | "worker.notify" => { + "worker.send_input" | "worker.messaging" => { control .send_input( subject, "input".into(), - if op == "worker.notify" { + if op == "worker.messaging" { WorkerControlInputKind::Notify } else { WorkerControlInputKind::User @@ -2368,6 +2431,21 @@ mod tests { "target-worker", "requested removal", ), + "worker.observe" => { + control + .ensure_permission( + &WorkerObservationSubjectRef::RuntimeWorker { + runtime_id: "runtime-a".into(), + worker_id: "worker-a".into(), + }, + "observe", + ) + .await?; + Ok(WorkspaceResponse { + status: 200, + body: "{}".into(), + }) + } "worker.create" => { control .spawn_worker(WorkerLifecycleSpawnRequest { @@ -2390,11 +2468,12 @@ mod tests { for op in [ "worker.list", "worker.send_input", - "worker.notify", + "worker.messaging", "worker.cancel", "worker.stop", "worker.restore", "worker.remove", + "worker.observe", "worker.create", ] { let client = Arc::new(RecordingWorkspaceClient::default()); @@ -2432,23 +2511,134 @@ mod tests { check.resource, if matches!(op, "worker.list" | "worker.create") { PermissionResource::Workspace { - workspace_id: if op == "worker.list" { - "workspace%2Ftest" - } else { - "workspace/test" - } - .into(), + workspace_id: "workspace/test".into(), } } else { - PermissionResource::Worker { - runtime_id: "target-runtime".into(), - worker_id: "target-worker".into(), + PermissionResource::WorkerOperation { + workspace_id: "workspace/test".into(), } } ); } } } + // Connection: published Tool registration and dispatch, production typed control + // service and authorization transport; the Host and Backend are fixture boundaries. + #[tokio::test] + async fn published_runtime_tools_reject_missing_operation_grant_without_backend_effects() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let mut pending_tools = Vec::new(); + let mut hooks = crate::HookRegistryBuilder::default(); + let report = crate::feature::FeatureRegistryBuilder::new() + .with_module(manage_worker_feature(client.clone(), None, true)) + .install_into_pending(&mut pending_tools, &mut hooks); + assert!(report.reports[0].installed); + let tools: Vec<_> = pending_tools + .into_iter() + .map(|definition| definition()) + .collect(); + client + .denied + .store(true, std::sync::atomic::Ordering::SeqCst); + for (tool_name, operation, fields) in [ + ( + "WorkerSendInput", + "worker.send_input", + serde_json::json!({"content":"continue"}), + ), + ( + "WorkerNotify", + "worker.messaging", + serde_json::json!({"content":"result ready"}), + ), + ( + "WorkerCancel", + "worker.cancel", + serde_json::json!({"reason":"cancel"}), + ), + ( + "WorkerStop", + "worker.stop", + serde_json::json!({"reason":"stop"}), + ), + ( + "WorkerRestore", + "worker.restore", + serde_json::json!({"expected_observation_token":"observed"}), + ), + ( + "WorkerRemove", + "worker.remove", + serde_json::json!({"reason":"retire"}), + ), + ] { + let (_, tool) = tools + .iter() + .find(|(meta, _)| meta.name == tool_name) + .unwrap(); + for worker_id in ["worker-a", "worker-b"] { + let mut input = fields.clone(); + input["subject"] = serde_json::json!({ + "kind":"runtime_worker", "runtime_id":"runtime-a", "worker_id":worker_id, + }); + let error = tool + .execute(&input.to_string(), ToolExecutionContext::direct()) + .await + .unwrap_err(); + assert!( + error.to_string().contains("authorization denied"), + "{tool_name}: {error}" + ); + let checks = client.authorizations.lock().unwrap(); + let check = checks.last().unwrap(); + assert_eq!(check.operation, operation); + assert_eq!( + check.resource, + PermissionResource::WorkerOperation { + workspace_id: "workspace/test".into(), + } + ); + assert!(client.requests.lock().unwrap().is_empty(), "{tool_name}"); + assert!(client.removals.lock().unwrap().is_empty(), "{tool_name}"); + } + } + assert_eq!(client.authorizations.lock().unwrap().len(), 12); + } + + #[test] + fn runtime_operation_denial_does_not_replace_host_bound_subworker_authorization() { + let client = Arc::new(RecordingWorkspaceClient::default()); + client + .denied + .store(true, std::sync::atomic::Ordering::SeqCst); + let control = WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace/test".into(), + registry: None, + runtime_worker_control: true, + }; + for operation in [ + "subworker.send_input", + "subworker.stop", + "subworker.observe", + ] { + control.authorize_subworker(operation, "child").unwrap(); + let checks = client.bound_authorizations.lock().unwrap(); + let check = checks.last().unwrap(); + assert_eq!(check.operation, operation); + assert_eq!( + check.target, + server_api::authorization::BoundAuthorizationTarget::SubWorker { + name: "child".into(), + } + ); + } + assert_eq!(client.bound_authorizations.lock().unwrap().len(), 3); + assert!(client.authorizations.lock().unwrap().is_empty()); + assert!(client.requests.lock().unwrap().is_empty()); + assert!(client.removals.lock().unwrap().is_empty()); + } + #[test] fn only_explicit_standalone_retains_local_subworker_authority_without_host() { for workspace_id in [None, Some("bound-workspace".to_string())] { diff --git a/crates/worker/src/feature/builtin/worker_observation.rs b/crates/worker/src/feature/builtin/worker_observation.rs index bb740ca69..2682a641f 100644 --- a/crates/worker/src/feature/builtin/worker_observation.rs +++ b/crates/worker/src/feature/builtin/worker_observation.rs @@ -125,14 +125,10 @@ impl WorkerObservationProvider for WorkspaceClientWorkerObservationProvider { ) })?; match subject { - WorkerObservationSubjectRef::RuntimeWorker { - runtime_id, - worker_id, - } => self.client.authorize_operation( + WorkerObservationSubjectRef::RuntimeWorker { .. } => self.client.authorize_operation( "worker.observe", - &server_api::authorization::PermissionResource::Worker { - runtime_id: runtime_id.clone(), - worker_id: worker_id.clone(), + &server_api::authorization::PermissionResource::WorkerOperation { + workspace_id: _workspace_id.into(), }, ), WorkerObservationSubjectRef::SubWorker { name } => self diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 799b67f80..f09399780 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -762,6 +762,11 @@ pub enum WorkspaceServerOperation { result_id: String, }, WorkerControlList, + WorkerControlAuthorize { + runtime_id: String, + worker_id: String, + permission: String, + }, WorkerControlSpawn(server_api::CreateWorkspaceWorkerRequest), WorkerControlInput { runtime_id: String, @@ -959,6 +964,16 @@ fn workspace_server_operation_request( WorkspaceServerOperation::WorkerControlList => Ok(WorkspaceRequest::get(format!( "{base}/worker-control/workers" ))), + WorkspaceServerOperation::WorkerControlAuthorize { + runtime_id, + worker_id, + permission, + } => Ok(WorkspaceRequest::get(format!( + "{base}/worker-control/workers/{}/{}/permissions/{}", + encode_workspace_path_segment(&runtime_id), + encode_workspace_path_segment(&worker_id), + encode_workspace_path_segment(&permission) + ))), WorkspaceServerOperation::WorkerControlSpawn(request) => { workspace_server_json_request(format!("{base}/worker-control/workers"), &request) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 0c7284053..c5aa27ccc 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2066,6 +2066,7 @@ struct WorkerRemovalService { journal_storage: crate::WorkspaceFeatureStorage, journal_registration: crate::RegisteredFeature, journal_dispatch_api: Weak, + worker_authority: subject_authorization::WorkerAuthorityView, store: Arc, host_executors: HostExecutors, runtime: Weak, @@ -2080,6 +2081,7 @@ impl WorkerRemovalService { fn new(api: &WorkspaceApi) -> Self { Self { workspace_id: api.config.workspace_id.clone(), + worker_authority: subject_authorization::WorkerAuthorityView::from(api), journal_storage: api.feature_storage.clone(), journal_registration: api.subjektiv_registration.clone(), journal_dispatch_api: api @@ -2402,23 +2404,14 @@ impl WorkerRemovalService { )); } let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let grant = self - .store - .get_active_worker_control_grant(&self.workspace_id, &controller, &target) - .map_err(|_| "Worker control grant authority is unavailable".to_string())? - .filter(|grant| { - grant - .permissions - .iter() - .any(|permission| permission == "remove") - }); - let Some(grant) = grant else { - return Ok(worker_remove_error_response( - StatusCode::NOT_FOUND, - "unknown_worker", - "The target Worker is not known to the current Worker", - )); - }; + let grant = + match self + .worker_authority + .authorize_control_permission(&controller, &target, "remove") + { + Ok(grant) => grant, + Err(error) => return Ok(worker_control_denied_response(error)), + }; let control_lock = { let mut locks = self .worker_control_locks @@ -2430,17 +2423,15 @@ impl WorkerRemovalService { .clone() }; let _control_guard = control_lock.lock().await; - let still_granted = self - .store - .get_active_worker_control_grant(&self.workspace_id, &controller, &target) - .map_err(|_| "Worker control grant authority is unavailable".to_string())? - .is_some_and(|current| { - current.grant_id == grant.grant_id - && current - .permissions - .iter() - .any(|permission| permission == "remove") - }); + let current = + match self + .worker_authority + .authorize_control_permission(&controller, &target, "remove") + { + Ok(grant) => grant, + Err(error) => return Ok(worker_control_denied_response(error)), + }; + let still_granted = current.grant_id == grant.grant_id; if !still_granted { return Ok(worker_remove_error_response( StatusCode::NOT_FOUND, @@ -7857,6 +7848,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou .merge(server_api::server_api_axum::worker_control_list( service.clone(), )) + .merge(server_api::server_api_axum::worker_control_authorize( + service.clone(), + )) .merge(server_api::server_api_axum::worker_control_spawn( service.clone(), )) @@ -10344,6 +10338,46 @@ impl server_api::ServerApi for ServerApiContractService { Ok(response) } + async fn worker_control_authorize( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + runtime_id: String, + worker_id: String, + permission: String, + ) -> std::result::Result<(), server_api::RepositoryApiError> { + let api = self.workspace_api()?.clone(); + let headers = current_worker_contract_headers(&context)?; + api.host_executors + .host() + .map_err(|error| ApiError::from(error).into_repository_api_error())? + .run(move || -> ApiResult<()> { + validate_workspace_scope(&api, &workspace_id)?; + let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?; + if !matches!( + permission.as_str(), + "observe" | "notify" | "send_input" | "cancel" | "stop" | "restore" | "remove" + ) { + return Err( + Error::InvalidInput("unknown Worker control permission".into()).into(), + ); + } + // This is a current, target-only preflight for providers that read + // locally. It discloses no registry/session data and issues no + // reusable receipt. Actual Backend effects still recheck authority. + authorize_known_worker_permission( + &api, + &workspace_id, + &RuntimeWorkerRef::new(source.runtime_id, source.worker_id), + &RuntimeWorkerRef::new(runtime_id, worker_id), + &permission, + )?; + Ok(()) + }) + .await + .map_err(ApiError::into_repository_api_error) + } + async fn worker_control_spawn( &self, context: server_api::ServerRequestContext, @@ -26814,6 +26848,16 @@ async fn list_known_workers( validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); + subject_authorization::authorize( + &api, + &controller, + &server_api::authorization::OperationAuthorizationRequest { + operation: "worker.list".into(), + resource: server_api::authorization::PermissionResource::Workspace { + workspace_id: path.workspace_id.clone(), + }, + }, + )?; let limit = api.config.max_records.clamp(1, 500); let grants = api.store.list_active_worker_control_grants( &path.workspace_id, @@ -26936,6 +26980,15 @@ async fn spawn_known_worker_authenticated_on_host( context, ) .await?; + // Creation has now committed the real registry row. Keep reply-grant failure + // outside spawn compensation so an exact operation replay repairs only the + // missing grant, without a second Worker or repeated initial Submit. + subject_authorization::persist_reply_grant( + &api, + &controller, + &RuntimeWorkerRef::new(&response.0.runtime_id, &response.0.worker_id), + &scoped_worker_control_operation_id(&controller, &operation_id), + )?; if let Err(error) = api .store .create_worker_control_grant(&WorkerControlGrantRecord { @@ -26980,6 +27033,26 @@ fn worker_control_lock(api: &WorkspaceApi, grant_id: &str) -> Arc worker::WorkspaceResponse { + // Keep removal's existing opaque target-grant refusal; operation-policy + // refusals remain distinct and do not impersonate an unknown target. + if matches!(&error, Error::UnknownWorker { .. }) + || matches!(&error, Error::InvalidInput(message) + if message.starts_with("worker control permission ")) + { + return worker_remove_error_response( + StatusCode::NOT_FOUND, + "unknown_worker", + "The target Worker is not known to the current Worker", + ); + } + worker_remove_error_response( + api_error_status(&error), + "worker_control_denied", + &error.to_string(), + ) +} + fn authorize_known_worker_permission( api: &WorkspaceApi, workspace_id: &str, @@ -26987,54 +27060,13 @@ fn authorize_known_worker_permission( subject: &RuntimeWorkerRef, permission: &str, ) -> Result { - // Child result return is deliberately narrower than control or public chat - // posting. The execution grant names only this origin Worker and only notify. - if permission == "notify" { - if let Some(permissions) = subject_authorization::worker_permissions(api, controller) - .map_err(|error| error.error)? - { - let requested = server_api::authorization::ScopedPermission { - operation: "worker.result".into(), - resource: server_api::authorization::PermissionResource::Worker { - runtime_id: subject.runtime_id.clone(), - worker_id: subject.worker_id.clone(), - }, - }; - if permissions.iter().any(|granted| granted.covers(&requested)) { - return Ok(WorkerControlGrantRecord { - workspace_id: workspace_id.into(), - grant_id: format!( - "execution-result:{}/{}", - controller.runtime_id, controller.worker_id - ), - controller: controller.clone(), - subject: subject.clone(), - relation: "result_return".into(), - origin: "execution_grant".into(), - permissions: vec!["notify".into()], - operation_id: "execution-result".into(), - created_at: String::new(), - revoked_at: None, - }); - } - } + if workspace_id != api.workspace_id() { + return Err(Error::WorkspacePermissionDenied( + "resource belongs to another Workspace".into(), + )); } - let grant = api - .store - .get_active_worker_control_grant(workspace_id, controller, subject)? - .ok_or_else(|| Error::UnknownWorker { - worker: subject.clone(), - })?; - if !grant - .permissions - .iter() - .any(|candidate| candidate == permission) - { - return Err(Error::InvalidInput(format!( - "worker control permission `{permission}` was not granted" - ))); - } - Ok(grant) + subject_authorization::WorkerAuthorityView::from(api) + .authorize_control_permission(controller, subject, permission) } async fn send_known_worker_input( @@ -27153,6 +27185,16 @@ async fn scoped_list_worker_observation_sessions( validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); + subject_authorization::authorize( + &api, + &controller, + &server_api::authorization::OperationAuthorizationRequest { + operation: "worker.observe.list".into(), + resource: server_api::authorization::PermissionResource::Workspace { + workspace_id: path.workspace_id.clone(), + }, + }, + )?; let sessions = api .store .list_active_worker_control_grants(&path.workspace_id, &controller, 100)? @@ -27164,6 +27206,14 @@ async fn scoped_list_worker_observation_sessions( .any(|permission| permission == "observe") }) .filter_map(|grant| { + authorize_known_worker_permission( + &api, + &path.workspace_id, + &controller, + &grant.subject, + "observe", + ) + .ok()?; let worker = api.runtime.worker(&grant.subject).ok()?; if matches!( worker.state.as_str(), diff --git a/crates/workspace-server/src/server/subject_authorization.rs b/crates/workspace-server/src/server/subject_authorization.rs index ba6ba4786..10b3aa2a7 100644 --- a/crates/workspace-server/src/server/subject_authorization.rs +++ b/crates/workspace-server/src/server/subject_authorization.rs @@ -16,7 +16,7 @@ use super::*; use server_api::authorization::{ OperationAuthorizationRequest, PermissionResource, ScopedPermission, SubjectPermissions, - WorkerDelegationRequest, WorkerExecutionGrant, + WorkerDelegationRequest, WorkerExecutionGrant, canonical_worker_operation, }; use std::collections::BTreeSet; @@ -24,123 +24,282 @@ fn denied(message: impl Into) -> ApiError { Error::WorkspacePermissionDenied(message.into()).into() } -fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult> { - let store = open_subjektiv_store(api)?; - let subject = store - .subject(subject_id) - .map_err(subjektiv_store_error)? - .ok_or_else(|| denied("Subject permission source no longer exists"))?; - if subject.state != crate::subjektiv::SubjectState::Active { - return Err(denied("Subject permission source is retired")); +/// Minimal shared authority for Worker control and removal. In particular this +/// must not retain WorkspaceApi or Runtime: the embedded removal dispatcher is +/// itself owned by Runtime. Policy, lineage and body ownership keep one checker. +#[derive(Clone)] +pub(super) struct WorkerAuthorityView { + workspace_id: String, + store: Arc, + config_store: Arc, + feature_storage: crate::WorkspaceFeatureStorage, + subjektiv_registration: crate::RegisteredFeature, +} + +impl From<&WorkspaceApi> for WorkerAuthorityView { + fn from(api: &WorkspaceApi) -> Self { + Self { + workspace_id: api.workspace_id().into(), + store: api.store.clone(), + config_store: api.config_store.clone(), + feature_storage: api.feature_storage.clone(), + subjektiv_registration: api.subjektiv_registration.clone(), + } } - Ok(store - .subject_permissions(subject_id) - .map_err(subjektiv_store_error)?) +} + +impl WorkerAuthorityView { + fn subject_store(&self) -> ApiResult { + crate::subjektiv::SubjektivStore::open(&self.feature_storage, &self.subjektiv_registration) + .map_err(|error| Error::Store(error.to_string()).into()) + } + + fn require_not_deleted(&self, worker: &RuntimeWorkerRef) -> ApiResult<()> { + let deleting = self.config_store.with_conn(|connection| { + Ok(connection.query_row( + "SELECT EXISTS(SELECT 1 FROM subject_deletion_workers WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3)", + rusqlite::params![self.workspace_id, worker.runtime_id, worker.worker_id], + |row| row.get::<_, bool>(0), + )?) + })?; + if deleting + || self + .subject_store()? + .worker_subject_deleted(&worker.runtime_id, &worker.worker_id) + .map_err(subjektiv_store_error)? + { + return Err(denied("Worker belonged to an explicitly deleted Subject")); + } + Ok(()) + } + + fn policy(&self, subject_id: &str) -> ApiResult> { + let store = self.subject_store()?; + let subject = store + .subject(subject_id) + .map_err(subjektiv_store_error)? + .ok_or_else(|| denied("Subject permission source no longer exists"))?; + if subject.state != crate::subjektiv::SubjectState::Active { + return Err(denied("Subject permission source is retired")); + } + Ok(store + .subject_permissions(subject_id) + .map_err(subjektiv_store_error)?) + } + + fn current_body_subject(&self, worker: &RuntimeWorkerRef) -> ApiResult> { + let retained = self + .store + .retained_worker_singleton_lease(&self.workspace_id, worker)?; + let Some(retained) = retained else { + return Ok(None); + }; + let Some(subject) = retained.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else { + return Ok(None); + }; + // This also rejects a retained lease from a released/replaced body. Never + // fall through to non-Subject compatibility after ownership has ended. + let current = self + .store + .require_current_worker_singleton_owner(&self.workspace_id, worker)? + .ok_or_else(|| denied("Subject body no longer owns its singleton"))?; + if current.key != retained.key { + return Err(denied("Subject body ownership changed")); + } + Ok(Some(subject.to_owned())) + } + + fn live_child_grant( + &self, + worker: &RuntimeWorkerRef, + ) -> ApiResult> { + let store = self.subject_store()?; + let Some(grant) = store + .worker_execution_grant(&worker.runtime_id, &worker.worker_id) + .map_err(subjektiv_store_error)? + else { + return Ok(None); + }; + let source = self + .policy(&grant.subject_id)? + .ok_or_else(|| denied("Subject permissions are unset"))?; + if source.change_id != grant.source_change_id { + return Err(denied( + "delegated execution grant was revoked by permission replacement", + )); + } + let mut current = grant.clone(); + let mut visited = HashSet::new(); + for _ in 0..64 { + if !visited.insert((current.runtime_id.clone(), current.worker_id.clone())) { + return Err(denied("cyclic execution grant lineage")); + } + if current.subject_id != grant.subject_id + || current.source_change_id != source.change_id + || current + .permissions + .iter() + .chain(current.redelegation.iter()) + .any(|permission| { + !source + .delegation + .iter() + .any(|ceiling| ceiling.covers(permission)) + }) + { + return Err(denied( + "delegated execution exceeds the current Subject ceiling", + )); + } + let origin = + RuntimeWorkerRef::new(¤t.origin_runtime_id, ¤t.origin_worker_id); + match store + .worker_execution_grant(&origin.runtime_id, &origin.worker_id) + .map_err(subjektiv_store_error)? + { + Some(parent) => { + if current + .permissions + .iter() + .chain(current.redelegation.iter()) + .any(|permission| { + !parent + .redelegation + .iter() + .any(|ceiling| ceiling.covers(permission)) + }) + { + return Err(denied( + "execution grant exceeds its immediate parent's redelegation ceiling", + )); + } + current = parent; + } + None => { + if self.current_body_subject(&origin)?.as_deref() + != Some(grant.subject_id.as_str()) + { + return Err(denied( + "delegation origin is no longer the current Subject body", + )); + } + return Ok(Some(grant)); + } + } + } + Err(denied( + "execution grant lineage exceeds the bounded delegation depth", + )) + } + + fn subject_permissions( + &self, + worker: &RuntimeWorkerRef, + ) -> ApiResult>> { + if let Some(grant) = self.live_child_grant(worker)? { + return Ok(Some(grant.permissions)); + } + let Some(subject) = self.current_body_subject(worker)? else { + return Ok(None); + }; + Ok(Some( + self.policy(&subject)? + .map(|policy| policy.execution) + .unwrap_or_default(), + )) + } + + /// Saved individual-Worker permissions remain unchanged. Their canonical + /// projection authorizes operation use only; without an explicit current + /// WorkerOperation capability, the original exact target still restricts + /// Backend control. Neither covers nor delegation ceilings are widened. + fn authorize_control_use( + &self, + controller: &RuntimeWorkerRef, + target: &RuntimeWorkerRef, + permission: &str, + ) -> ApiResult<()> { + self.require_not_deleted(controller)?; + if self + .store + .worker_registry_projection(&self.workspace_id, controller)? + .is_some_and(|record| record.job.is_some()) + { + return Err(denied("Job domain grants do not authorize Worker control")); + } + let Some(permissions) = self.subject_permissions(controller)? else { + return Ok(()); // Preserve ordinary Workers' existing Backend contract. + }; + let operation = if permission == "notify" { + "worker.messaging".to_owned() + } else { + format!("worker.{permission}") + }; + let requested = ScopedPermission { + operation: operation.clone(), + resource: PermissionResource::WorkerOperation { + workspace_id: self.workspace_id.clone(), + }, + }; + if permissions.iter().any(|granted| granted.covers(&requested)) { + return Ok(()); + } + if permissions.iter().any(|granted| { + granted.validate().is_ok() + && canonical_worker_operation(&granted.operation) == Some(operation.as_str()) + && matches!(&granted.resource, PermissionResource::Worker { runtime_id, worker_id } + if runtime_id == &target.runtime_id && worker_id == &target.worker_id) + }) { + return Ok(()); + } + Err(denied(format!( + "Worker operation {operation} is not granted for this target" + ))) + } + + pub(super) fn authorize_control_permission( + &self, + controller: &RuntimeWorkerRef, + target: &RuntimeWorkerRef, + permission: &str, + ) -> Result { + self.authorize_control_use(controller, target, permission) + .map_err(|error| error.error)?; + let grant = self + .store + .get_active_worker_control_grant(&self.workspace_id, controller, target)? + .ok_or_else(|| Error::UnknownWorker { + worker: target.clone(), + })?; + if !grant + .permissions + .iter() + .any(|candidate| candidate == permission) + { + return Err(Error::InvalidInput(format!( + "worker control permission {permission} was not granted" + ))); + } + Ok(grant) + } +} + +fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult> { + WorkerAuthorityView::from(api).policy(subject_id) } fn current_body_subject( api: &WorkspaceApi, worker: &RuntimeWorkerRef, ) -> ApiResult> { - let retained = api - .store - .retained_worker_singleton_lease(&api.config.workspace_id, worker)?; - let Some(retained) = retained else { - return Ok(None); - }; - let Some(subject) = retained.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else { - return Ok(None); - }; - // This also rejects a retained lease from a released/replaced body. Never - // fall through to non-Subject compatibility after ownership has ended. - let current = api - .store - .require_current_worker_singleton_owner(&api.config.workspace_id, worker)? - .ok_or_else(|| denied("Subject body no longer owns its singleton"))?; - if current.key != retained.key { - return Err(denied("Subject body ownership changed")); - } - Ok(Some(subject.to_owned())) + WorkerAuthorityView::from(api).current_body_subject(worker) } fn live_child_grant( api: &WorkspaceApi, worker: &RuntimeWorkerRef, ) -> ApiResult> { - let store = open_subjektiv_store(api)?; - let Some(grant) = store - .worker_execution_grant(&worker.runtime_id, &worker.worker_id) - .map_err(subjektiv_store_error)? - else { - return Ok(None); - }; - let source = - policy(api, &grant.subject_id)?.ok_or_else(|| denied("Subject permissions are unset"))?; - if source.change_id != grant.source_change_id { - return Err(denied( - "delegated execution grant was revoked by permission replacement", - )); - } - let mut current = grant.clone(); - let mut visited = HashSet::new(); - for _ in 0..64 { - if !visited.insert((current.runtime_id.clone(), current.worker_id.clone())) { - return Err(denied("cyclic execution grant lineage")); - } - if current.subject_id != grant.subject_id - || current.source_change_id != source.change_id - || current - .permissions - .iter() - .chain(current.redelegation.iter()) - .any(|permission| { - !source - .delegation - .iter() - .any(|ceiling| ceiling.covers(permission)) - }) - { - return Err(denied( - "delegated execution exceeds the current Subject ceiling", - )); - } - let origin = RuntimeWorkerRef::new(¤t.origin_runtime_id, ¤t.origin_worker_id); - match store - .worker_execution_grant(&origin.runtime_id, &origin.worker_id) - .map_err(subjektiv_store_error)? - { - Some(parent) => { - if current - .permissions - .iter() - .chain(current.redelegation.iter()) - .any(|permission| { - !parent - .redelegation - .iter() - .any(|ceiling| ceiling.covers(permission)) - }) - { - return Err(denied( - "execution grant exceeds its immediate parent's redelegation ceiling", - )); - } - current = parent; - } - None => { - if current_body_subject(api, &origin)?.as_deref() != Some(grant.subject_id.as_str()) - { - return Err(denied( - "delegation origin is no longer the current Subject body", - )); - } - return Ok(Some(grant)); - } - } - } - Err(denied( - "execution grant lineage exceeds the bounded delegation depth", - )) + WorkerAuthorityView::from(api).live_child_grant(worker) } /// Job domain capabilities attenuate this policy; they are never non-Subject @@ -291,27 +450,12 @@ pub(super) fn worker_permissions( api: &WorkspaceApi, worker: &RuntimeWorkerRef, ) -> ApiResult>> { - if subjektiv_deletion::deleting_worker(api, worker)? - || open_subjektiv_store(api)? - .worker_subject_deleted(&worker.runtime_id, &worker.worker_id) - .map_err(subjektiv_store_error)? - { - return Err(denied("Worker belonged to an explicitly deleted Subject")); - } + let authority = WorkerAuthorityView::from(api); + authority.require_not_deleted(worker)?; if let Some(permissions) = job_permissions(api, worker)? { return Ok(Some(permissions)); } - if let Some(grant) = live_child_grant(api, worker)? { - return Ok(Some(grant.permissions)); - } - let Some(subject) = current_body_subject(api, worker)? else { - return Ok(None); - }; - Ok(Some( - policy(api, &subject)? - .map(|policy| policy.execution) - .unwrap_or_default(), - )) + authority.subject_permissions(worker) } pub(super) fn authorize( @@ -383,6 +527,20 @@ pub(super) fn authorize_scope( let Some(permissions) = worker_permissions(api, worker)? else { return Ok(None); }; + // Read-only migration of saved target-scoped permissions. Do not mutate the + // assignment/grant or use this projection for delegation; Backend separately + // checks the original exact target at every control effect. + if matches!(&needed.resource, PermissionResource::WorkerOperation { workspace_id } + if workspace_id == api.workspace_id()) + && permissions.iter().any(|permission| { + permission.validate().is_ok() + && canonical_worker_operation(&permission.operation) + == Some(needed.operation.as_str()) + && matches!(permission.resource, PermissionResource::Worker { .. }) + }) + { + return Ok(Some(needed.resource)); + } for permission in &permissions { if permission.operation == needed.operation && resources_cover(api, &permission.resource, &needed.resource)? @@ -518,9 +676,59 @@ pub(super) fn bind_reserved_grant( ) -> ApiResult<()> { let mut bound = grant.clone(); bound.worker_id = worker_id.to_owned(); - Ok(open_subjektiv_store(api)? + open_subjektiv_store(api)? .bind_worker_execution_grant(&bound) - .map_err(subjektiv_store_error)?) + .map_err(subjektiv_store_error)?; + Ok(()) +} + +/// Persist only the confirmed execution delegation's reply direction, never an +/// inferred inverse of an ordinary A -> B control grant. A reservation is not a +/// Worker registry row: this runs after normal creation has committed that row. +/// Initial Runtime execution may already have started; replies are unavailable +/// until this write, and initial input is not queued or redispatched to hide that +/// window. Spawn replay repairs a failed write under the same allocation identity +/// without replacing the Worker or reissuing a revoked saved grant. +pub(super) fn persist_reply_grant( + api: &WorkspaceApi, + controller: &RuntimeWorkerRef, + child: &RuntimeWorkerRef, + allocation_key: &str, +) -> ApiResult<()> { + let authority = WorkerAuthorityView::from(api); + authority.require_not_deleted(child)?; + let Some(bound) = authority.live_child_grant(child)? else { + return Ok(()); // Ordinary Workers do not acquire reverse control. + }; + let origin = RuntimeWorkerRef::new(&bound.origin_runtime_id, &bound.origin_worker_id); + if &origin != controller { + return Err(denied( + "reply grant origin does not match authenticated spawn source", + )); + } + if api + .store + .get_worker_registry(api.workspace_id(), child)? + .is_none() + { + return Err(denied( + "reply grant requires the committed Worker registry identity", + )); + } + api.store + .create_worker_control_grant(&WorkerControlGrantRecord { + workspace_id: api.workspace_id().into(), + grant_id: new_id("wcg"), + controller: child.clone(), + subject: origin, + relation: "reply".into(), + origin: "execution_grant".into(), + permissions: vec!["notify".into()], + operation_id: format!("worker-reply:{allocation_key}"), + created_at: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + revoked_at: None, + })?; + Ok(()) } /// Host-owned binding for an already authenticated Backend operation. No model diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index c1406f884..902e4ea5e 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs @@ -7,6 +7,10 @@ use super::*; #[path = "subject_job_model_supply_tests.rs"] mod subject_job_model_supply; +#[path = "worker_messaging_connected_tests.rs"] +mod worker_messaging_connected; +#[path = "worker_operation_authorization_tests.rs"] +mod worker_operation_authorization; use agen::tool::{Tool, ToolExecutionContext, ToolOutput}; use server_api::authorization::{ PermissionResource, ScopedPermission, SubjectPermissions, SubjectPermissionsUpdate, @@ -568,7 +572,7 @@ fn concrete_resource_permissions(f: &Fixture) -> Vec { }, ), ( - "worker.observe", + "subworker.observe", PermissionResource::SubWorker { runtime_id: f.source.runtime_id.clone(), worker_id: f.source.worker_id.clone(), @@ -2656,32 +2660,61 @@ async fn delegated_subject_child_cannot_downgrade_ticket_start_to_manual_by_omit ); } -/// Shared real HTTP Backend/Runtime fixture. Only the injected model is scripted; -/// Workdir materialization, Profile Controllers and tool dispatch are production. -struct ConnectedAuthorizationFixture { - _runtime_server: ConnectedRuntimeServer, - _backend_server: AuthorizationCallbackServer, +const CONNECTED_RUNTIME: &str = "connected-subject-authorization"; + +/// Shared signed Backend/Runtime connection. Only the supplied model is replaced; +/// temporary roots, request authentication and Controller/tool dispatch stay real. +struct ConnectedWorkerHarness { api: WorkspaceApi, + runtime: worker_runtime::Runtime, + factory: worker_runtime::worker_backend::ProfileRuntimeWorkerFactory, identity: RuntimeIdentityMaterial, backend_url: String, - materializer: worker_runtime::working_directory::RuntimeGitMaterializer, - runtime_errors: Arc>>, - authorizations: Arc>>, owner: String, + materializer: worker_runtime::working_directory::RuntimeGitMaterializer, + authorizations: Arc>>, + runtime_errors: Arc>>, + _backend_server: AuthorizationCallbackServer, + _runtime_server: ConnectedRuntimeServer, _dir: tempfile::TempDir, } -impl ConnectedAuthorizationFixture { - const RUNTIME: &'static str = "connected-subject-authorization"; +impl ConnectedWorkerHarness { + async fn human(&self, method: reqwest::Method, path: &str, body: Value) -> (StatusCode, Value) { + let response = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .unwrap() + .request(method.clone(), format!("{}{path}", self.backend_url)) + .bearer_auth(&self.owner) + .json(&body) + .send() + .await + .unwrap(); + let status = StatusCode::from_u16(response.status().as_u16()).unwrap(); + let bytes = response.bytes().await.unwrap(); + let body = if status == StatusCode::NO_CONTENT { + assert!(bytes.is_empty(), "{method} {path}: HTTP204 must be empty"); + Value::Null + } else { + serde_json::from_slice(&bytes).unwrap_or_else(|error| { + panic!( + "{method} {path}: HTTP{status}, invalid JSON: {error}; body={:?}", + String::from_utf8_lossy(&bytes) + ) + }) + }; + (status, body) + } - async fn new(model: M) -> Self { - let runtime_id = Self::RUNTIME; + async fn start(model: M) -> Self { + const RUNTIME: &str = CONNECTED_RUNTIME; let dir = tempfile::tempdir().unwrap(); init_clean_git_workspace(dir.path()); let backend_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let backend_url = format!("http://{}", backend_listener.local_addr().unwrap()); let runtime_url = format!("http://{}", runtime_listener.local_addr().unwrap()); - let identity = RuntimeIdentityMaterial::generate(runtime_id).unwrap(); + let identity = RuntimeIdentityMaterial::generate(RUNTIME).unwrap(); let runtime_root = dir.path().join("connected-runtime"); let materializer = worker_runtime::working_directory::RuntimeGitMaterializer::new( dir.path().join("connected-workdirs"), @@ -2694,19 +2727,20 @@ impl ConnectedAuthorizationFixture { worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new( TEST_WORKSPACE_ID, &backend_url, - runtime_id, + RUNTIME, ) .with_runtime_request_source(&identity, &backend_url), ) .with_runtime_store_dir(&runtime_root) .with_spawn_model_client_factory(move || Box::new(model.clone())); - let execution = worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory) - .unwrap() - .with_working_directory_materializer(materializer.clone()); + let execution = + worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(factory.clone()) + .unwrap() + .with_working_directory_materializer(materializer.clone()); let runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend( worker_runtime::fs_store::FsRuntimeStoreOptions { root: runtime_root, - runtime_id: runtime_id.into(), + runtime_id: RUNTIME.into(), display_name: None, }, Arc::new(execution), @@ -2733,8 +2767,8 @@ impl ConnectedAuthorizationFixture { .upsert_workspace_runtime_binding_record( WorkspaceRuntimeBinding { workspace_id: TEST_WORKSPACE_ID.into(), - runtime_id: runtime_id.into(), - display_name: runtime_id.into(), + runtime_id: RUNTIME.into(), + display_name: RUNTIME.into(), base_url: runtime_url.clone(), public_key: identity.public_key.clone(), public_key_fingerprint: String::new(), @@ -2750,7 +2784,7 @@ impl ConnectedAuthorizationFixture { .unwrap(); let binding = api .store - .get_workspace_runtime_binding(TEST_WORKSPACE_ID, runtime_id) + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, RUNTIME) .await .unwrap() .unwrap(); @@ -2792,7 +2826,7 @@ impl ConnectedAuthorizationFixture { None, worker_runtime::http_server::WorkspaceRuntimeHttpAuth { verifier, - runtime_id: runtime_id.into(), + runtime_id: RUNTIME.into(), }, ) .layer(axum::middleware::from_fn( @@ -2800,7 +2834,9 @@ impl ConnectedAuthorizationFixture { let errors = captured_errors.clone(); async move { let response = next.run(request).await; - if response.status().is_success() { + if !response.status().is_client_error() + && !response.status().is_server_error() + { return response; } let (parts, body) = response.into_parts(); @@ -2820,9 +2856,9 @@ impl ConnectedAuthorizationFixture { api.runtime.register_or_replace( RemoteWorkerRuntime::new( RemoteRuntimeConfig { - runtime_id: runtime_id.into(), + runtime_id: RUNTIME.into(), workspace_id: Some(TEST_WORKSPACE_ID.into()), - display_name: runtime_id.into(), + display_name: RUNTIME.into(), base_url: runtime_url, bearer_token: None, workspace_authorization: Some( @@ -2845,7 +2881,6 @@ impl ConnectedAuthorizationFixture { ) .unwrap(), ); - install_subject_profiles(&api, "workspace_write"); seed_test_delivery_binding(&api); let authorizations = Arc::new(Mutex::new(Vec::<(Value, StatusCode)>::new())); @@ -2875,34 +2910,22 @@ impl ConnectedAuthorizationFixture { let owner = seed_test_workspace_owner_token(api.store.as_ref(), "connected-authorization-owner") .await; + Self { api, + runtime, + factory, identity, backend_url, - materializer, - runtime_errors, - authorizations, owner, - _runtime_server: runtime_server, + materializer, + authorizations, + runtime_errors, _backend_server: backend_server, + _runtime_server: runtime_server, _dir: dir, } } - - async fn human(&self, method: reqwest::Method, path: &str, body: Value) -> (StatusCode, Value) { - let response = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .build() - .unwrap() - .request(method, format!("{}{path}", self.backend_url)) - .bearer_auth(&self.owner) - .json(&body) - .send() - .await - .unwrap(); - let status = StatusCode::from_u16(response.status().as_u16()).unwrap(); - (status, response.json().await.unwrap()) - } } #[tokio::test(flavor = "multi_thread", worker_threads = 6)] @@ -2910,16 +2933,17 @@ async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_r { use worker_runtime::working_directory::WorkingDirectoryMaterializer; - const RUNTIME: &str = ConnectedAuthorizationFixture::RUNTIME; + const RUNTIME: &str = CONNECTED_RUNTIME; let (completed, mut completion) = tokio::sync::mpsc::channel(4); - let f = ConnectedAuthorizationFixture::new(ConnectedAuthorizationModel { completed }).await; - let api = &f.api; - let backend_url = &f.backend_url; - let identity = &f.identity; - let materializer = &f.materializer; - let runtime_errors = &f.runtime_errors; - let authorizations = &f.authorizations; - let owner = &f.owner; + let harness = ConnectedWorkerHarness::start(ConnectedAuthorizationModel { completed }).await; + let api = &harness.api; + let identity = &harness.identity; + let backend_url = &harness.backend_url; + let owner = &harness.owner; + let materializer = &harness.materializer; + let authorizations = &harness.authorizations; + let runtime_errors = &harness.runtime_errors; + install_subject_profiles(api, "workspace_write"); let (status, created_workdir) = response(&api, human_request(&owner, "POST", &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), Some(json!({"runtime_id":RUNTIME, "repository_key":"test-repository", "selector":"HEAD", "operation_id":"connected-authorization-workdir"})))).await; @@ -3301,12 +3325,13 @@ async fn connected_two_subjects_delegate_distinct_tickets_concurrently_on_separa request: Value, } let (completed, mut completion) = tokio::sync::mpsc::channel(2); - let f = ConnectedAuthorizationFixture::new(ConcurrentTicketAuthorizationModel { + let f = ConnectedWorkerHarness::start(ConcurrentTicketAuthorizationModel { overlap: Arc::new(tokio::sync::Barrier::new(2)), completed, }) .await; let api = &f.api; + install_subject_profiles(api, "workspace_write"); let subjects = open_subjektiv_store(api).unwrap(); let mut cases = Vec::new(); for label in ["alpha", "beta"] { @@ -3320,10 +3345,14 @@ async fn connected_two_subjects_delegate_distinct_tickets_concurrently_on_separa .unwrap() .success() ); - let (status, created) = f.human(reqwest::Method::POST, - &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), - json!({"runtime_id":ConnectedAuthorizationFixture::RUNTIME, "repository_key":"test-repository", - "selector":branch, "operation_id":format!("parallel-{label}-workdir")})).await; + let (status, created) = f + .human( + reqwest::Method::POST, + &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), + json!({"runtime_id":CONNECTED_RUNTIME, "repository_key":"test-repository", + "selector":branch, "operation_id":format!("parallel-{label}-workdir")}), + ) + .await; assert_eq!(status, StatusCode::CREATED, "{created}"); let workdir = created["item"]["working_directory_id"] .as_str() @@ -3395,6 +3424,12 @@ async fn connected_two_subjects_delegate_distinct_tickets_concurrently_on_separa .map(|operation| subject_permission(&subject, operation)) .collect(); execution.push(ticket_permission("ticket.assignment.assign")); + execution.push(ScopedPermission { + operation: "worker.create".into(), + resource: PermissionResource::Workspace { + workspace_id: TEST_WORKSPACE_ID.into(), + }, + }); let (status, policy) = f .human( reqwest::Method::POST, @@ -3408,7 +3443,7 @@ async fn connected_two_subjects_delegate_distinct_tickets_concurrently_on_separa assert_eq!(policy.delegation, permissions); let (status, source) = f.human(reqwest::Method::POST, &format!("/api/w/{TEST_WORKSPACE_ID}/workers"), - json!({"runtime_id":ConnectedAuthorizationFixture::RUNTIME, "display_name":format!("{label} read-only body"), + json!({"runtime_id":CONNECTED_RUNTIME, "display_name":format!("{label} read-only body"), "profile":"project:authority-source", "initial_submit":[], "workdir_attachments":[], "feature_connections":{"subjektiv":{"subject_id":subject}}})).await; assert_eq!( @@ -3447,7 +3482,7 @@ async fn connected_two_subjects_delegate_distinct_tickets_concurrently_on_separa .unwrap() .to_owned(); let content = format!("connected {label} result"); - let request = json!({"runtime_id":ConnectedAuthorizationFixture::RUNTIME, + let request = json!({"runtime_id":CONNECTED_RUNTIME, "display_name":format!("{label} delegated Ticket editor"), "profile":"project:authority-editor", "ticket_assignment":{"ticket_id":ticket.id, "operation_id":format!("parallel-{label}-start"), "orchestrator_assignment_id":delegation}, diff --git a/crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs b/crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs new file mode 100644 index 000000000..dc701fbf2 --- /dev/null +++ b/crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs @@ -0,0 +1,637 @@ +//! Public Worker tools over the shared signed Backend/Runtime connection. The +//! external model alone is scripted; permissions API, Controller, tool engine, +//! control grants, notification admission and canonical Session storage are real. +//! No developer model credentials or live Runtime state are used. +use super::*; +use agen::llm_client::{ + event::{Event, StopReason}, + types::Item, +}; +use worker_runtime::worker_backend::RuntimeWorkerFactory; + +const ORIGIN: &str = "ROUNDTRIP_ORIGIN"; +const PING: &str = "ROUNDTRIP_QUESTION"; +const REPLY: &str = "ROUNDTRIP_PROGRESS_RESULT"; +const EARLY: &str = "ROUNDTRIP_INITIAL_CHILD"; + +fn operation(operation: &str) -> ScopedPermission { + ScopedPermission { + operation: operation.into(), + resource: PermissionResource::WorkerOperation { + workspace_id: TEST_WORKSPACE_ID.into(), + }, + } +} +fn collection(operation: &str) -> ScopedPermission { + ScopedPermission { + operation: operation.into(), + resource: PermissionResource::Workspace { + workspace_id: TEST_WORKSPACE_ID.into(), + }, + } +} +fn result(request: &agen::llm_client::Request, id: &str) -> Option { + request.items.iter().find_map(|item| match item { + Item::ToolResult { + call_id, + is_error, + content, + summary, + .. + } if call_id == id => { + assert!(!is_error, "public tool {id}: {summary}; {content:?}"); + Some(serde_json::from_str(content.as_deref().expect("public JSON result")).unwrap()) + } + _ => None, + }) +} +fn call(id: &str, name: &str, input: Value) -> Vec { + vec![ + Event::tool_use_start(0, id, name), + Event::tool_input_delta(0, input.to_string()), + Event::tool_use_stop(0), + ] +} +fn done() -> Vec { + vec![ + Event::text_block_start(0), + Event::text_delta(0, "roundtrip observed"), + Event::text_block_stop(0, Some(StopReason::EndTurn)), + ] +} + +struct RepairSignals { + early_observation: tokio::sync::mpsc::Sender, + spawn_refusal: tokio::sync::mpsc::Sender<()>, + allow_replay: tokio::sync::Notify, + origin: Mutex>, + origin_ready: tokio::sync::Notify, +} +#[derive(Clone)] +struct MessagingModel { + child_permissions: Arc>>, + workdir: Arc>>, + completed: tokio::sync::mpsc::Sender, + child_replied: Arc, + repair: Option>, +} +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for MessagingModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + async fn stream( + &self, + request: agen::llm_client::Request, + ) -> std::result::Result + { + let messages = request + .items + .iter() + .filter_map(|item| match item { + Item::Message { content, .. } => Some(serde_json::to_string(content).unwrap()), + _ => None, + }) + .collect::>() + .join("\n"); + let origin = messages.contains(ORIGIN); + let events = if origin { + if let Some(stopped) = result(&request, "roundtrip-stop") { + assert_eq!(stopped["state"], "accepted", "{stopped}"); + let child = result(&request, "roundtrip-spawn").unwrap(); + if let Some(is_error) = request.items.iter().find_map(|item| match item { + Item::ToolResult { + call_id, is_error, .. + } if call_id == "stopped-notify" => Some(*is_error), + _ => None, + }) { + assert!( + is_error, + "Notify to stopped child must refuse, never restore or claim success" + ); + let _ = self + .completed + .try_send(json!({"spawn":child,"messages":messages})); + done() + } else { + call( + "stopped-notify", + "WorkerNotify", + json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":"MUST_NOT_REVIVE_STOPPED_CHILD"}), + ) + } + } else if messages.contains(REPLY) { + let child = result(&request, "roundtrip-spawn").unwrap(); + call( + "roundtrip-stop", + "WorkerStop", + json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"reason":"connected roundtrip completed"}), + ) + } else if result(&request, "roundtrip-ping").is_some() { + // Hold this model boundary while the real child replies to a Busy + // origin. This is a deterministic external-model barrier, not a + // replacement for notification admission or Controller scheduling. + tokio::time::timeout( + std::time::Duration::from_secs(30), + self.child_replied.notified(), + ) + .await + .expect("child public reply result"); + done() + } else if let Some(child) = result(&request, "roundtrip-spawn") { + if result(&request, "roundtrip-observe").is_some() { + call( + "roundtrip-ping", + "WorkerNotify", + json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":PING}), + ) + } else { + // The origin has observe, not worker.list. A known target must + // remain readable without collection discovery authority. + call( + "roundtrip-observe", + "ViewSessionOverview", + json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]}}), + ) + } + } else { + let call_id = if let Some(repair) = &self.repair { + if let Some((is_error, summary)) = + request.items.iter().find_map(|item| match item { + Item::ToolResult { + call_id, + is_error, + summary, + .. + } if call_id == "roundtrip-spawn-first" => Some((*is_error, summary)), + _ => None, + }) + { + assert!( + is_error, + "grant-save fault must not report successful Spawn: {summary}" + ); + let _ = repair.spawn_refusal.try_send(()); + tokio::time::timeout( + std::time::Duration::from_secs(30), + repair.allow_replay.notified(), + ) + .await + .expect("explicit original-operation replay authorization"); + "roundtrip-spawn" + } else { + "roundtrip-spawn-first" + } + } else { + "roundtrip-spawn" + }; + let mut input = json!({"runtime_id":CONNECTED_RUNTIME,"profile":"project:messaging-peer","operation_id":"public-roundtrip-spawn","display_name":"Notify-only delegated peer","initial_submit":if self.repair.is_some() {vec![protocol::Segment::Text {content:EARLY.into()}]} else {vec![]},"execution_grant":{"permissions":self.child_permissions.lock().unwrap().clone(),"redelegation":[]}}); + if let Some(workdir) = self.workdir.lock().unwrap().as_ref() { + input["working_directory_id"] = json!(workdir); + } + call(call_id, "WorkerSpawn", input) + } + } else if self.repair.is_some() && !messages.contains(PING) { + assert!( + messages.contains(EARLY), + "early child started through original initial Submit" + ); + if let Some(available) = request.items.iter().find_map(|item| match item { + Item::ToolResult { + call_id, + is_error, + content, + .. + } if call_id == "early-origin-list" => { + let available = !is_error + && content.as_deref().is_some_and(|text| { + serde_json::from_str::(text).unwrap()["items"] + .as_array() + .is_some_and(|items| !items.is_empty()) + }); + Some(available) + } + _ => None, + }) { + assert!( + !available, + "an unpersisted reply grant cannot be discovered as authority" + ); + let repair = self.repair.as_ref().unwrap(); + if let Some(is_error) = request.items.iter().find_map(|item| match item { + Item::ToolResult { + call_id, is_error, .. + } if call_id == "early-denied-notify" => Some(*is_error), + _ => None, + }) { + assert!( + is_error, + "missing reply grant must deny public Notify, not report acceptance" + ); + let _ = repair.early_observation.try_send(available); + done() + } else { + tokio::time::timeout( + std::time::Duration::from_secs(30), + repair.origin_ready.notified(), + ) + .await + .expect("test obtains original origin identity"); + // The harness supplies only a target identity, never authority. + // Discovery was empty; even knowing the ID cannot authorize Notify. + let origin = repair.origin.lock().unwrap().clone().unwrap(); + call( + "early-denied-notify", + "WorkerNotify", + json!({"subject":{"kind":"runtime_worker","runtime_id":origin.runtime_id,"worker_id":origin.worker_id},"content":"MUST_NOT_ACCEPT_EARLY_REPLY"}), + ) + } + } else { + call("early-origin-list", "WorkerList", json!({})) + } + } else { + assert!( + messages.contains(PING), + "child starts from real Idle Notify, not hidden initial Submit: {messages}" + ); + if let Some(reply) = result(&request, "roundtrip-reply") { + assert_eq!(reply["state"], "accepted", "{reply}"); + self.child_replied.notify_one(); + done() + } else if let Some(list) = result(&request, "roundtrip-origin-list") { + let origins = list["items"] + .as_array() + .unwrap() + .iter() + .filter(|item| item["relation"] == "reply") + .collect::>(); + assert_eq!( + origins.len(), + 1, + "public discovery must expose only the confirmed reply origin: {list}" + ); + assert_eq!(origins[0]["permissions"], json!(["notify"])); + call( + "roundtrip-reply", + "WorkerNotify", + json!({"subject":origins[0]["subject"],"content":REPLY}), + ) + } else { + call("roundtrip-origin-list", "WorkerList", json!({})) + } + }; + Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok)))) + } +} + +fn install_messaging_profiles(api: &WorkspaceApi) { + let state = api + .config_store + .load_workspace_config(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let main = state + .snapshot + .get(&config_source::VirtualPath::parse("main.dcdl").unwrap()) + .unwrap(); + let content = r#"{ profile = { default_profile = "project:messaging-origin"; entries = [ + { selector = "project:messaging-origin"; label = "Messaging origin"; description = "Public delegation boundary"; + profile = (import "$builtin/profiles/companion.dcdl") // { slug = "messaging-origin"; scope = "workspace_read"; delegation_scope = "workspace_read"; feature = { worker = { enabled = true; direct_spawn = true; }; }; }; + }, + { selector = "project:messaging-peer"; label = "Messaging peer"; description = "Notify-only delegated peer"; + profile = (import "$builtin/profiles/companion.dcdl") // { slug = "messaging-peer"; scope = "workspace_read"; feature = { subjektiv = { enabled = false; }; worker = { enabled = true; direct_spawn = false; }; }; }; + } + ]; }; }"#; + commit_workspace_config_tree( + api, + TEST_WORKSPACE_ID, + &ConfigCommitRequest { + base_digest: state.snapshot.digest.clone(), + entrypoints: state.contract.entrypoints.clone(), + changes: vec![config_source::ConfigTreeChange::Update { + path: main.path.clone(), + expected_digest: main.content_digest.clone(), + content: content.into(), + }], + }, + ) + .unwrap(); +} + +async fn connected_roundtrip(explicit_workdir: bool, repair: bool) { + use worker_runtime::working_directory::WorkingDirectoryMaterializer; + let (completed, mut completion) = tokio::sync::mpsc::channel(2); + let permissions = Arc::new(Mutex::new(vec![ + operation("worker.messaging"), + collection("worker.list"), + ])); + let workdir = Arc::new(Mutex::new(None)); + let (early_observation, mut early) = tokio::sync::mpsc::channel(1); + let (spawn_refusal, mut refused) = tokio::sync::mpsc::channel(1); + let repair_signals = repair.then(|| { + Arc::new(RepairSignals { + early_observation, + spawn_refusal, + allow_replay: tokio::sync::Notify::new(), + origin: Mutex::new(None), + origin_ready: tokio::sync::Notify::new(), + }) + }); + let harness = ConnectedWorkerHarness::start(MessagingModel { + child_permissions: permissions.clone(), + workdir: workdir.clone(), + completed, + child_replied: Arc::new(tokio::sync::Notify::new()), + repair: repair_signals.clone(), + }) + .await; + let api = &harness.api; + install_messaging_profiles(api); + if explicit_workdir { + let (status, created) = response(api, human_request(&harness.owner, "POST", &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), Some(json!({"runtime_id":CONNECTED_RUNTIME,"repository_key":"test-repository","selector":"HEAD","operation_id":"public-messaging-workdir"})))).await; + assert_eq!(status, StatusCode::CREATED, "{created}"); + let id = created["item"]["working_directory_id"] + .as_str() + .unwrap() + .to_owned(); + permissions.lock().unwrap().push(ScopedPermission { + operation: "filesystem.read".into(), + resource: PermissionResource::Workdir { + working_directory_id: id.clone(), + path: ".".into(), + recursive: true, + }, + }); + *workdir.lock().unwrap() = Some(id); + } + let store = open_subjektiv_store(api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("public-messaging").unwrap()) + .unwrap() + .id; + let mut execution = [ + "session.record", + "memory.resident.read", + "memory.resident.supply", + "journal.startup.read", + "journal.startup.supply", + ] + .into_iter() + .map(|op| subject_permission(&subject, op)) + .collect::>(); + execution.extend([ + collection("worker.create"), + operation("worker.messaging"), + operation("worker.stop"), + operation("worker.observe"), + ]); + let (status, assigned) = response(api, human_request(&harness.owner,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),Some(json!({"operation_id":"public-messaging-policy","expected_change_id":null,"execution":execution,"delegation":permissions.lock().unwrap().clone()})))).await; + assert_eq!(status, StatusCode::OK, "{assigned}"); + let before = store.subject_permissions(&subject).unwrap().unwrap(); + if repair { + api.config_store.with_conn(|connection| { + connection.execute_batch("CREATE TRIGGER t803_reply_save_fault BEFORE INSERT ON worker_control_grants WHEN NEW.relation='reply' BEGIN SELECT RAISE(ABORT,'injected reply grant save failure'); END;")?; + Ok(()) + }).unwrap(); + } + let Json(source) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: CONNECTED_RUNTIME.into(), + display_name: "Public WorkerSpawn Subject origin".into(), + singleton_key: None, + profile: Some("project:messaging-origin".into()), + ticket_assignment: None, + initial_submit: vec![protocol::Segment::Text { + content: ORIGIN.into(), + }], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: subject.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap_or_else(|error| { + panic!( + "origin: {error:?}; Runtime: {:?}", + harness.runtime_errors.lock().unwrap() + ) + }); + if let Some(signals) = &repair_signals { + *signals.origin.lock().unwrap() = + Some(RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id)); + signals.origin_ready.notify_one(); + assert!( + !tokio::time::timeout(std::time::Duration::from_secs(60), early.recv()) + .await + .expect("earliest child public discovery") + .unwrap() + ); + tokio::time::timeout(std::time::Duration::from_secs(60), refused.recv()) + .await + .expect("public Spawn save-failure refusal") + .unwrap(); + assert_eq!( + harness.runtime.list_workers().unwrap().len(), + 2, + "confirmed original child is retained despite reply-save failure" + ); + api.config_store + .with_conn(|connection| { + connection.execute_batch("DROP TRIGGER t803_reply_save_fault")?; + Ok(()) + }) + .unwrap(); + signals.allow_replay.notify_one(); + } + let observed = tokio::time::timeout(std::time::Duration::from_secs(60), completion.recv()) + .await + .expect("public Spawn/Notify/reply/Stop complete") + .expect("model completion evidence"); + let child = RuntimeWorkerRef::new( + observed["spawn"]["runtime_id"].as_str().unwrap(), + observed["spawn"]["worker_id"].as_str().unwrap(), + ); + let origin = RuntimeWorkerRef::new(source.runtime_id, source.worker_id); + assert!(observed["messages"].as_str().unwrap().contains(REPLY)); + assert_eq!( + store.subject_permissions(&subject).unwrap().unwrap(), + before, + "allocated ID was never appended to Subject policy" + ); + let forward = api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &origin, &child) + .unwrap() + .unwrap(); + assert!(forward.permissions.contains(&"stop".into())); + let reverse = api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &origin) + .unwrap() + .unwrap(); + assert_eq!(reverse.permissions, vec!["notify".to_owned()]); + let child_ref = worker_runtime::identity::WorkerRef::try_from(&child).unwrap(); + let snapshot = harness + .factory + .retained_session_snapshot(&child_ref) + .unwrap(); + let history = serde_json::to_string(&snapshot.snapshot).unwrap(); + assert!( + !history.contains("MUST_NOT_REVIVE_STOPPED_CHILD"), + "stopped destination received no notification/mailbox input" + ); + assert_eq!( + harness.runtime.worker_detail(&child_ref).unwrap().status, + worker_runtime::catalog::WorkerStatus::Stopped + ); + if repair { + assert_eq!( + snapshot + .snapshot + .entries + .iter() + .filter(|item| serde_json::to_string(item).unwrap().contains(EARLY)) + .count(), + 1, + "original initial input was not redispatched on recovery" + ); + assert_eq!( + harness.runtime.list_workers().unwrap().len(), + 2, + "one origin and one child after original-operation replay" + ); + } + assert!( + history.contains(PING), + "Idle notification persisted before model supply" + ); + let source_ref = worker_runtime::identity::WorkerRef::try_from(&origin).unwrap(); + let origin_history = serde_json::to_string( + &harness + .factory + .retained_session_snapshot(&source_ref) + .unwrap() + .snapshot, + ) + .unwrap(); + assert!( + !origin_history.contains("MUST_NOT_ACCEPT_EARLY_REPLY"), + "unauthorized early Notify caused no origin input" + ); + assert!( + origin_history.contains(REPLY), + "Busy reply persisted before origin model supply" + ); + assert!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &origin) + .unwrap() + .is_empty() + ); + let links = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &child) + .unwrap(); + assert_eq!(links.len(), usize::from(explicit_workdir)); + if let Some(id) = workdir.lock().unwrap().as_ref() { + assert_eq!( + fs::read_to_string( + harness + .materializer + .bind_working_directory(id, None) + .unwrap() + .root() + .join("README.md") + ) + .unwrap(), + "clean\n" + ); + } + api.store + .revoke_worker_control_grant( + TEST_WORKSPACE_ID, + &forward.grant_id, + &now_registry_timestamp(), + ) + .unwrap(); + let (denied, body) = connected_authorization_post( + &harness.backend_url, + &harness.identity, + &origin, + &format!( + "/api/w/{TEST_WORKSPACE_ID}/worker-control/workers/{}/{}/input", + child.runtime_id, child.worker_id + ), + &json!({"kind":"notify","content":"MUST_NOT_ACCEPT_REVOKED_CONTROL"}), + ) + .await; + assert_eq!( + denied, + StatusCode::NOT_FOUND, + "revoked forward control: {body}" + ); + assert_eq!( + harness.runtime.worker_detail(&child_ref).unwrap().status, + worker_runtime::catalog::WorkerStatus::Stopped + ); + assert!( + !serde_json::to_string( + &harness + .factory + .retained_session_snapshot(&child_ref) + .unwrap() + .snapshot + ) + .unwrap() + .contains("MUST_NOT_ACCEPT_REVOKED_CONTROL") + ); + let calls = harness.authorizations.lock().unwrap(); + for op in [ + "worker.create", + "worker.messaging", + "worker.stop", + "worker.observe", + ] { + assert!( + calls + .iter() + .any(|(request, status)| request["operation"] == op && *status == StatusCode::OK), + "real signed common authorization for {op}: {calls:?}" + ); + } + assert!( + calls + .iter() + .filter(|(request, _)| request["operation"] == "worker.messaging" + || request["operation"] == "worker.stop") + .all(|(request, _)| request["resource"]["kind"] == "worker_operation") + ); + assert!( + browser_ticket_backend(api) + .unwrap() + .list(Default::default()) + .unwrap() + .is_empty() + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn public_spawn_notify_reply_and_stop_without_ticket_or_attachment() { + connected_roundtrip(false, false).await; +} +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn public_spawn_notify_reply_and_stop_with_explicit_workdir() { + connected_roundtrip(true, false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn public_initial_reply_unavailable_until_original_spawn_replay_repairs_grant() { + connected_roundtrip(false, true).await; +} diff --git a/crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs b/crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs new file mode 100644 index 000000000..900cea084 --- /dev/null +++ b/crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs @@ -0,0 +1,945 @@ +//! Signed-router component tests for operation use and independent directional control. +//! Subject policy APIs, SQLite grants, Feature installation and public Tools are real. +//! WorkdirlessFixtureRuntime substitutes execution; Accepted is not Controller processing. +//! Run with `cargo test -p yoi-workspace-server --lib worker_operation_authorization`. +use super::*; +use worker::feature::builtin::manage_worker::manage_worker_feature; + +struct OperationFixture { + inner: Fixture, + stops: Arc, +} + +impl std::ops::Deref for OperationFixture { + type Target = Fixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl OperationFixture { + async fn new() -> Self { + let mut inner = Fixture::new().await; + let stops = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counter = stops.clone(); + inner.runtime.stop_callback = Some(Arc::new(move || { + counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + })); + inner.api.runtime.register_or_replace(inner.runtime.clone()); + Self { inner, stops } + } +} + +const CONTROL_OPERATIONS: &[(&str, &str)] = &[ + ("worker.messaging", "notify"), + ("worker.send_input", "send_input"), + ("worker.cancel", "cancel"), + ("worker.stop", "stop"), + ("worker.restore", "restore"), + ("worker.remove", "remove"), + ("worker.observe", "observe"), +]; + +fn workspace_permission(operation: &str) -> ScopedPermission { + ScopedPermission { + operation: operation.into(), + resource: PermissionResource::Workspace { + workspace_id: TEST_WORKSPACE_ID.into(), + }, + } +} + +fn operation_permission(operation: &str) -> ScopedPermission { + ScopedPermission { + operation: operation.into(), + resource: PermissionResource::WorkerOperation { + workspace_id: TEST_WORKSPACE_ID.into(), + }, + } +} + +fn control_permissions() -> Vec { + CONTROL_OPERATIONS + .iter() + .map(|(operation, _)| operation_permission(operation)) + .collect() +} + +fn target_permission(operation: &str, target: &RuntimeWorkerRef) -> ScopedPermission { + ScopedPermission { + operation: operation.into(), + resource: PermissionResource::Worker { + runtime_id: target.runtime_id.clone(), + worker_id: target.worker_id.clone(), + }, + } +} + +fn subject_value(worker: &RuntimeWorkerRef) -> Value { + json!({"kind":"runtime_worker", "runtime_id":worker.runtime_id, "worker_id":worker.worker_id}) +} + +fn peer_grant( + f: &OperationFixture, + controller: &RuntimeWorkerRef, + target: &RuntimeWorkerRef, + permissions: &[&str], +) -> String { + let id = Uuid::now_v7().to_string(); + f.api + .store + .create_worker_control_grant(&WorkerControlGrantRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + grant_id: id.clone(), + controller: controller.clone(), + subject: target.clone(), + relation: "peer".into(), + origin: "explicit_test_delegation".into(), + permissions: permissions + .iter() + .map(|permission| (*permission).into()) + .collect(), + operation_id: id.clone(), + created_at: now_registry_timestamp(), + revoked_at: None, + }) + .unwrap(); + id +} + +async fn backend_operation( + f: &OperationFixture, + controller: &RuntimeWorkerRef, + target: &RuntimeWorkerRef, + permission: &str, +) -> (StatusCode, Value) { + let base = format!( + "/api/w/{TEST_WORKSPACE_ID}/worker-control/workers/{}/{}", + target.runtime_id, target.worker_id + ); + let (path, body) = match permission { + "notify" | "send_input" => ( + format!("{base}/input"), + json!({"kind":if permission == "notify" {"notify"} else {"user"}, "content":"authority matrix input"}), + ), + "restore" => { + let observation = f + .api + .runtime + .worker(target) + .unwrap() + .restore_observation_token + .unwrap(); + ( + format!("{base}/restore"), + json!({"expected_observation_token":observation, "request_id":Uuid::now_v7().to_string()}), + ) + } + "observe" => ( + format!("/api/w/{TEST_WORKSPACE_ID}/worker-observation/session"), + subject_value(target), + ), + "remove" => ( + format!("/api/w/{TEST_WORKSPACE_ID}/workers/remove"), + json!({"target_runtime_id":target.runtime_id,"target_worker_id":target.worker_id,"reason":"authorization matrix"}), + ), + "cancel" | "stop" => ( + format!("{base}/{permission}"), + json!({"reason":"authorization matrix"}), + ), + _ => panic!("unknown matrix operation {permission}"), + }; + let mut request = runtime_source_request( + &f.identity, + Some(&controller.worker_id), + "POST", + &path, + serde_json::to_vec(&body).unwrap(), + ); + if permission == "remove" { + let proof = + worker_runtime::auth::RuntimeWorkerMutationSourceSigner::from_identity(&f.identity) + .issue_worker_remove( + "server-test", + TEST_WORKSPACE_ID, + &controller.worker_id, + &target.runtime_id, + &target.worker_id, + 30, + ) + .unwrap(); + request.headers_mut().insert( + worker_runtime::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER, + proof.parse().unwrap(), + ); + } + tokio::time::timeout( + std::time::Duration::from_secs(20), + response(&f.api, request), + ) + .await + .unwrap_or_else(|_| panic!("Backend {permission} did not complete")) +} + +#[derive(Debug, PartialEq, Eq)] +struct RuntimeEffects { + inputs: usize, + stops: usize, + spawns: usize, + restores: usize, + workers: usize, +} + +fn effects(f: &OperationFixture) -> RuntimeEffects { + RuntimeEffects { + inputs: f.runtime.input_requests.lock().unwrap().len(), + stops: f.stops.load(std::sync::atomic::Ordering::SeqCst), + spawns: f.runtime.spawn_requests().len(), + restores: *f.runtime.restore_attempts.lock().unwrap(), + workers: f.runtime.workers.lock().unwrap().len(), + } +} + +async fn assert_backend_denied( + f: &OperationFixture, + controller: &RuntimeWorkerRef, + target: &RuntimeWorkerRef, + expected: Option, +) { + let before = effects(f); + let record = f + .api + .store + .get_worker_registry(TEST_WORKSPACE_ID, target) + .unwrap(); + for (_, permission) in CONTROL_OPERATIONS { + let (status, body) = backend_operation(f, controller, target, permission).await; + if let Some(expected) = expected { + assert_eq!(status, expected, "{permission}: {body}"); + } else { + assert!(status.is_client_error(), "{permission}: {status} {body}"); + } + assert_eq!( + effects(f), + before, + "denied {permission} reached Runtime effects" + ); + assert_eq!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, target) + .unwrap(), + record, + "denied {permission} mutated target registry" + ); + } +} + +async fn worker_tool_host( + root: &Path, + client: Arc, +) -> worker::Worker { + worker_tool_host_with_observation(root, client, None).await +} + +async fn worker_tool_host_with_observation( + root: &Path, + client: Arc, + observation: Option>, +) -> worker::Worker { + let manifest = worker::WorkerManifest::from_toml( + r#" + [worker] + name = "worker-operation-tools" + [model] + scheme = "anthropic" + model_id = "unused" + [engine] + [scope] + allow = [] + [feature.workdir_catalog] + enabled = false + [feature.worker] + enabled = true + "#, + ) + .unwrap(); + let engine = + agen::Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated( + UnusedModel, + ); + let mut host = worker::Worker::new( + manifest, + engine, + session_store::FsStore::new(root).unwrap(), + worker::WorkerWorkspaceContext::with_client( + Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()), + client.clone(), + ), + worker::WorkerFilesystemAuthority::None, + manifest::Scope::empty(), + ) + .await + .unwrap(); + let mut features = worker::feature::FeatureRegistryBuilder::new() + .with_module(manage_worker_feature(client, None, true)); + if let Some(provider) = observation { + features = features.with_module( + worker::feature::builtin::worker_observation::WorkerObservationFeature::new(provider), + ); + } + let report = host.install_features(features); + assert!( + !report.has_errors(), + "Worker Feature must install without granting operation use" + ); + host.engine().tool_server_handle().flush_pending(); + host +} + +fn installed_tool( + host: &worker::Worker, + name: &str, +) -> Arc { + host.engine() + .tool_server_handle() + .get_tool(name) + .unwrap_or_else(|| panic!("missing installed public {name}")) + .1 +} + +fn execute_tool_result( + host: &worker::Worker, + name: &str, + input: Value, +) -> std::result::Result { + let tool = installed_tool(host, name); + // The signed fixture transport is synchronous. Leave Tokio's async driver + // before executing the public Tool, just as the real blocking adapter does. + tokio::task::block_in_place(|| { + futures::executor::block_on( + tool.execute(&input.to_string(), ToolExecutionContext::direct()), + ) + }) +} + +async fn execute_tool( + host: &worker::Worker, + name: &str, + input: Value, +) -> Value { + value(execute_tool_result(host, name, input).unwrap_or_else(|error| panic!("{name}: {error}"))) +} + +fn tool_arguments(name: &str, target: &RuntimeWorkerRef) -> Value { + let mut args = json!({"subject":subject_value(target)}); + match name { + "WorkerSendInput" | "WorkerNotify" => args["content"] = json!("public tool input"), + "WorkerRemove" => args["reason"] = json!("must not remove"), + "WorkerRestore" => { + args["expected_observation_token"] = json!("not-used-before-authorization") + } + _ => {} + } + args +} + +// Only the protected Session read is substituted here. Installed observation / +// control Features and signed Host / Backend target checks remain production code. +struct CountedObservation(std::sync::atomic::AtomicUsize); +#[async_trait::async_trait] +impl worker::feature::contracts::WorkerObservationProvider for CountedObservation { + async fn list_worker_sessions( + &self, + ) -> std::result::Result< + Vec, + worker::feature::contracts::WorkerObservationError, + > { + panic!("known-target capture must not discover the collection") + } + async fn capture_worker_session( + &self, + _: &worker::feature::contracts::WorkerObservationSubjectRef, + ) -> std::result::Result< + worker::feature::contracts::WorkerSessionCapture, + worker::feature::contracts::WorkerObservationError, + > { + self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(worker::feature::contracts::WorkerSessionCapture { + segment_id: "target-session".into(), + session: protocol::SessionSnapshot { + pending_submissions: Default::default(), + entries: vec![protocol::SessionSnapshotEntry { + use_policy: None, + entry_id: "observed-entry".into(), + timestamp: 1, + provenance: protocol::SessionEntryProvenance::LegacyUnknown, + derived_from: vec![], + data: protocol::SessionSnapshotEntryData::Message { + role: protocol::SessionMessageRole::User, + content: vec![protocol::SessionContentPart::Text { + text: "observed input".into(), + }], + }, + }], + }, + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn installed_observation_needs_only_observe_and_target_grant_not_collection_discovery() { + let f = OperationFixture::new().await; + f.assign(vec![workspace_permission("worker.create")], vec![]) + .await; + let target = f.child("observed-target", vec![]).await; + f.assign(vec![operation_permission("worker.observe")], vec![]) + .await; + let provider = Arc::new(CountedObservation(std::sync::atomic::AtomicUsize::new(0))); + let client = f.client(&f.source); + let host = worker_tool_host_with_observation( + &f._dir.path().join("observation-tools"), + client.clone(), + Some(provider.clone()), + ) + .await; + assert!( + execute_tool_result(&host, "WorkerList", json!({})).is_err(), + "observe never grants collection discovery" + ); + let overview = execute_tool( + &host, + "ViewSessionOverview", + json!({"subject":subject_value(&target)}), + ) + .await; + let entry_ref = overview["entries"][0]["entry_ref"].as_str().unwrap(); + let tools = [ + ( + "ViewSessionOverview", + json!({"subject":subject_value(&target)}), + ), + ( + "SearchSessionEntries", + json!({"subject":subject_value(&target), "query":"observed"}), + ), + ( + "ReadSessionEntry", + json!({"subject":subject_value(&target), "entry_ref":entry_ref}), + ), + ]; + client.requests.lock().unwrap().clear(); + for (name, input) in &tools { + execute_tool(&host, name, input.clone()).await; + } + assert_eq!(provider.0.load(std::sync::atomic::Ordering::SeqCst), 4); + assert!( + client + .requests + .lock() + .unwrap() + .iter() + .all(|(_, path)| !path.ends_with("/worker-control/workers")), + "known-target public observation must not invoke WorkerList" + ); + let grant = f + .api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &f.source, &target) + .unwrap() + .unwrap(); + for missing in ["operation", "target", "revoked"] { + match missing { + "operation" => { + f.assign(vec![], vec![]).await; + } + "target" => { + f.assign(vec![operation_permission("worker.observe")], vec![]) + .await; + f.api + .store + .revoke_worker_control_grant( + TEST_WORKSPACE_ID, + &grant.grant_id, + &now_registry_timestamp(), + ) + .unwrap(); + } + "revoked" => {} + _ => unreachable!(), + } + for (name, input) in &tools { + let mut input = input.clone(); + if missing == "target" { + input["subject"]["worker_id"] = json!("unknown-target"); + } + let error = execute_tool_result(&host, name, input) + .unwrap_err() + .to_string(); + assert!( + error.contains("not found or is not accessible"), + "{missing} {name}: {error}" + ); + } + assert_eq!( + provider.0.load(std::sync::atomic::Ordering::SeqCst), + 4, + "denied {missing} must not read or supply protected Session data" + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn backend_control_grant_does_not_replace_exact_operation_use_permission() { + let f = OperationFixture::new().await; + f.assign(vec![workspace_permission("worker.create")], vec![]) + .await; + let target = f.child("permission-matrix-target", vec![]).await; + assert!( + f.api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &f.source, &target) + .unwrap() + .is_some() + ); + assert_backend_denied(&f, &f.source, &target, Some(StatusCode::FORBIDDEN)).await; + // Old collection/Feature permissions and another Workspace never authorize + // individual current Worker operations, even with a real target grant. + let mut wrong_workspace = control_permissions(); + for permission in &mut wrong_workspace { + permission.resource = PermissionResource::WorkerOperation { + workspace_id: "another-workspace".into(), + }; + } + for mut permissions in [ + CONTROL_OPERATIONS + .iter() + .map(|(operation, _)| workspace_permission(operation)) + .collect(), + vec![activation("worker")], + wrong_workspace, + ] { + permissions.push(workspace_permission("worker.create")); + f.assign(permissions, vec![]).await; + assert_backend_denied(&f, &f.source, &target, Some(StatusCode::FORBIDDEN)).await; + } + f.assign(vec![operation_permission("worker.messaging")], vec![]) + .await; + assert_eq!( + backend_operation(&f, &f.source, &target, "notify").await.0, + StatusCode::OK + ); + for (_, permission) in CONTROL_OPERATIONS + .iter() + .filter(|(_, permission)| *permission != "notify") + { + let before = effects(&f); + let (status, body) = backend_operation(&f, &f.source, &target, permission).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "messaging is not {permission}: {body}" + ); + assert_eq!(effects(&f), before); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn peer_operation_use_requires_explicit_directional_target_grant_and_revocation_is_live() { + let f = OperationFixture::new().await; + let permissions = control_permissions(); + f.assign( + vec![workspace_permission("worker.create")], + permissions.clone(), + ) + .await; + let sender = f.child("peer-sender", permissions.clone()).await; + let recipient = f.child("peer-recipient", permissions).await; + assert_eq!( + f.check(&sender, &operation_permission("worker.messaging")) + .await + .0, + StatusCode::OK + ); + assert_backend_denied(&f, &sender, &recipient, None).await; + let grant = peer_grant(&f, &sender, &recipient, &["notify"]); + let before = effects(&f); + let (status, body) = backend_operation(&f, &sender, &recipient, "notify").await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["state"], "accepted"); + assert_eq!(effects(&f).inputs, before.inputs + 1); + assert_backend_denied(&f, &recipient, &sender, None).await; + for (_, permission) in CONTROL_OPERATIONS + .iter() + .filter(|(_, permission)| *permission != "notify") + { + let before = effects(&f); + let (status, body) = backend_operation(&f, &sender, &recipient, permission).await; + assert!( + status.is_client_error(), + "notify-only peer {permission}: {body}" + ); + assert_eq!(effects(&f), before); + } + f.api + .store + .revoke_worker_control_grant(TEST_WORKSPACE_ID, &grant, &now_registry_timestamp()) + .unwrap(); + assert_eq!( + f.check(&sender, &operation_permission("worker.messaging")) + .await + .0, + StatusCode::OK, + "revoking target control does not revoke operation use" + ); + assert_backend_denied(&f, &sender, &recipient, None).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn saved_targeted_message_permissions_preserve_target_direction_and_reject_alias_requests() { + for legacy in ["worker.notify", "worker.result"] { + let f = OperationFixture::new().await; + f.assign(vec![workspace_permission("worker.create")], vec![]) + .await; + let target = f.child("saved-target", vec![]).await; + let other = f.child("saved-other-target", vec![]).await; + let saved = target_permission(legacy, &target); + let policy = f.assign(vec![saved.clone()], vec![]).await; + assert_eq!( + policy.execution, + vec![saved.clone()], + "saved target must not be rewritten into an operation-wide capability" + ); + assert_eq!( + f.check(&f.source, &operation_permission("worker.messaging")) + .await + .0, + StatusCode::OK + ); + let (status, body) = backend_operation(&f, &f.source, &target, "notify").await; + assert_eq!(status, StatusCode::OK, "{legacy}: {body}"); + let before = effects(&f); + let (status, body) = backend_operation(&f, &f.source, &other, "notify").await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "{legacy} must retain exact target despite another real control grant: {body}" + ); + assert_eq!(effects(&f), before); + for request in [ + saved, + ScopedPermission { + operation: legacy.into(), + resource: operation_permission("worker.messaging").resource, + }, + target_permission("worker.messaging", &target), + ] { + let (status, body) = f.check(&f.source, &request).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "storage-only alias/target must not become a public authorization request: {body}" + ); + } + assert_backend_denied(&f, &target, &other, None).await; + assert_eq!( + open_subjektiv_store(&f.api) + .unwrap() + .subject_permissions(&f.subject) + .unwrap() + .unwrap(), + policy + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn installed_worker_tools_register_without_authority_and_enforce_operation_matrix_before_effects() + { + let f = OperationFixture::new().await; + let client = f.client(&f.source); + let host = worker_tool_host(&f._dir.path().join("unset-worker-tools"), client.clone()).await; + assert!( + client.requests.lock().unwrap().is_empty(), + "registration is local" + ); + let spawn = json!({"runtime_id":f.source.runtime_id,"operation_id":"public-spawn-denied","profile":"builtin:coder","execution_grant":{"permissions":[],"redelegation":[]}}); + for (name, args) in [("WorkerList", json!({})), ("WorkerSpawn", spawn)] { + let before = effects(&f); + assert!( + execute_tool_result(&host, name, args).is_err(), + "unset {name}" + ); + assert_eq!(effects(&f), before); + } + f.assign(vec![workspace_permission("worker.create")], vec![]) + .await; + let target = f.child("public-denied-target", vec![]).await; + for name in [ + "WorkerNotify", + "WorkerSendInput", + "WorkerCancel", + "WorkerStop", + "WorkerRestore", + "WorkerRemove", + ] { + let before = effects(&f); + client.requests.lock().unwrap().clear(); + assert!( + execute_tool_result(&host, name, tool_arguments(name, &target)).is_err(), + "missing operation {name}" + ); + assert_eq!(effects(&f), before); + assert!( + client + .requests + .lock() + .unwrap() + .iter() + .all(|(_, path)| path == &authorization_path()), + "{name} must refuse before the control backend" + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn installed_public_spawn_notify_stop_and_child_reply_use_real_backend_grants() { + let f = OperationFixture::new().await; + let reply = operation_permission("worker.messaging"); + let mut permissions = control_permissions(); + permissions.extend([ + workspace_permission("worker.create"), + workspace_permission("worker.list"), + ]); + f.assign(permissions, vec![reply.clone()]).await; + let client = f.client(&f.source); + let host = worker_tool_host(&f._dir.path().join("parent-tools"), client).await; + let created = execute_tool(&host, "WorkerSpawn", json!({ + "runtime_id":f.source.runtime_id,"operation_id":"public-worker-child","profile":"builtin:coder", + "execution_grant":{"permissions":[reply],"redelegation":[]} + })).await; + let child = RuntimeWorkerRef::new( + created["runtime_id"].as_str().unwrap(), + created["worker_id"].as_str().unwrap(), + ); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &child) + .unwrap() + .is_some() + ); + let listed = execute_tool(&host, "WorkerList", json!({})).await; + assert!( + listed["items"] + .as_array() + .unwrap() + .iter() + .any(|item| item["subject"] == subject_value(&child)) + ); + let notified = execute_tool( + &host, + "WorkerNotify", + tool_arguments("WorkerNotify", &child), + ) + .await; + assert_eq!(notified["state"], "accepted"); + let child_host = worker_tool_host(&f._dir.path().join("reply-tools"), f.client(&child)).await; + let before = effects(&f); + let reply = execute_tool( + &child_host, + "WorkerNotify", + json!({"subject":subject_value(&f.source),"content":"completed child result"}), + ) + .await; + assert_eq!(reply["state"], "accepted"); + assert_eq!(effects(&f).inputs, before.inputs + 1); + for name in [ + "WorkerSendInput", + "WorkerCancel", + "WorkerStop", + "WorkerRestore", + "WorkerRemove", + ] { + let before = effects(&f); + assert!( + execute_tool_result(&child_host, name, tool_arguments(name, &f.source)).is_err(), + "reply cannot authorize {name}" + ); + assert_eq!(effects(&f), before); + } + for (_, permission) in CONTROL_OPERATIONS + .iter() + .filter(|(_, permission)| *permission != "notify") + { + let before = effects(&f); + let (status, body) = backend_operation(&f, &child, &f.source, permission).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "direct reply escalation {permission}: {body}" + ); + assert_eq!(effects(&f), before); + } + let reverse = f + .api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &f.source) + .unwrap() + .unwrap(); + f.api + .store + .revoke_worker_control_grant( + TEST_WORKSPACE_ID, + &reverse.grant_id, + &now_registry_timestamp(), + ) + .unwrap(); + let replayed = execute_tool(&host, "WorkerSpawn", json!({ + "runtime_id":f.source.runtime_id,"operation_id":"public-worker-child","profile":"builtin:coder", + "execution_grant":{"permissions":[operation_permission("worker.messaging")],"redelegation":[]} + })).await; + assert_eq!(replayed["worker_id"], created["worker_id"]); + assert!( + f.api + .store + .get_active_worker_control_grant(TEST_WORKSPACE_ID, &child, &f.source) + .unwrap() + .is_none(), + "exact Spawn replay must not revive revoked reply control" + ); + let before = effects(&f); + assert!( + execute_tool_result( + &child_host, + "WorkerNotify", + json!({"subject":subject_value(&f.source),"content":"revoked reply"}) + ) + .is_err() + ); + assert_eq!( + effects(&f), + before, + "revoked public Notify had no Runtime effect" + ); + let stopped = execute_tool(&host, "WorkerStop", tool_arguments("WorkerStop", &child)).await; + assert_eq!(stopped["state"], "accepted"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn replaced_policy_and_body_never_revive_delegated_worker_control() { + let f = OperationFixture::new().await; + let permissions = control_permissions(); + f.assign( + vec![workspace_permission("worker.create")], + permissions.clone(), + ) + .await; + let child = f.child("old-policy-controller", permissions.clone()).await; + let target = f.child("policy-control-target", vec![]).await; + peer_grant( + &f, + &child, + &target, + &[ + "notify", + "send_input", + "cancel", + "stop", + "restore", + "remove", + "observe", + ], + ); + let grant = open_subjektiv_store(&f.api) + .unwrap() + .worker_execution_grant(&child.runtime_id, &child.worker_id) + .unwrap() + .unwrap(); + assert_eq!( + backend_operation(&f, &child, &target, "notify").await.0, + StatusCode::OK + ); + f.assign( + vec![workspace_permission("worker.create")], + permissions.clone(), + ) + .await; + assert_backend_denied(&f, &child, &target, Some(StatusCode::FORBIDDEN)).await; + // Identical replacement permissions do not refresh the captured policy ID. + f.assign( + vec![workspace_permission("worker.create")], + permissions.clone(), + ) + .await; + assert_backend_denied(&f, &child, &target, Some(StatusCode::FORBIDDEN)).await; + assert_eq!( + open_subjektiv_store(&f.api) + .unwrap() + .worker_execution_grant(&child.runtime_id, &child.worker_id) + .unwrap(), + Some(grant) + ); + let current = f.child("old-body-controller", permissions).await; + peer_grant( + &f, + ¤t, + &target, + &[ + "notify", + "send_input", + "cancel", + "stop", + "restore", + "remove", + "observe", + ], + ); + assert_eq!( + backend_operation(&f, ¤t, &target, "notify").await.0, + StatusCode::OK + ); + let store = open_subjektiv_store(&f.api).unwrap(); + let attribution = store + .subject_session_attributions(&f.subject) + .unwrap() + .into_iter() + .find(|attr| attr.runtime_id == f.source.runtime_id && attr.worker_id == f.source.worker_id) + .unwrap(); + let writer = + subjektiv::journal::JournalHostContext::validated_session(&store, attribution).unwrap(); + store + .release_journal_writer(&subjektiv::journal::JournalWriterBinding { + subject_id: f.subject.clone(), + runtime_id: f.source.runtime_id.clone(), + worker_id: f.source.worker_id.clone(), + binding_id: writer.writer_binding_id().unwrap().into(), + }) + .unwrap(); + f.release_source_lease(); + let Json(replacement) = create_workspace_worker( + State(f.api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: f.source.runtime_id.clone(), + display_name: "Replacement control origin".into(), + singleton_key: None, + profile: Some("project:authority-source".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: f.subject.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap(); + assert_ne!(replacement.worker_id, f.source.worker_id); + assert_backend_denied(&f, ¤t, &target, None).await; + assert_backend_denied(&f, &f.source, &target, Some(StatusCode::UNAUTHORIZED)).await; +} diff --git a/openapi/server-api.json b/openapi/server-api.json index 6405468ae..a6b02936d 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -7336,6 +7336,24 @@ ], "type": "object" }, + { + "additionalProperties": false, + "description": "Authorizes operation use, not a target Worker or a Workspace collection.\nThe Worker Feature must still enforce its directional target control grant.", + "properties": { + "kind": { + "const": "worker_operation", + "type": "string" + }, + "workspace_id": { + "type": "string" + } + }, + "required": [ + "kind", + "workspace_id" + ], + "type": "object" + }, { "additionalProperties": false, "properties": { @@ -10515,6 +10533,7 @@ }, "ScopedPermission": { "additionalProperties": false, + "description": "Stored scopes retain their original resources and exact operation names.\n\nA legacy Worker read adapter may use [`canonical_worker_operation`] to project\noperation use, but must retain the original target separately for the Worker\nFeature's control-grant check, including direction, revocation and delegation\nlimits. Replacing an old Worker resource with WorkerOperation and discarding\nthat target would broaden authority. Neither validation nor `covers` performs\nthat conversion.", "properties": { "operation": { "type": "string" @@ -19095,7 +19114,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38" + "x-yoi-source-digest": "sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472" }, "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 b41bb4fe7..edd0077fb 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:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 c63459336..4ec456781 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:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 d547a29fc..728827b8c 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:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 4b4b103fc..d59ea7b1e 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:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 b81accd57..e146390c0 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:1f5ae1d92ac04d33a05b9d4a34252e017e668d5c33520490bdfff7a6cde1e3cb) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 39dc8f42b..6e7171a00 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:f2ecdab70235cce3230dfc84579f5d7e7b69884308bb1286337d071aa337fc38) +// Input contract: 0.1.0 (sha256:7a78c365c7f30639cff040698f94ba6430be9e5bdeba866a3037bafc2fe1a472) // 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 @@ -35,7 +35,7 @@ export type PasteArtifactMediaType = "text_plain_utf8"; export type PasteArtifactRef = { artifact_id: string; availability: PasteArtifactAvailability; byte_len: number; char_count: number; created_at_ms: number; line_count: number; media_type: PasteArtifactMediaType; sha256: string; source_entry_id: string; }; -export type PermissionResource = { job_id: string; kind: "job"; workspace_id: string; } | { kind: "ticket"; ticket_id: string; workspace_id: string; } | { kind: "objective"; objective_id: string; workspace_id: string; } | { kind: "merge_request"; merge_request_id: string; workspace_id: string; } | { kind: "subject"; subject_id: string; workspace_id: string; } | { kind: "drive_entry"; node_id: string; workspace_id: string; } | { kind: "drive_request"; request_id: string; workspace_id: string; } | { kind: "repository"; repository_key: string; workspace_id: string; } | { connection_id: string; kind: "workdir_attachment"; workspace_id: string; } | { kind: "workspace_config"; path: string; recursive: boolean; workspace_id: string; } | { kind: "network"; url: string; } | { kind: "sub_worker"; name: string; runtime_id: string; worker_id: string; } | { kind: "feature"; } | { kind: "workspace"; workspace_id: string; } | { kind: "workdir"; path: string; recursive: boolean; working_directory_id: string; } | { kind: "worker"; runtime_id: string; worker_id: string; }; +export type PermissionResource = { job_id: string; kind: "job"; workspace_id: string; } | { kind: "ticket"; ticket_id: string; workspace_id: string; } | { kind: "objective"; objective_id: string; workspace_id: string; } | { kind: "merge_request"; merge_request_id: string; workspace_id: string; } | { kind: "subject"; subject_id: string; workspace_id: string; } | { kind: "drive_entry"; node_id: string; workspace_id: string; } | { kind: "drive_request"; request_id: string; workspace_id: string; } | { kind: "repository"; repository_key: string; workspace_id: string; } | { connection_id: string; kind: "workdir_attachment"; workspace_id: string; } | { kind: "workspace_config"; path: string; recursive: boolean; workspace_id: string; } | { kind: "network"; url: string; } | { kind: "sub_worker"; name: string; runtime_id: string; worker_id: string; } | { kind: "feature"; } | { kind: "workspace"; workspace_id: string; } | { kind: "workdir"; path: string; recursive: boolean; working_directory_id: string; } | { kind: "worker_operation"; workspace_id: string; } | { kind: "worker"; runtime_id: string; worker_id: string; }; export type RepositoryApiError = { diagnostics?: Array; error: string; message: string; };