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

Separate Worker operation authorization from directional control grants

Browse Source
This commit is contained in:
Hare
2026-10-29 22:26:35 +09:00
parent 7ad6030498
commit 74ddd568eb
15 changed files with 2650 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+82 -75
View File
@@ -2066,6 +2066,7 @@ struct WorkerRemovalService {
journal_storage: crate::WorkspaceFeatureStorage,
journal_registration: crate::RegisteredFeature,
journal_dispatch_api: Weak<WorkspaceApi>,
worker_authority: subject_authorization::WorkerAuthorityView,
store: Arc<dyn ControlPlaneStore>,
host_executors: HostExecutors,
runtime: Weak<RuntimeRegistry>,
@@ -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<tokio::sync::M
)
}
fn worker_control_denied_response(error: Error) -> 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<WorkerControlGrantRecord> {
// 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(),
crates/workspace-server/src/server/subject_authorization.rs
+333 -125
View File
@@ -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<String>) -> ApiError {
Error::WorkspacePermissionDenied(message.into()).into()
}
fn policy(api: &WorkspaceApi, subject_id: &str) -> ApiResult<Option<SubjectPermissions>> {
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<dyn ControlPlaneStore>,
config_store: Arc<crate::SqliteWorkspaceStore>,
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> {
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<Option<SubjectPermissions>> {
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<Option<String>> {
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<Option<WorkerExecutionGrant>> {
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(&current.origin_runtime_id, &current.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<Option<Vec<ScopedPermission>>> {
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<WorkerControlGrantRecord> {
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<Option<SubjectPermissions>> {
WorkerAuthorityView::from(api).policy(subject_id)
}
fn current_body_subject(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> ApiResult<Option<String>> {
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<Option<WorkerExecutionGrant>> {
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(&current.origin_runtime_id, &current.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<Option<Vec<ScopedPermission>>> {
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
crates/workspace-server/src/server/tests/subject_authorization_tests.rs
+254 -197
View File
@@ -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<ScopedPermission> {
},
),
(
"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<Mutex<Vec<(Value, StatusCode)>>>,
runtime_errors: Arc<Mutex<Vec<String>>>,
_backend_server: AuthorizationCallbackServer,
_runtime_server: ConnectedRuntimeServer,
_dir: tempfile::TempDir,
}
impl ConnectedWorkerHarness {
async fn start<M: agen::llm_client::LlmClient + Clone + 'static>(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::<String>::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::<String>::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;
crates/workspace-server/src/server/tests/worker_messaging_connected_tests.rs
+621
View File
@@ -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<Value> {
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<Event> {
vec![
Event::tool_use_start(0, id, name),
Event::tool_input_delta(0, input.to_string()),
Event::tool_use_stop(0),
]
}
fn done() -> Vec<Event> {
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<bool>,
spawn_refusal: tokio::sync::mpsc::Sender<()>,
allow_replay: tokio::sync::Notify,
origin: Mutex<Option<RuntimeWorkerRef>>,
origin_ready: tokio::sync::Notify,
}
#[derive(Clone)]
struct MessagingModel {
child_permissions: Arc<Mutex<Vec<ScopedPermission>>>,
workdir: Arc<Mutex<Option<String>>>,
completed: tokio::sync::mpsc::Sender<Value>,
child_replied: Arc<tokio::sync::Notify>,
repair: Option<Arc<RepairSignals>>,
}
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for MessagingModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
request: agen::llm_client::Request,
) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
{
let messages = request
.items
.iter()
.filter_map(|item| match item {
Item::Message { content, .. } => Some(serde_json::to_string(content).unwrap()),
_ => None,
})
.collect::<Vec<_>>()
.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::<Value>(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::<Vec<_>>();
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::<Vec<_>>();
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;
}
crates/workspace-server/src/server/tests/worker_operation_authorization_tests.rs
+788
View File
@@ -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<std::sync::atomic::AtomicUsize>,
}
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<ScopedPermission> {
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<StatusCode>,
) {
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<dyn WorkspaceClient>,
) -> worker::Worker<UnusedModel, session_store::FsStore> {
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<UnusedModel, session_store::FsStore>,
name: &str,
) -> Arc<dyn Tool> {
host.engine()
.tool_server_handle()
.get_tool(name)
.unwrap_or_else(|| panic!("missing installed public {name}"))
.1
}
fn execute_tool_result(
host: &worker::Worker<UnusedModel, session_store::FsStore>,
name: &str,
input: Value,
) -> std::result::Result<ToolOutput, agen::tool::ToolError> {
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<UnusedModel, session_store::FsStore>,
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,
&current,
&target,
&[
"notify",
"send_input",
"cancel",
"stop",
"restore",
"remove",
"observe",
],
);
assert_eq!(
backend_operation(&f, &current, &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, &current, &target, None).await;
assert_backend_denied(&f, &f.source, &target, Some(StatusCode::UNAUTHORIZED)).await;
}