From 74ddd568eb28c6b8cce8424967d0fd977ca0be7d Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 29 Oct 2026 22:26:35 +0900 Subject: [PATCH] Separate Worker operation authorization from directional control grants --- crates/server-api/src/authorization.rs | 286 ++++++- .../src/feature/builtin/manage_worker.rs | 308 +++++-- .../src/feature/builtin/worker_observation.rs | 10 +- crates/workspace-server/src/server.rs | 157 ++-- .../src/server/subject_authorization.rs | 458 +++++++--- .../tests/subject_authorization_tests.rs | 451 +++++----- .../tests/worker_messaging_connected_tests.rs | 621 ++++++++++++++ .../worker_operation_authorization_tests.rs | 788 ++++++++++++++++++ openapi/server-api.json | 21 +- .../src/lib/generated/companion-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 2 +- web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/workdir-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 4 +- 15 files changed, 2650 insertions(+), 464 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs create mode 100644 crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs 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/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 078e91741..a080f80b5 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,11 +403,7 @@ 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, - )?; + self.authorize_runtime_operation(&format!("worker.{permission}"))?; let response = self.client.execute_server_operation( crate::worker::WorkspaceServerOperation::WorkerControlList, )?; @@ -1105,6 +1089,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 +1297,8 @@ mod tests { struct RecordingWorkspaceClient { requests: Mutex>, authorizations: Mutex>, + bound_authorizations: + Mutex>, denied: std::sync::atomic::AtomicBool, removals: Mutex>, } @@ -1320,7 +1332,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 +1353,7 @@ mod tests { "runtime_id": "runtime-a", "worker_id": "worker-a" }, + "permissions": ["observe"], "summary": { "status": "idle" } }, { @@ -1339,11 +1361,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 +1500,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 +1739,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 +1775,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 +1824,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 +1892,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 +2198,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 +2408,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 +2451,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 +2488,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 +2531,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/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index bb01ce571..80d900b84 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, @@ -26423,6 +26414,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, @@ -26545,6 +26546,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 { @@ -26589,6 +26599,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, @@ -26596,54 +26626,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( @@ -26762,6 +26751,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)? @@ -26773,6 +26772,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 06253056d..ee49c8cfa 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(), @@ -2602,208 +2606,261 @@ impl Drop for ConnectedRuntimeServer { } } +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, + owner: String, + materializer: worker_runtime::working_directory::RuntimeGitMaterializer, + authorizations: Arc>>, + runtime_errors: Arc>>, + _backend_server: AuthorizationCallbackServer, + _runtime_server: ConnectedRuntimeServer, + _dir: tempfile::TempDir, +} +impl ConnectedWorkerHarness { + 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).unwrap(); + let runtime_root = dir.path().join("connected-runtime"); + let materializer = worker_runtime::working_directory::RuntimeGitMaterializer::new( + dir.path().join("connected-workdirs"), + ); + let factory = worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new( + dir.path().join("connected-profiles"), + ) + .with_remote_worker_mutation_identity(identity.clone()) + .with_workspace_request_client( + worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new( + TEST_WORKSPACE_ID, + &backend_url, + 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.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.into(), + display_name: None, + }, + Arc::new(execution), + ) + .unwrap(); + + let mut config = test_server_config(dir.path()); + config.backend_base_url = Some(backend_url.clone()); + // The embedded Runtime is unused. Both source and child below execute on the + // connected real Runtime above, not on this existing Host fixture backend. + let control = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(control.as_ref(), &config) + .await + .unwrap(); + let api = WorkspaceApi::new_with_execution_backend( + config, + control, + Arc::new(DeterministicExecutionBackend::default()), + ) + .await + .unwrap(); + let now = now_registry_timestamp(); + api.store + .upsert_workspace_runtime_binding_record( + WorkspaceRuntimeBinding { + workspace_id: TEST_WORKSPACE_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(), + binding_id: "connected-authorization-binding".into(), + authentication_mode: StoredRuntimeAuthenticationMode::WorkspaceIdentity, + created_at: now.clone(), + updated_at: now, + revoked_at: None, + }, + false, + ) + .await + .unwrap(); + let binding = api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, RUNTIME) + .await + .unwrap() + .unwrap(); + // Repository materialization and Worker startup must authenticate as this + // Workspace, not the Runtime HTTP local-token fallback's `local` Workspace. + let workspace_identity = api + .signing_identities + .provision_existing(TEST_WORKSPACE_ID, &test_owner_actor().account_id) + .unwrap(); + let verifier = worker_runtime::workspace_issuer::WorkspaceCapabilityVerifier::new( + vec![ + worker_runtime::workspace_issuer::WorkspaceIssuerTrustRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + backend_url: backend_url.clone(), + key_id: workspace_identity.key_id.clone(), + algorithm: workspace_identity.algorithm.clone(), + public_key: workspace_identity.public_key.clone().unwrap(), + public_key_fingerprint: workspace_identity + .public_key_fingerprint + .clone() + .unwrap(), + trust_id: "connected-authorization-workspace-trust".into(), + state: worker_runtime::workspace_issuer::WorkspaceIssuerTrustState::Active, + registered_at_unix: 1, + updated_at_unix: 1, + }, + ], + Arc::new( + worker_runtime::workspace_issuer::InMemoryWorkspaceClaimReplayProtection::default(), + ), + ) + .unwrap(); + let runtime_errors = Arc::new(Mutex::new(Vec::::new())); + let captured_errors = runtime_errors.clone(); + let runtime_server = ConnectedRuntimeServer::start( + runtime_listener, + worker_runtime::http_server::runtime_http_router_with_workspace_auth( + runtime.clone(), + None, + worker_runtime::http_server::WorkspaceRuntimeHttpAuth { + verifier, + runtime_id: RUNTIME.into(), + }, + ) + .layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let errors = captured_errors.clone(); + async move { + let response = next.run(request).await; + if response.status().is_success() { + return response; + } + let (parts, body) = response.into_parts(); + let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + errors.lock().unwrap().push( + body["error"]["message"] + .as_str() + .unwrap_or("unknown fixture error") + .to_owned(), + ); + Response::from_parts(parts, Body::from(bytes)) + } + }, + )), + ); + api.runtime.register_or_replace( + RemoteWorkerRuntime::new( + RemoteRuntimeConfig { + runtime_id: RUNTIME.into(), + workspace_id: Some(TEST_WORKSPACE_ID.into()), + display_name: RUNTIME.into(), + base_url: runtime_url, + bearer_token: None, + workspace_authorization: Some( + crate::hosts::WorkspaceRuntimeAuthorization::new( + api.store.clone(), + api.signing_identities.clone(), + backend_url.clone(), + binding, + ), + ), + strict_public_egress: false, + cached_worker_creation_available: true, + cached_os: "linux".into(), + cached_arch: "x86_64".into(), + cached_status: "active".into(), + timeout: std::time::Duration::from_secs(30), + }, + TEST_WORKSPACE_ID.to_string(), + backend_url.clone(), + ) + .unwrap(), + ); + seed_test_delivery_binding(&api); + + let authorizations = Arc::new(Mutex::new(Vec::<(Value, StatusCode)>::new())); + let observed = authorizations.clone(); + let app = build_router(api.clone()).layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let observed = observed.clone(); + async move { + if request.uri().path() != authorization_path() { + return next.run(request).await; + } + let (parts, body) = request.into_parts(); + let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); + let permission: Value = serde_json::from_slice(&bytes).unwrap(); + let response = next + .run(Request::from_parts(parts, Body::from(bytes))) + .await; + observed + .lock() + .unwrap() + .push((permission, response.status())); + response + } + }, + )); + let backend_server = AuthorizationCallbackServer::start(backend_listener, app); + let owner = + seed_test_workspace_owner_token(api.store.as_ref(), "connected-authorization-owner") + .await; + + Self { + api, + runtime, + factory, + identity, + backend_url, + owner, + materializer, + authorizations, + runtime_errors, + _backend_server: backend_server, + _runtime_server: runtime_server, + _dir: dir, + } + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 6)] async fn connected_read_only_subject_delegates_editing_profile_to_real_runtime_read_but_not_write() { use worker_runtime::working_directory::WorkingDirectoryMaterializer; - const RUNTIME: &str = "connected-subject-authorization"; - 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).unwrap(); - let runtime_root = dir.path().join("connected-runtime"); - let materializer = worker_runtime::working_directory::RuntimeGitMaterializer::new( - dir.path().join("connected-workdirs"), - ); + const RUNTIME: &str = CONNECTED_RUNTIME; let (completed, mut completion) = tokio::sync::mpsc::channel(4); - let model = ConnectedAuthorizationModel { completed }; - let factory = worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new( - dir.path().join("connected-profiles"), - ) - .with_remote_worker_mutation_identity(identity.clone()) - .with_workspace_request_client( - worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new( - TEST_WORKSPACE_ID, - &backend_url, - 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 runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend( - worker_runtime::fs_store::FsRuntimeStoreOptions { - root: runtime_root, - runtime_id: RUNTIME.into(), - display_name: None, - }, - Arc::new(execution), - ) - .unwrap(); - - let mut config = test_server_config(dir.path()); - config.backend_base_url = Some(backend_url.clone()); - // The embedded Runtime is unused. Both source and child below execute on the - // connected real Runtime above, not on this existing Host fixture backend. - let control = Arc::new(test_control_store(&config)); - seed_test_registered_workspace(control.as_ref(), &config) - .await - .unwrap(); - let api = WorkspaceApi::new_with_execution_backend( - config, - control, - Arc::new(DeterministicExecutionBackend::default()), - ) - .await - .unwrap(); - let now = now_registry_timestamp(); - api.store - .upsert_workspace_runtime_binding_record( - WorkspaceRuntimeBinding { - workspace_id: TEST_WORKSPACE_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(), - binding_id: "connected-authorization-binding".into(), - authentication_mode: StoredRuntimeAuthenticationMode::WorkspaceIdentity, - created_at: now.clone(), - updated_at: now, - revoked_at: None, - }, - false, - ) - .await - .unwrap(); - let binding = api - .store - .get_workspace_runtime_binding(TEST_WORKSPACE_ID, RUNTIME) - .await - .unwrap() - .unwrap(); - // Repository materialization and Worker startup must authenticate as this - // Workspace, not the Runtime HTTP local-token fallback's `local` Workspace. - let workspace_identity = api - .signing_identities - .provision_existing(TEST_WORKSPACE_ID, &test_owner_actor().account_id) - .unwrap(); - let verifier = worker_runtime::workspace_issuer::WorkspaceCapabilityVerifier::new( - vec![ - worker_runtime::workspace_issuer::WorkspaceIssuerTrustRecord { - workspace_id: TEST_WORKSPACE_ID.into(), - backend_url: backend_url.clone(), - key_id: workspace_identity.key_id.clone(), - algorithm: workspace_identity.algorithm.clone(), - public_key: workspace_identity.public_key.clone().unwrap(), - public_key_fingerprint: workspace_identity.public_key_fingerprint.clone().unwrap(), - trust_id: "connected-authorization-workspace-trust".into(), - state: worker_runtime::workspace_issuer::WorkspaceIssuerTrustState::Active, - registered_at_unix: 1, - updated_at_unix: 1, - }, - ], - Arc::new( - worker_runtime::workspace_issuer::InMemoryWorkspaceClaimReplayProtection::default(), - ), - ) - .unwrap(); - let runtime_errors = Arc::new(Mutex::new(Vec::::new())); - let captured_errors = runtime_errors.clone(); - let _runtime_server = ConnectedRuntimeServer::start( - runtime_listener, - worker_runtime::http_server::runtime_http_router_with_workspace_auth( - runtime.clone(), - None, - worker_runtime::http_server::WorkspaceRuntimeHttpAuth { - verifier, - runtime_id: RUNTIME.into(), - }, - ) - .layer(axum::middleware::from_fn( - move |request: axum::extract::Request, next: axum::middleware::Next| { - let errors = captured_errors.clone(); - async move { - let response = next.run(request).await; - if response.status().is_success() { - return response; - } - let (parts, body) = response.into_parts(); - let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); - let body: Value = serde_json::from_slice(&bytes).unwrap(); - errors.lock().unwrap().push( - body["error"]["message"] - .as_str() - .unwrap_or("unknown fixture error") - .to_owned(), - ); - Response::from_parts(parts, Body::from(bytes)) - } - }, - )), - ); - api.runtime.register_or_replace( - RemoteWorkerRuntime::new( - RemoteRuntimeConfig { - runtime_id: RUNTIME.into(), - workspace_id: Some(TEST_WORKSPACE_ID.into()), - display_name: RUNTIME.into(), - base_url: runtime_url, - bearer_token: None, - workspace_authorization: Some(crate::hosts::WorkspaceRuntimeAuthorization::new( - api.store.clone(), - api.signing_identities.clone(), - backend_url.clone(), - binding, - )), - strict_public_egress: false, - cached_worker_creation_available: true, - cached_os: "linux".into(), - cached_arch: "x86_64".into(), - cached_status: "active".into(), - timeout: std::time::Duration::from_secs(30), - }, - TEST_WORKSPACE_ID.to_string(), - backend_url.clone(), - ) - .unwrap(), - ); - install_subject_profiles(&api, "workspace_write"); - seed_test_delivery_binding(&api); - - let authorizations = Arc::new(Mutex::new(Vec::<(Value, StatusCode)>::new())); - let observed = authorizations.clone(); - let app = build_router(api.clone()).layer(axum::middleware::from_fn( - move |request: axum::extract::Request, next: axum::middleware::Next| { - let observed = observed.clone(); - async move { - if request.uri().path() != authorization_path() { - return next.run(request).await; - } - let (parts, body) = request.into_parts(); - let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); - let permission: Value = serde_json::from_slice(&bytes).unwrap(); - let response = next - .run(Request::from_parts(parts, Body::from(bytes))) - .await; - observed - .lock() - .unwrap() - .push((permission, response.status())); - response - } - }, - )); - let _backend_server = AuthorizationCallbackServer::start(backend_listener, app); - let owner = - seed_test_workspace_owner_token(api.store.as_ref(), "connected-authorization-owner").await; + 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; 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..f072cc727 --- /dev/null +++ b/crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs @@ -0,0 +1,621 @@ +//! 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") { + call( + "roundtrip-ping", + "WorkerNotify", + json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":PING}), + ) + } 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"), + ]); + 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"] { + 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..202e79bc8 --- /dev/null +++ b/crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs @@ -0,0 +1,788 @@ +//! 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 { + 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 report = host.install_features( + worker::feature::FeatureRegistryBuilder::new() + .with_module(manage_worker_feature(client, None, true)), + ); + 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 +} + +#[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 ad01a6caf..389f794ac 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" @@ -19062,7 +19081,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f" + "x-yoi-source-digest": "sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00" }, "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 e99726a94..d04296d6d 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:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f) +// Input contract: 0.1.0 (sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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 f71281eb9..dd9d8c896 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:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f) +// Input contract: 0.1.0 (sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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 efa2e84e4..0293f1c88 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:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f) +// Input contract: 0.1.0 (sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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 b6953e4d0..28b9374fb 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:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f) +// Input contract: 0.1.0 (sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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..c49d53845 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:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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 ba17a053c..c3f0fd68a 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:eca6e2cd9309ecf9fe73dcb044d82a5549950d66e93cd33157b2b2b3fa8ea82f) +// Input contract: 0.1.0 (sha256:24c38dc5714a966e6adde9ea8483ccc9e02cb6a0c2f01f776548155d1a488b00) // 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; };