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

Authorize known Worker observations without collection discovery

Browse Source
This commit is contained in:
Hare
2026-10-30 00:37:56 +09:00
parent 74ddd568eb
commit 51f0c7e019
19 changed files with 406 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/runtime-api/src/lib.rs
+3
View File
@@ -447,6 +447,9 @@ pub struct CreateWorkerRequest {
pub workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
/// Enables the optional Runtime-local peer observation provider, not operation
/// authority. Scoped Backend observation follows the Worker Feature and live
/// operation/control checks, including when this saved flag is false.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub worker_observation_enabled: bool,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
crates/server-api/src/lib.rs
+16
View File
@@ -1781,6 +1781,21 @@ pub trait ServerApi {
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
) -> Result<WorkerControlListResponse, RepositoryApiError>;
#[get(
"/api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/permissions/{permission}",
status = 204,
error_status = 400,
additional_error_statuses = [401, 403, 404, 500],
openapi = false,
)]
async fn worker_control_authorize(
&self,
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
#[path] runtime_id: String,
#[path] worker_id: String,
#[path] permission: String,
) -> Result<(), RepositoryApiError>;
#[post(
"/api/w/{workspace_id}/worker-control/workers",
status = 200,
@@ -14911,6 +14926,7 @@ mod openapi_artifact_tests {
"merge_request_review_submit",
"merge_request_reviewer_child_register",
"runtime_resource_fetch",
"worker_control_authorize",
"worker_control_cancel",
"worker_control_input",
"worker_control_list",
crates/worker-runtime/src/catalog.rs
+3 -2
View File
@@ -284,8 +284,9 @@ pub struct CreateWorkerRequest {
pub workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
/// Backend-only feature enablement. Grants still define local Runtime peers;
/// the Workspace provider reauthorizes its dynamic set per operation.
/// Backend-only enablement of the optional Runtime-local peer provider.
/// Scoped Backend observation is composed by the Worker Feature, independently
/// of this saved flag, and rechecks operation use plus current target control.
#[serde(default, skip_serializing_if = "is_false")]
pub worker_observation_enabled: bool,
/// Backend-authored, bounded peer session grants. Runtime revalidates each
crates/worker-runtime/src/worker_backend.rs
+17 -31
View File
@@ -143,9 +143,6 @@ use workdir::{
#[cfg(test)]
use worker::WorkerController;
use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession;
use worker::feature::builtin::{
CompositeWorkerObservationProvider, WorkspaceClientWorkerObservationProvider,
};
use worker::feature::contracts::{
WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject,
WorkerObservationSubjectRef, WorkerSessionCapture,
@@ -2161,20 +2158,15 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
if let (Some(runtime_id), Some(workspace_id)) =
(observation_runtime_id, observation_workspace_id.clone())
&& observation_enabled
&& !observation_grants.is_empty()
{
let mut providers: Vec<Arc<dyn WorkerObservationProvider>> = vec![Arc::new(
WorkspaceClientWorkerObservationProvider::new(worker.workspace_client_handle()),
)];
if !observation_grants.is_empty() {
providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider {
worker.bind_worker_observation_provider(Some(Arc::new(
RuntimeGrantedWorkerObservationProvider {
runtime_id,
workspace_id,
grants: observation_grants.into_iter().take(100).collect(),
hub: self.observation_hub.clone(),
}));
}
worker.bind_worker_observation_provider(Some(Arc::new(
CompositeWorkerObservationProvider::new(providers),
},
)));
}
if flow_transition_enabled {
@@ -2877,30 +2869,24 @@ impl ProfileRuntimeWorkerFactory {
.expect("retained restore owner")
.sessions
.push(workdir_sessions.clone());
let mut providers: Vec<Arc<dyn WorkerObservationProvider>> = vec![Arc::new(
WorkspaceClientWorkerObservationProvider::new(workspace_client.clone()),
)];
let observation_provider = if let (Some(runtime_id), Some(api)) = (
self.runtime_id.clone(),
request.request.workspace_api.as_ref(),
) && request.request.worker_observation_enabled
&& !request.request.worker_observation_grants.is_empty()
{
if !request.request.worker_observation_grants.is_empty() {
providers.push(Arc::new(RuntimeGrantedWorkerObservationProvider {
runtime_id,
workspace_id: api.workspace_id.clone(),
grants: request
.request
.worker_observation_grants
.iter()
.take(100)
.cloned()
.collect(),
hub: self.observation_hub.clone(),
}));
}
Some(Arc::new(CompositeWorkerObservationProvider::new(providers))
as Arc<dyn WorkerObservationProvider>)
Some(Arc::new(RuntimeGrantedWorkerObservationProvider {
runtime_id,
workspace_id: api.workspace_id.clone(),
grants: request
.request
.worker_observation_grants
.iter()
.take(100)
.cloned()
.collect(),
hub: self.observation_hub.clone(),
}) as Arc<dyn WorkerObservationProvider>)
} else {
None
};
crates/worker-runtime/src/worker_source.rs
+14
View File
@@ -705,6 +705,20 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
client.worker_control_list(workspace_id).await,
)
}
WorkspaceServerOperation::WorkerControlAuthorize {
runtime_id,
worker_id,
permission,
} => generated_workspace_response(
client
.worker_control_authorize(
workspace_id,
runtime_id,
worker_id,
permission,
)
.await,
),
WorkspaceServerOperation::WorkerControlSpawn(request) => {
generated_workspace_response(
client.worker_control_spawn(workspace_id, request).await,
crates/worker/src/controller.rs
+86 -1
View File
@@ -2045,7 +2045,7 @@ where
crate::feature::builtin::objective::composition::compose(
&mut feature_registry,
feature_config.objective.enabled,
workspace_client,
workspace_client.clone(),
objective_namespace.map(|route| (route, wip_permissions.clone())),
)?;
@@ -2066,6 +2066,15 @@ where
) {
observation_providers.push(provider);
}
// A scoped Backend provider belongs to the Worker Feature's composition,
// including ordinary spawn and restore. Runtime's optional local peer
// provider is not its enablement or authority: every actual capture still
// requires current operation use and the target's Backend control grant.
if worker_enabled {
observation_providers.push(Arc::new(
crate::feature::builtin::worker_observation::WorkspaceClientWorkerObservationProvider::new(workspace_client),
));
}
if let Some(provider) = host_worker_observation_provider {
observation_providers.push(provider);
}
@@ -4141,6 +4150,82 @@ mod tests {
}
}
#[tokio::test]
async fn scoped_observation_is_composed_only_with_runtime_worker_service_without_startup_access()
{
for enabled in [false, true] {
let root = TempDir::new().unwrap();
let mut manifest = manifest::WorkerManifest::from_toml(
"[worker]\nname = 'observation-wiring'\n[model]\nscheme = 'anthropic'\nmodel_id = 'test'\n[engine]\n[scope]\n"
).unwrap();
manifest.feature.worker.enabled = enabled;
manifest.feature.sub_worker.enabled = false;
manifest.feature.workdir_catalog.enabled = false;
let model = ScriptedClient::new(vec![]);
let engine =
Engine::<_, agen::state::Mutable, crate::SessionHistoryMetadata>::new_annotated(
model,
);
let store = session_store::CombinedStore::new(
session_store::FsStore::new(root.path().join("sessions")).unwrap(),
session_store::FsWorkerStore::new(root.path().join("workers")).unwrap(),
);
let operations = Arc::new(std::sync::Mutex::new(Vec::new()));
let host = Arc::new(DriveWorkspaceHost {
available: true,
workspace: Some("workspace-1"),
granted: false,
operations: operations.clone(),
});
let mut worker = Worker::new(
manifest,
engine,
store,
crate::worker::WorkerWorkspaceContext::with_client(
Some(crate::worker::WorkspaceId::new("workspace-1").unwrap()),
host,
),
crate::worker::WorkerFilesystemAuthority::None,
manifest::Scope::empty(),
)
.await
.unwrap();
worker.set_system_prompt_template(
crate::SystemPromptTemplate::parse(
"default",
crate::PromptCatalogSource::builtins_only(),
)
.unwrap(),
);
let prepared = WorkerController::initialize(worker, root.path(), None)
.await
.unwrap();
let Event::Snapshot { greeting, .. } =
crate::ipc::protocol_session::subscribe_worker_protocol_session(
&prepared.handle,
)
.snapshot_event
else {
panic!("initial committed greeting");
};
for name in [
"ViewSessionOverview",
"SearchSessionEntries",
"ReadSessionEntry",
] {
assert_eq!(
greeting.tools.iter().any(|tool| tool == name),
enabled,
"Worker service configuration must determine observation publication, not client availability alone"
);
}
assert!(
operations.lock().unwrap().is_empty(),
"provider composition may not authorize or read protected resources"
);
}
}
#[tokio::test]
async fn drive_controller_registers_without_authorization_and_checks_each_access_in_both_modes()
{
crates/worker/src/feature/builtin/manage_worker.rs
+16 -36
View File
@@ -403,46 +403,26 @@ impl WorkerControlService for WorkspaceWorkerControlService {
"Runtime Worker control is not enabled for this Worker".to_string(),
));
}
self.authorize_runtime_operation(&format!("worker.{permission}"))?;
let operation = if permission == "notify" {
"worker.messaging".to_owned()
} else {
format!("worker.{permission}")
};
self.authorize_runtime_operation(&operation)?;
let response = self.client.execute_server_operation(
crate::worker::WorkspaceServerOperation::WorkerControlList,
crate::worker::WorkspaceServerOperation::WorkerControlAuthorize {
runtime_id: runtime_id.clone(),
worker_id: worker_id.clone(),
permission: permission.into(),
},
)?;
if !response.is_success() {
return Err(WorkspaceClientError::Request(format!(
"Workspace control request returned {}: {}",
response.status, response.body
)));
}
let body: serde_json::Value =
serde_json::from_str(&response.body).map_err(|error| {
WorkspaceClientError::Request(format!(
"invalid Workspace control response: {error}"
))
})?;
let granted = body
.get("items")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
.any(|item| {
item.get("subject")
.and_then(|value| value.get("runtime_id"))
== Some(&serde_json::Value::String(runtime_id.clone()))
&& item.get("subject").and_then(|value| value.get("worker_id"))
== Some(&serde_json::Value::String(worker_id.clone()))
&& item
.get("permissions")
.and_then(serde_json::Value::as_array)
.is_some_and(|permissions| {
permissions.iter().any(|candidate| candidate == permission)
})
});
if granted {
if response.is_success() {
Ok(())
} else {
Err(WorkspaceClientError::Request(
"unknown Worker or permission not granted".to_string(),
))
Err(WorkspaceClientError::Request(format!(
"Worker target permission check returned {}: {}",
response.status, response.body
)))
}
}
}
crates/worker/src/worker.rs
+15
View File
@@ -762,6 +762,11 @@ pub enum WorkspaceServerOperation {
result_id: String,
},
WorkerControlList,
WorkerControlAuthorize {
runtime_id: String,
worker_id: String,
permission: String,
},
WorkerControlSpawn(server_api::CreateWorkspaceWorkerRequest),
WorkerControlInput {
runtime_id: String,
@@ -959,6 +964,16 @@ fn workspace_server_operation_request(
WorkspaceServerOperation::WorkerControlList => Ok(WorkspaceRequest::get(format!(
"{base}/worker-control/workers"
))),
WorkspaceServerOperation::WorkerControlAuthorize {
runtime_id,
worker_id,
permission,
} => Ok(WorkspaceRequest::get(format!(
"{base}/worker-control/workers/{}/{}/permissions/{}",
encode_workspace_path_segment(&runtime_id),
encode_workspace_path_segment(&worker_id),
encode_workspace_path_segment(&permission)
))),
WorkspaceServerOperation::WorkerControlSpawn(request) => {
workspace_server_json_request(format!("{base}/worker-control/workers"), &request)
}
crates/workspace-server/src/server.rs
+43
View File
@@ -7808,6 +7808,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::worker_control_list(
service.clone(),
))
.merge(server_api::server_api_axum::worker_control_authorize(
service.clone(),
))
.merge(server_api::server_api_axum::worker_control_spawn(
service.clone(),
))
@@ -10295,6 +10298,46 @@ impl server_api::ServerApi for ServerApiContractService {
Ok(response)
}
async fn worker_control_authorize(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
runtime_id: String,
worker_id: String,
permission: String,
) -> std::result::Result<(), server_api::RepositoryApiError> {
let api = self.workspace_api()?.clone();
let headers = current_worker_contract_headers(&context)?;
api.host_executors
.host()
.map_err(|error| ApiError::from(error).into_repository_api_error())?
.run(move || -> ApiResult<()> {
validate_workspace_scope(&api, &workspace_id)?;
let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?;
if !matches!(
permission.as_str(),
"observe" | "notify" | "send_input" | "cancel" | "stop" | "restore" | "remove"
) {
return Err(
Error::InvalidInput("unknown Worker control permission".into()).into(),
);
}
// This is a current, target-only preflight for providers that read
// locally. It discloses no registry/session data and issues no
// reusable receipt. Actual Backend effects still recheck authority.
authorize_known_worker_permission(
&api,
&workspace_id,
&RuntimeWorkerRef::new(source.runtime_id, source.worker_id),
&RuntimeWorkerRef::new(runtime_id, worker_id),
&permission,
)?;
Ok(())
})
.await
.map_err(ApiError::into_repository_api_error)
}
async fn worker_control_spawn(
&self,
context: server_api::ServerRequestContext,
crates/workspace-server/src/server/tests/subject_authorization_tests.rs
+3 -1
View File
@@ -2753,7 +2753,9 @@ impl ConnectedWorkerHarness {
let errors = captured_errors.clone();
async move {
let response = next.run(request).await;
if response.status().is_success() {
if !response.status().is_client_error()
&& !response.status().is_server_error()
{
return response;
}
let (parts, body) = response.into_parts();
crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs
+22 -6
View File
@@ -139,11 +139,21 @@ impl agen::llm_client::LlmClient for MessagingModel {
.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}),
)
if result(&request, "roundtrip-observe").is_some() {
call(
"roundtrip-ping",
"WorkerNotify",
json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]},"content":PING}),
)
} else {
// The origin has observe, not worker.list. A known target must
// remain readable without collection discovery authority.
call(
"roundtrip-observe",
"ViewSessionOverview",
json!({"subject":{"kind":"runtime_worker","runtime_id":child["runtime_id"],"worker_id":child["worker_id"]}}),
)
}
} else {
let call_id = if let Some(repair) = &self.repair {
if let Some((is_error, summary)) =
@@ -373,6 +383,7 @@ async fn connected_roundtrip(explicit_workdir: bool, repair: bool) {
collection("worker.create"),
operation("worker.messaging"),
operation("worker.stop"),
operation("worker.observe"),
]);
let (status, assigned) = response(api, human_request(&harness.owner,"POST",&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),Some(json!({"operation_id":"public-messaging-policy","expected_change_id":null,"execution":execution,"delegation":permissions.lock().unwrap().clone()})))).await;
assert_eq!(status, StatusCode::OK, "{assigned}");
@@ -582,7 +593,12 @@ async fn connected_roundtrip(explicit_workdir: bool, repair: bool) {
.contains("MUST_NOT_ACCEPT_REVOKED_CONTROL")
);
let calls = harness.authorizations.lock().unwrap();
for op in ["worker.create", "worker.messaging", "worker.stop"] {
for op in [
"worker.create",
"worker.messaging",
"worker.stop",
"worker.observe",
] {
assert!(
calls
.iter()
crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs
+161 -4
View File
@@ -238,6 +238,14 @@ async fn assert_backend_denied(
async fn worker_tool_host(
root: &Path,
client: Arc<dyn WorkspaceClient>,
) -> worker::Worker<UnusedModel, session_store::FsStore> {
worker_tool_host_with_observation(root, client, None).await
}
async fn worker_tool_host_with_observation(
root: &Path,
client: Arc<dyn WorkspaceClient>,
observation: Option<Arc<dyn worker::feature::contracts::WorkerObservationProvider>>,
) -> worker::Worker<UnusedModel, session_store::FsStore> {
let manifest = worker::WorkerManifest::from_toml(
r#"
@@ -273,10 +281,14 @@ async fn worker_tool_host(
)
.await
.unwrap();
let report = host.install_features(
worker::feature::FeatureRegistryBuilder::new()
.with_module(manage_worker_feature(client, None, true)),
);
let mut features = worker::feature::FeatureRegistryBuilder::new()
.with_module(manage_worker_feature(client, None, true));
if let Some(provider) = observation {
features = features.with_module(
worker::feature::builtin::worker_observation::WorkerObservationFeature::new(provider),
);
}
let report = host.install_features(features);
assert!(
!report.has_errors(),
"Worker Feature must install without granting operation use"
@@ -332,6 +344,151 @@ fn tool_arguments(name: &str, target: &RuntimeWorkerRef) -> Value {
args
}
// Only the protected Session read is substituted here. Installed observation /
// control Features and signed Host / Backend target checks remain production code.
struct CountedObservation(std::sync::atomic::AtomicUsize);
#[async_trait::async_trait]
impl worker::feature::contracts::WorkerObservationProvider for CountedObservation {
async fn list_worker_sessions(
&self,
) -> std::result::Result<
Vec<worker::feature::contracts::WorkerObservationSubject>,
worker::feature::contracts::WorkerObservationError,
> {
panic!("known-target capture must not discover the collection")
}
async fn capture_worker_session(
&self,
_: &worker::feature::contracts::WorkerObservationSubjectRef,
) -> std::result::Result<
worker::feature::contracts::WorkerSessionCapture,
worker::feature::contracts::WorkerObservationError,
> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(worker::feature::contracts::WorkerSessionCapture {
segment_id: "target-session".into(),
session: protocol::SessionSnapshot {
pending_submissions: Default::default(),
entries: vec![protocol::SessionSnapshotEntry {
use_policy: None,
entry_id: "observed-entry".into(),
timestamp: 1,
provenance: protocol::SessionEntryProvenance::LegacyUnknown,
derived_from: vec![],
data: protocol::SessionSnapshotEntryData::Message {
role: protocol::SessionMessageRole::User,
content: vec![protocol::SessionContentPart::Text {
text: "observed input".into(),
}],
},
}],
},
})
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn installed_observation_needs_only_observe_and_target_grant_not_collection_discovery() {
let f = OperationFixture::new().await;
f.assign(vec![workspace_permission("worker.create")], vec![])
.await;
let target = f.child("observed-target", vec![]).await;
f.assign(vec![operation_permission("worker.observe")], vec![])
.await;
let provider = Arc::new(CountedObservation(std::sync::atomic::AtomicUsize::new(0)));
let client = f.client(&f.source);
let host = worker_tool_host_with_observation(
&f._dir.path().join("observation-tools"),
client.clone(),
Some(provider.clone()),
)
.await;
assert!(
execute_tool_result(&host, "WorkerList", json!({})).is_err(),
"observe never grants collection discovery"
);
let overview = execute_tool(
&host,
"ViewSessionOverview",
json!({"subject":subject_value(&target)}),
)
.await;
let entry_ref = overview["entries"][0]["entry_ref"].as_str().unwrap();
let tools = [
(
"ViewSessionOverview",
json!({"subject":subject_value(&target)}),
),
(
"SearchSessionEntries",
json!({"subject":subject_value(&target), "query":"observed"}),
),
(
"ReadSessionEntry",
json!({"subject":subject_value(&target), "entry_ref":entry_ref}),
),
];
client.requests.lock().unwrap().clear();
for (name, input) in &tools {
execute_tool(&host, name, input.clone()).await;
}
assert_eq!(provider.0.load(std::sync::atomic::Ordering::SeqCst), 4);
assert!(
client
.requests
.lock()
.unwrap()
.iter()
.all(|(_, path)| !path.ends_with("/worker-control/workers")),
"known-target public observation must not invoke WorkerList"
);
let grant = f
.api
.store
.get_active_worker_control_grant(TEST_WORKSPACE_ID, &f.source, &target)
.unwrap()
.unwrap();
for missing in ["operation", "target", "revoked"] {
match missing {
"operation" => {
f.assign(vec![], vec![]).await;
}
"target" => {
f.assign(vec![operation_permission("worker.observe")], vec![])
.await;
f.api
.store
.revoke_worker_control_grant(
TEST_WORKSPACE_ID,
&grant.grant_id,
&now_registry_timestamp(),
)
.unwrap();
}
"revoked" => {}
_ => unreachable!(),
}
for (name, input) in &tools {
let mut input = input.clone();
if missing == "target" {
input["subject"]["worker_id"] = json!("unknown-target");
}
let error = execute_tool_result(&host, name, input)
.unwrap_err()
.to_string();
assert!(
error.contains("not found or is not accessible"),
"{missing} {name}: {error}"
);
}
assert_eq!(
provider.0.load(std::sync::atomic::Ordering::SeqCst),
4,
"denied {missing} must not read or supply protected Session data"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn backend_control_grant_does_not_replace_exact_operation_use_permission() {
let f = OperationFixture::new().await;