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

Authorize Feature effects and protected input at operation boundaries

Browse Source 
Remove non-MCP activation gates without expanding legacy grants. Bind resource resolution, background Job attenuation, protected prompt provenance and repeated model supply to current Host authority. Align generated contracts and regression fixtures.
This commit is contained in:
Hare
2026-10-28 11:44:34 +09:00
parent 618880d38d
commit 960a1b4364
75 changed files with 13191 additions and 1143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/notification_diagnostics.rs
+59
View File
@@ -6,6 +6,65 @@ use server_api::{
NotificationDiagnosticsQuery,
};
pub(super) fn source_resource(
store: &SqliteWorkspaceStore,
workspace: &str,
source_event_id: &str,
) -> Result<Option<server_api::authorization::PermissionResource>> {
use server_api::authorization::PermissionResource;
if source_event_id.is_empty()
|| source_event_id.len() > 512
|| source_event_id.chars().any(char::is_control)
{
return Ok(None);
}
store.with_conn(|conn| {
// Read selectors only, not facts_json, requests, rendered content or Job
// results. Legacy arbitrary delivery IDs are retained as input_event_id;
// neither their spelling nor a caller's Job filter establishes ownership.
let mut statement = conn.prepare(
"SELECT DISTINCT
CASE WHEN COALESCE(json_extract(facts_json,'$.job_id'),json_extract(facts_json,'$.backend_job.job_id')) IS NOT NULL THEN 'job'
WHEN json_extract(facts_json,'$.merge_request_id') IS NOT NULL THEN 'mr'
ELSE 'ticket' END,
COALESCE(json_extract(facts_json,'$.job_id'),json_extract(facts_json,'$.backend_job.job_id'),json_extract(facts_json,'$.merge_request_id'),ticket_id)
FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2 LIMIT 2",
)?;
let targets = statement
.query_map(params![workspace, source_event_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
if targets.is_empty() {
// Uncaptured MR origins still have a concrete resource selector, so
// diagnostics can distinguish awaiting capture from absent origin.
// Persisted ownership always wins: legacy Job IDs may look like MR IDs.
let mr = source_event_id.strip_prefix("mr-event:")
.or_else(|| source_event_id.strip_prefix("mr-source:"))
.and_then(|value| value.split_once(':'))
.map(|(mr, _)| mr);
if let Some(mr) = mr {
let resource = PermissionResource::MergeRequest {
workspace_id: workspace.into(), merge_request_id: mr.into(),
};
return Ok(resource.validate().is_ok().then_some(resource));
}
}
let [(kind, Some(id))] = targets.as_slice() else {
return Ok(None);
};
let resource = match kind.as_str() {
"job" => PermissionResource::Job { workspace_id: workspace.into(), job_id: id.clone() },
"mr" => PermissionResource::MergeRequest { workspace_id: workspace.into(), merge_request_id: id.clone() },
_ => PermissionResource::Ticket { workspace_id: workspace.into(), ticket_id: id.clone() },
};
if resource.validate().is_err() {
return Ok(None);
}
Ok(Some(resource))
})
}
fn source_checks(
conn: &Connection,
workspace: &str,
crates/workspace-server/src/notification_diagnostics_tests.rs
+364 -85
View File
@@ -1,31 +1,72 @@
// Production Store and SQLite. No Runtime/model side effects are required for diagnostics.
#[tokio::test]
async fn notification_diagnostics_preserves_unknown_acceptance_and_marks_replacement_ambiguity() {
use server_api::{NotificationDiagnosticsQuery as Q,NotificationDiagnosticStage as S,NotificationTargetResolution as T};
for initial_failure in [NotificationFailureCode::AcceptanceUnknown,NotificationFailureCode::AcknowledgementFailed] {
let (_dir,store)=tracking_store_fixture().await;
tracking_store_assign(&store,"a","assign-a");
let source=tracking_store_source();
let delivery=delivery_store_reserve_initial(&store,&source);
store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"frozen body",&source).unwrap();
let key=NotificationDeliveryKey::AssignmentInput{assignment_id:"a".into(),input_event_id:delivery.input_event_id.clone()};
store.record_notification_delivery_failure("w",&key,initial_failure).unwrap();
store.record_notification_delivery_failure("w",&key,NotificationFailureCode::TargetStopped).unwrap();
let query=Q{delivery_id:Some(delivery.delivery_id.clone()),..Default::default()};
let read=||store.notification_diagnostics("w",&query).unwrap().items.remove(0);
assert_eq!(read().stage,S::AcceptanceUnknown);
use server_api::{
NotificationDiagnosticStage as S, NotificationDiagnosticsQuery as Q,
NotificationTargetResolution as T,
};
for initial_failure in [
NotificationFailureCode::AcceptanceUnknown,
NotificationFailureCode::AcknowledgementFailed,
] {
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let source = tracking_store_source();
let delivery = delivery_store_reserve_initial(&store, &source);
store
.prepare_ticket_assignment_delivery_content(
"w",
&delivery.delivery_id,
"frozen body",
&source,
)
.unwrap();
let key = NotificationDeliveryKey::AssignmentInput {
assignment_id: "a".into(),
input_event_id: delivery.input_event_id.clone(),
};
store
.record_notification_delivery_failure("w", &key, initial_failure)
.unwrap();
store
.record_notification_delivery_failure("w", &key, NotificationFailureCode::TargetStopped)
.unwrap();
let query = Q {
delivery_id: Some(delivery.delivery_id.clone()),
..Default::default()
};
let read = || {
store
.notification_diagnostics("w", &query)
.unwrap()
.items
.remove(0)
};
assert_eq!(read().stage, S::AcceptanceUnknown);
assert!(read().acceptance_outcome_unknown);
assert_eq!(read().last_failure_code.as_deref(),Some("target_stopped"));
assert_eq!(read().last_failure_code.as_deref(), Some("target_stopped"));
store.with_conn(|conn|{conn.execute("UPDATE worker_singleton_owners SET worker_id='replacement' WHERE workspace_id='w'",[])?;Ok(())}).unwrap();
let replacement=TicketAssignmentTrackingSourceContext{worker_id:"replacement".into(),..source};
store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"ignored new body",&replacement).unwrap();
let ambiguous=read();
assert_eq!(ambiguous.target_resolution,T::MultipleAuthorizedTargets);
let replacement = TicketAssignmentTrackingSourceContext {
worker_id: "replacement".into(),
..source
};
store
.prepare_ticket_assignment_delivery_content(
"w",
&delivery.delivery_id,
"ignored new body",
&replacement,
)
.unwrap();
let ambiguous = read();
assert_eq!(ambiguous.target_resolution, T::MultipleAuthorizedTargets);
assert!(ambiguous.target_runtime_id.is_none() && ambiguous.target_worker_id.is_none());
store.ack_ticket_assignment_delivery("w",&delivery.delivery_id,&replacement).unwrap();
let accepted=read();
assert_eq!(accepted.target_resolution,T::RuntimeAccepted);
assert_eq!(accepted.target_worker_id.as_deref(),Some("replacement"));
store
.ack_ticket_assignment_delivery("w", &delivery.delivery_id, &replacement)
.unwrap();
let accepted = read();
assert_eq!(accepted.target_resolution, T::RuntimeAccepted);
assert_eq!(accepted.target_worker_id.as_deref(), Some("replacement"));
assert!(!accepted.acceptance_outcome_unknown);
}
}
@@ -33,8 +74,8 @@ async fn notification_diagnostics_preserves_unknown_acceptance_and_marks_replace
#[tokio::test]
async fn notification_diagnostics_detects_exact_missing_mr_event_after_earlier_capture() {
use server_api::NotificationDiagnosticsQuery as Q;
let (_dir,store)=tracking_store_fixture().await;
tracking_store_assign(&store,"a","assign-a");
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
store.with_conn(|conn| {
conn.execute("INSERT INTO merge_requests(workspace_id,merge_request_id,repository_id,state,selector_to,created_at,updated_at) SELECT workspace_id,'mr-diagnostic',repository_id,'open','develop','1','1' FROM repositories WHERE workspace_id='w' LIMIT 1",[])?;
for (event,seq) in [("first",1),("missing",2)] {
@@ -43,107 +84,345 @@ async fn notification_diagnostics_detects_exact_missing_mr_event_after_earlier_c
conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,'mr-event:mr-diagnostic:first',role,assignment_id,'mr_review',json_set(facts_json,'$.merge_request_id','mr-diagnostic') FROM notification_sources WHERE assignment_id='a' LIMIT 1",[])?;
Ok(())
}).unwrap();
let page=store.notification_diagnostics("w",&Q{merge_request_id:Some("mr-diagnostic".into()),source_event_id:Some("mr-event:mr-diagnostic:missing".into()),..Default::default()}).unwrap();
let page = store
.notification_diagnostics(
"w",
&Q {
merge_request_id: Some("mr-diagnostic".into()),
source_event_id: Some("mr-event:mr-diagnostic:missing".into()),
..Default::default()
},
)
.unwrap();
assert!(page.items.is_empty());
assert!(page.source_checks[0].captured);
assert_eq!(page.source_checks[1].origin_present,Some(true));
assert_eq!(page.source_checks[1].origin_present, Some(true));
assert!(!page.source_checks[1].captured);
}
#[tokio::test]
async fn notification_diagnostics_bounds_saved_reasons_without_returning_source_payloads() {
use server_api::NotificationDiagnosticsQuery as Q;
let (_dir,store)=tracking_store_fixture().await;
tracking_store_assign(&store,"a","assign-a");
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
store.with_conn(|conn| {
conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) SELECT sequence,workspace_id,'bounded-reason','suppress',?1,'attention' FROM notification_sources WHERE assignment_id='a' LIMIT 1",["長".repeat(2000)])?;
Ok(())
}).unwrap();
let page=store.notification_diagnostics("w",&Q{delivery_id:Some("bounded-reason".into()),..Q::default()}).unwrap();
assert_eq!(page.items[0].reason.as_ref().unwrap().chars().count(),512);
let page = store
.notification_diagnostics(
"w",
&Q {
delivery_id: Some("bounded-reason".into()),
..Q::default()
},
)
.unwrap();
assert_eq!(page.items[0].reason.as_ref().unwrap().chars().count(), 512);
assert!(!serde_json::to_string(&page).unwrap().contains("facts_json"));
}
#[tokio::test]
async fn notification_diagnostics_detects_unconnected_origin_and_retains_history_without_pending_work() {
use server_api::{NotificationDiagnosticsQuery as Q,NotificationDiagnosticStage as S};
let (_dir,store)=tracking_store_fixture().await;
tracking_store_assign(&store,"a","assign-a");
let delivery=delivery_store_reserve_initial(&store,&tracking_store_source());
store.clear_current_ticket_role_assignment("w","tracking",TicketAssignmentRole::Orchestrator,"a","clear-a","clear-a","owner","2",None).unwrap();
let page=store.notification_diagnostics("w",&Q{delivery_id:Some(delivery.delivery_id),..Q::default()}).unwrap();
assert_eq!(page.items[0].stage,S::Historical);
assert_eq!(page.items[0].recipient_current,Some(false));
async fn notification_diagnostics_detects_unconnected_origin_and_retains_history_without_pending_work()
{
use server_api::{NotificationDiagnosticStage as S, NotificationDiagnosticsQuery as Q};
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
let delivery = delivery_store_reserve_initial(&store, &tracking_store_source());
store
.clear_current_ticket_role_assignment(
"w",
"tracking",
TicketAssignmentRole::Orchestrator,
"a",
"clear-a",
"clear-a",
"owner",
"2",
None,
)
.unwrap();
let page = store
.notification_diagnostics(
"w",
&Q {
delivery_id: Some(delivery.delivery_id),
..Q::default()
},
)
.unwrap();
assert_eq!(page.items[0].stage, S::Historical);
assert_eq!(page.items[0].recipient_current, Some(false));
assert!(!page.items[0].retry_candidate);
// Simulate a missing producer adapter; a diagnostic read must not repair or hide it.
store.with_conn(|conn|{conn.execute_batch("DROP TRIGGER notification_ticket_event")?;Ok(())}).unwrap();
delivery_store_external_event(&store,999,"comment",None);
store.with_conn(|conn|{conn.execute_batch("PRAGMA query_only=ON")?;Ok(())}).unwrap();
let query=Q{ticket:Some("tracking".into()),source_event_id:Some("ticket-event:999".into()),..Q::default()};
let page=store.notification_diagnostics("w",&query).unwrap();
store
.with_conn(|conn| {
conn.execute_batch("DROP TRIGGER notification_ticket_event")?;
Ok(())
})
.unwrap();
delivery_store_external_event(&store, 999, "comment", None);
store
.with_conn(|conn| {
conn.execute_batch("PRAGMA query_only=ON")?;
Ok(())
})
.unwrap();
let query = Q {
ticket: Some("tracking".into()),
source_event_id: Some("ticket-event:999".into()),
..Q::default()
};
let page = store.notification_diagnostics("w", &query).unwrap();
assert!(page.items.is_empty());
assert_eq!(page.source_checks[0].origin_present,Some(true));
assert_eq!(page.source_checks[0].origin_present, Some(true));
assert!(!page.source_checks[0].captured);
let other=store.notification_diagnostics("other",&query).unwrap();
assert_eq!(other.source_checks[0].origin_present,Some(false));
let other = store.notification_diagnostics("other", &query).unwrap();
assert_eq!(other.source_checks[0].origin_present, Some(false));
assert!(!other.source_checks[0].captured);
}
#[tokio::test]
async fn notification_diagnostics_is_readonly_paginated_and_workspace_scoped() {
use server_api::{NotificationDiagnosticsQuery as Q, NotificationDiagnosticStage as S};
use server_api::{NotificationDiagnosticStage as S, NotificationDiagnosticsQuery as Q};
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store, "a", "assign-a");
delivery_store_answer(&store, "answer-a");
let before = store.with_conn(|conn| Ok((conn.total_changes(),stored_fixture_rows(conn,"notification_decisions")))).unwrap();
store.with_conn(|conn| { conn.execute_batch("PRAGMA query_only=ON")?; Ok(()) }).unwrap();
let first = store.notification_diagnostics("w", &Q{limit:Some(1), ..Q::default()}).unwrap();
assert_eq!(first.items.len(),1);
assert_eq!(first.items[0].stage,S::AwaitingRouting);
let next = store.notification_diagnostics("w", &Q{cursor:first.next_cursor.clone(), ..Q::default()}).unwrap();
assert!(next.items.iter().all(|i| i.source_id != first.items[0].source_id));
let before = store
.with_conn(|conn| {
Ok((
conn.total_changes(),
stored_fixture_rows(conn, "notification_decisions"),
))
})
.unwrap();
store
.with_conn(|conn| {
conn.execute_batch("PRAGMA query_only=ON")?;
Ok(())
})
.unwrap();
let first = store
.notification_diagnostics(
"w",
&Q {
limit: Some(1),
..Q::default()
},
)
.unwrap();
assert_eq!(first.items.len(), 1);
assert_eq!(first.items[0].stage, S::AwaitingRouting);
let next = store
.notification_diagnostics(
"w",
&Q {
cursor: first.next_cursor.clone(),
..Q::default()
},
)
.unwrap();
assert!(
next.items
.iter()
.all(|i| i.source_id != first.items[0].source_id)
);
assert!(!next.items.is_empty());
assert!(store.notification_diagnostics("other-workspace", &Q{cursor:Some(first.items[0].source_id.clone()), ..Q::default()}).is_err());
assert!(store.notification_diagnostics("w", &Q{ticket:Some("missing".into()), cursor:Some(first.items[0].source_id.clone()), ..Q::default()}).is_err());
for limit in [0,101] { assert!(store.notification_diagnostics("w", &Q{limit:Some(limit), ..Q::default()}).is_err()); }
assert!(store.notification_diagnostics("w", &Q{cursor:Some("not-a-source".into()), ..Q::default()}).is_err());
assert_eq!(store.with_conn(|conn| Ok((conn.total_changes(),stored_fixture_rows(conn,"notification_decisions")))).unwrap(),before);
store.with_conn(|conn| { conn.execute_batch("PRAGMA query_only=OFF")?; Ok(()) }).unwrap();
assert!(
store
.notification_diagnostics(
"other-workspace",
&Q {
cursor: Some(first.items[0].source_id.clone()),
..Q::default()
}
)
.is_err()
);
assert!(
store
.notification_diagnostics(
"w",
&Q {
ticket: Some("missing".into()),
cursor: Some(first.items[0].source_id.clone()),
..Q::default()
}
)
.is_err()
);
for limit in [0, 101] {
assert!(
store
.notification_diagnostics(
"w",
&Q {
limit: Some(limit),
..Q::default()
}
)
.is_err()
);
}
assert!(
store
.notification_diagnostics(
"w",
&Q {
cursor: Some("not-a-source".into()),
..Q::default()
}
)
.is_err()
);
assert_eq!(
store
.with_conn(|conn| Ok((
conn.total_changes(),
stored_fixture_rows(conn, "notification_decisions")
)))
.unwrap(),
before
);
store
.with_conn(|conn| {
conn.execute_batch("PRAGMA query_only=OFF")?;
Ok(())
})
.unwrap();
}
#[tokio::test]
async fn notification_diagnostics_preserves_suppression_failure_and_acceptance_boundaries() {
use server_api::{NotificationDiagnosticsQuery as Q, NotificationDiagnosticStage as S};
use server_api::{NotificationDiagnosticStage as S, NotificationDiagnosticsQuery as Q};
let (_dir, store) = tracking_store_fixture().await;
tracking_store_assign(&store,"a","assign-a");
let pending = store.list_pending_ticket_assignment_deliveries("w",100).unwrap();
assert_eq!(pending.len(),1);
let query = Q{assignment_id:Some("a".into()), source_event_id:Some(pending[0].input_event_id.clone()), ..Q::default()};
let read = || store.notification_diagnostics("w", &query).unwrap().items.remove(0);
assert_eq!(read().stage,S::Pending);
tracking_store_assign(&store, "a", "assign-a");
let pending = store
.list_pending_ticket_assignment_deliveries("w", 100)
.unwrap();
assert_eq!(pending.len(), 1);
let query = Q {
assignment_id: Some("a".into()),
source_event_id: Some(pending[0].input_event_id.clone()),
..Q::default()
};
let read = || {
store
.notification_diagnostics("w", &query)
.unwrap()
.items
.remove(0)
};
assert_eq!(read().stage, S::Pending);
let source = tracking_store_source();
let delivery = delivery_store_reserve_initial(&store,&source);
assert_eq!(read().stage,S::Reserved);
store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"PRIVATE-NOTIFICATION-CONTENT",&source).unwrap();
assert_eq!(read().stage,S::RequestPersisted);
let key = NotificationDeliveryKey::AssignmentInput{assignment_id:delivery.assignment_id.clone(),input_event_id:delivery.input_event_id.clone()};
store.record_notification_delivery_failure("w",&key,NotificationFailureCode::RuntimeRejected).unwrap();
assert_eq!(read().stage,S::RuntimeRejected);
store.record_notification_delivery_failure("w",&key,NotificationFailureCode::AcceptanceUnknown).unwrap();
assert_eq!(read().stage,S::AcceptanceUnknown);
store.ack_ticket_assignment_delivery("w",&delivery.delivery_id,&source).unwrap();
let delivery = delivery_store_reserve_initial(&store, &source);
assert_eq!(read().stage, S::Reserved);
store
.prepare_ticket_assignment_delivery_content(
"w",
&delivery.delivery_id,
"PRIVATE-NOTIFICATION-CONTENT",
&source,
)
.unwrap();
assert_eq!(read().stage, S::RequestPersisted);
let key = NotificationDeliveryKey::AssignmentInput {
assignment_id: delivery.assignment_id.clone(),
input_event_id: delivery.input_event_id.clone(),
};
store
.record_notification_delivery_failure("w", &key, NotificationFailureCode::RuntimeRejected)
.unwrap();
assert_eq!(read().stage, S::RuntimeRejected);
store
.record_notification_delivery_failure("w", &key, NotificationFailureCode::AcceptanceUnknown)
.unwrap();
assert_eq!(read().stage, S::AcceptanceUnknown);
store
.ack_ticket_assignment_delivery("w", &delivery.delivery_id, &source)
.unwrap();
let accepted = read();
assert_eq!(accepted.stage,S::RuntimeAccepted);
assert_eq!(accepted.stage, S::RuntimeAccepted);
assert!(!accepted.retry_candidate && !accepted.model_processing_confirmed);
assert_eq!(accepted.target_worker_id.as_deref(),Some(source.worker_id.as_str()));
assert_eq!(
accepted.target_worker_id.as_deref(),
Some(source.worker_id.as_str())
);
let serialized = serde_json::to_string(&accepted).unwrap();
assert!(!serialized.contains("PRIVATE-NOTIFICATION-CONTENT"));
assert!(!serialized.contains(&source.authentication_binding_id));
let all = store.notification_diagnostics("w",&Q::default()).unwrap();
assert!(all.items.iter().any(|i| i.stage==S::Suppressed && i.reason.as_deref()==Some("no_recipient")));
let all = store.notification_diagnostics("w", &Q::default()).unwrap();
assert!(
all.items
.iter()
.any(|i| i.stage == S::Suppressed && i.reason.as_deref() == Some("no_recipient"))
);
store.with_conn_mut(|conn| {
conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,'future-event',role,assignment_id,'unregistered-event',facts_json FROM notification_sources WHERE assignment_id='a' LIMIT 1",[])?;
notification_routing::route_pending(conn,"w")
}).unwrap();
let unknown = store.notification_diagnostics("w",&Q{source_event_id:Some("future-event".into()), ..Q::default()}).unwrap();
assert_eq!(unknown.items[0].stage,S::Deferred);
assert_eq!(unknown.items[0].reason.as_deref(),Some("unknown_event"));
let unknown = store
.notification_diagnostics(
"w",
&Q {
source_event_id: Some("future-event".into()),
..Q::default()
},
)
.unwrap();
assert_eq!(unknown.items[0].stage, S::Deferred);
assert_eq!(unknown.items[0].reason.as_deref(), Some("unknown_event"));
assert!(!unknown.items[0].retry_candidate);
}
#[tokio::test]
async fn notification_source_metadata_resolves_hashed_and_legacy_identities_without_job_body() {
use server_api::authorization::PermissionResource;
let (_dir, store) = tracking_store_fixture().await;
let (_, source) = job_notice_fixture(&store, "metadata-only", false);
let job = store.get_backend_job("w", "metadata-only").unwrap().unwrap();
let retired = store.reclaim_backend_job_input("w", "metadata-only", &job.input_digest, crate::backend_job::JobInputRetirementReason::ExplicitErasure).unwrap();
assert!(retired.load_executable_request().is_err());
store.with_conn(|conn| {
// Retired protected input cannot execute, but notification ownership
// remains resolvable from source metadata without restoring that input.
for event in ["legacy-arbitrary-source", "mr-event:legacy-job:e"] {
conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,?2,role,assignment_id,kind,facts_json FROM notification_sources WHERE input_event_id=?1", params![source.delivery_id,event])?;
}
Ok(())
}).unwrap();
for event in [
source.delivery_id.as_str(),
"legacy-arbitrary-source",
"mr-event:legacy-job:e",
] {
assert_eq!(
store.notification_source_resource("w", event).unwrap(),
Some(PermissionResource::Job {
workspace_id: "w".into(),
job_id: "metadata-only".into()
})
);
let other = store.notification_source_resource("other", event).unwrap();
if event == "mr-event:legacy-job:e" {
assert_eq!(other, Some(PermissionResource::MergeRequest { workspace_id: "other".into(), merge_request_id: "legacy-job".into() }));
} else {
assert!(other.is_none());
}
}
assert!(
store
.notification_source_resource("w", "unknown")
.unwrap()
.is_none()
);
tracking_store_assign(&store, "metadata-assignment", "metadata-assign");
store.with_conn(|conn| {
// A collision across distinct persisted owners must not select the first.
conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,?1,role,assignment_id,kind,facts_json FROM notification_sources WHERE assignment_id='metadata-assignment' LIMIT 1", [&source.delivery_id])?;
Ok(())
}).unwrap();
assert!(
store
.notification_source_resource("w", &source.delivery_id)
.unwrap()
.is_none()
);
}
crates/workspace-server/src/server.rs
+796 -5
View File
@@ -4709,6 +4709,8 @@ impl WorkspaceApi {
resource_reused: reservation.resource_reused,
});
}
subject_authorization::authorize_job_input(self, &reservation.job.request)
.map_err(|error| error.error)?;
subjektiv_journal::prepare(self, &reservation)?;
let (reservation, claimed) = self.store.claim_backend_job_attempt_dispatch(
&self.config.workspace_id,
@@ -7434,6 +7436,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::operation_authorize(
service.clone(),
))
.merge(server_api::server_api_axum::bound_operation_authorize(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_permissions_get(service.clone()))
.merge(server_api::server_api_axum::subjektiv_subject_permissions_set(service.clone()))
.merge(server_api::server_api_axum::subjektiv_subject_create(
@@ -9484,6 +9489,139 @@ impl server_api::ServerApi for ServerApiContractService {
.map_err(ApiError::into_repository_api_error)
}
async fn bound_operation_authorize(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request: server_api::authorization::BoundOperationAuthorizationRequest,
) -> std::result::Result<
server_api::authorization::OperationAuthorizationResponse,
server_api::RepositoryApiError,
> {
let api = self.workspace_api()?.clone();
api.store
.db_executor()
.run(move || {
use server_api::authorization::{
BoundAuthorizationTarget, OperationAuthorizationRequest,
OperationAuthorizationResponse, PermissionResource,
};
validate_workspace_scope(&api, &workspace_id)?;
let scope = match request.target {
BoundAuthorizationTarget::Subject {} => {
let worker = subject_authorization::source_from_context(&context)?;
let subject_id = if let Some(subject_id) =
subject_authorization::job_subject(&api, &worker)?
{
subject_id
} else {
subjektiv_worker_scope_from_store(&api, &workspace_id, &context)?
.subject_id
};
authorize_subjektiv_operation(
&api,
&worker,
&workspace_id,
&subject_id,
&request.operation,
)?
}
BoundAuthorizationTarget::WorkdirAttachment {
alias,
expected_connection_id,
} => {
let worker = subject_authorization::source_from_context(&context)?;
let unavailable = || {
Error::WorkspacePermissionDenied(
"Workdir attachment is absent or its connection changed".into(),
)
};
if api
.store
.get_worker_registry(&workspace_id, &worker)?
.is_none()
{
return Err(unavailable().into());
}
let link = api
.store
.list_worker_workdir_links(&workspace_id, &worker)?
.into_iter()
.find(|link| {
link.unlinked_at.is_none()
&& link.alias == alias
&& expected_connection_id.as_deref().is_none_or(|expected| {
expected == link.connection_id.as_str()
})
})
.ok_or_else(unavailable)?;
let resource = PermissionResource::WorkdirAttachment {
workspace_id: workspace_id.clone(),
connection_id: link.connection_id,
};
subject_authorization::authorize_scope(
&api,
&worker,
&OperationAuthorizationRequest {
operation: request.operation,
resource: resource.clone(),
},
)?;
// Even legacy authorization returns the observed concrete
// connection so the effect can compare it again, not reuse
// this query as a credential or resolve a replacement alias.
Some(resource)
}
BoundAuthorizationTarget::NotificationSource { source_event_id } => {
let worker = subject_authorization::source_from_context(&context)?;
let unavailable = || {
Error::WorkspacePermissionDenied(
"Notification source is unavailable or not authorized".into(),
)
};
let resource = api
.store
.notification_source_resource(&workspace_id, &source_event_id)?
.ok_or_else(unavailable)?;
subject_authorization::authorize_scope(
&api,
&worker,
&OperationAuthorizationRequest {
operation: request.operation,
resource: resource.clone(),
},
)
.map_err(|_| unavailable())?;
// Return the actual target even on the ordinary non-Subject
// path; a broader matching grant is not ownership metadata.
Some(resource)
}
BoundAuthorizationTarget::SubWorker { name } => {
let worker = subject_authorization::source_from_context(&context)?;
// Names select a parent-owned capability, not a lookup of
// a child session or a grant to read that child's data.
let resource = PermissionResource::SubWorker {
runtime_id: worker.runtime_id.clone(),
worker_id: worker.worker_id.clone(),
name,
};
let allowed = subject_authorization::authorize_scope(
&api,
&worker,
&OperationAuthorizationRequest {
operation: request.operation,
resource: resource.clone(),
},
)?;
allowed.map(|_| resource)
}
};
Ok(OperationAuthorizationResponse { scope })
})
.await
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_permissions_get(
&self,
context: server_api::ServerRequestContext,
@@ -22581,6 +22719,32 @@ fn subjektiv_subject_scope(
Ok((scope.subject_id, scope.worker, authority))
}
// Resolve Subject identity through the existing Host binding before calling this
// boundary. Model arguments and Feature configuration never choose its resource.
fn authorize_subjektiv_operation(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
workspace_id: &str,
subject_id: &str,
operation: &str,
) -> ApiResult<Option<server_api::authorization::PermissionResource>> {
let resource = server_api::authorization::PermissionResource::Subject {
workspace_id: workspace_id.into(),
subject_id: subject_id.into(),
};
subject_authorization::authorize_scope(
api,
worker,
&server_api::authorization::OperationAuthorizationRequest {
operation: operation.into(),
resource: resource.clone(),
},
)
// The binding query returns only the authorized concrete target, never a
// wildcard ceiling or a broader permission that happened to cover it.
.map(|scope| scope.map(|_| resource))
}
fn require_subjektiv_worker_authority(
actual: SubjektivWorkerAuthority,
expected: SubjektivWorkerAuthority,
@@ -22997,6 +23161,13 @@ async fn scoped_stage_subjektiv_candidate(
&request.session_id,
)?;
let subject_id = attribution.subject_id.clone();
authorize_subjektiv_operation(
&api,
&RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id),
&path.workspace_id,
&subject_id,
"memory.staging.write",
)?;
let session_id = attribution.session_id.as_str();
for source_ref in &mut request.operation.source_refs {
if source_ref
@@ -23073,6 +23244,13 @@ async fn scoped_record_subjektiv_session(
&request.session_id,
)?
};
authorize_subjektiv_operation(
&api,
&RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id),
&path.workspace_id,
&attribution.subject_id,
"session.record",
)?;
subjektiv_journal::bind_host_session(&api, attribution.clone())?;
Ok(Json(server_api::SubjektivRecordSessionResponse {
subject_id: attribution.subject_id,
@@ -23184,6 +23362,17 @@ async fn scoped_subjektiv_session_backend(
SubjektivWorkerAuthority::Subject,
"Session history read",
)?;
authorize_subjektiv_operation(
&api,
&controller,
&path.workspace_id,
&subject_id,
match &request.operation {
server_api::SubjektivSessionBackendOperation::List(_) => "session.list",
server_api::SubjektivSessionBackendOperation::Search(_) => "session.search",
server_api::SubjektivSessionBackendOperation::Read(_) => "session.read",
},
)?;
let store = match open_subjektiv_store(&api) {
Ok(store) => store,
Err(error) => {
@@ -24495,6 +24684,27 @@ async fn scoped_subjektiv_memory_backend(
} else {
subjektiv_subject_scope(&api, &path.workspace_id, &context)?
};
authorize_subjektiv_operation(
&api,
&source_worker,
&path.workspace_id,
&subject_id,
match &request.operation {
Op::ResidentSummary(_) | Op::ResidentContext(_) => "memory.resident.read",
Op::ValidateProposal(_) => "memory.proposal.validate",
Op::ReceiptStatus(_) => "memory.receipt.read",
Op::StageExplicit(_) => "memory.staging.write",
Op::ListCandidates(_) => "memory.staging.list",
Op::ReadCandidate(_) => "memory.staging.read",
Op::DecideCandidate(_) => "memory.candidate.decide",
Op::PrepareSurface(_) => "memory.surface.prepare",
Op::PublishSurface(_) => "memory.surface.publish",
Op::FailSurface(_) => "memory.surface.fail",
Op::Query(_) => "memory.query",
Op::Read(_) => "memory.read",
Op::ListChanges(_) => "memory.changes.list",
},
)?;
let store = open_subjektiv_store(&api)?;
// Authentication, live Worker/Job authority and committed Session provenance
// remain Backend-owned. Only the domain operation/projection is shared.
@@ -24601,23 +24811,31 @@ async fn scoped_subjektiv_memory_consolidation(
Json(operation): Json<MemoryConsolidateStagingOperation>,
) -> ApiResult<Json<MemoryConsolidationOutput>> {
let scope_api = api.clone();
let subject_id = api
let (subject_id, source_worker) = api
.host_executors
.host()?
.run(move || {
validate_workspace_scope(&scope_api, &path.workspace_id)?;
let (subject_id, _, authority) =
let (subject_id, worker, authority) =
subjektiv_subject_scope(&scope_api, &path.workspace_id, &context)?;
require_subjektiv_worker_authority(
authority,
SubjektivWorkerAuthority::Subject,
"consolidation request",
)?;
Ok::<_, ApiError>(subject_id)
authorize_subjektiv_operation(
&scope_api,
&worker,
&path.workspace_id,
&subject_id,
"memory.consolidation.request",
)?;
Ok::<_, ApiError>((subject_id, worker))
})
.await?;
Ok(Json(
start_subjektiv_staging_consolidation(api, &subject_id, operation).await?,
start_subjektiv_staging_consolidation(api, &subject_id, Some(source_worker), operation)
.await?,
))
}
@@ -24674,14 +24892,25 @@ mod subject_surface_consolidation_policy_tests {
async fn start_subjektiv_staging_consolidation(
api: WorkspaceApi,
subject_id: &str,
source_worker: Option<RuntimeWorkerRef>,
operation: MemoryConsolidateStagingOperation,
) -> ApiResult<MemoryConsolidationOutput> {
let job_api = api.clone();
let subject = subject_id.to_owned();
let read_worker = source_worker.clone();
let (candidate_count, total_bytes) = api
.store
.db_executor()
.run(move || {
if let Some(worker) = &read_worker {
authorize_subjektiv_operation(
&job_api,
worker,
job_api.workspace_id(),
&subject,
"memory.consolidation.request",
)?;
}
open_subjektiv_store(&job_api)?
.pending_staging_backlog(&subject)
.map_err(|error| ApiError::from(subjektiv_store_error(error)))
@@ -24696,10 +24925,20 @@ async fn start_subjektiv_staging_consolidation(
}
let job_api = api.clone();
let subject = subject_id.to_owned();
let read_worker = source_worker.clone();
let surface_availability = api
.store
.db_executor()
.run(move || {
if let Some(worker) = &read_worker {
authorize_subjektiv_operation(
&job_api,
worker,
job_api.workspace_id(),
&subject,
"memory.consolidation.request",
)?;
}
open_subjektiv_store(&job_api)?
.resident_surface(&subject)
.map(|surface| surface.availability)
@@ -24735,7 +24974,18 @@ async fn start_subjektiv_staging_consolidation(
let subject = subject_id.to_owned();
api.host_executors
.host()?
.run(move || subjektiv_job::dispatch(&api, &subject, candidate_count, total_bytes))
.run(move || {
if let Some(worker) = &source_worker {
authorize_subjektiv_operation(
&api,
worker,
api.workspace_id(),
&subject,
"memory.consolidation.request",
)?;
}
subjektiv_job::dispatch(&api, &subject, source_worker, candidate_count, total_bytes)
})
.await
}
@@ -37777,6 +38027,360 @@ mod tests {
mod restore_workdir_tests;
mod restored_parent_control_tests;
mod subject_authorization_tests;
mod notification_source_authorization_tests {
// Real Feature -> signed router -> Subject policy -> SQLite metadata and
// diagnostics. Only transport scheduling and Runtime execution are fixtures.
use super::*;
use agen::tool::ToolExecutionContext;
use server_api::authorization::{
BoundAuthorizationTarget, PermissionResource, ScopedPermission,
SubjectPermissionsUpdate,
};
use worker::{
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
WorkspaceResponse,
};
struct Client {
api: WorkspaceApi,
identity: Arc<RuntimeIdentityMaterial>,
worker: RuntimeWorkerRef,
handle: tokio::runtime::Handle,
reads: std::sync::atomic::AtomicUsize,
}
impl std::fmt::Debug for Client {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("NotificationRouterClient")
}
}
impl WorkspaceClient for Client {
fn workspace_id(&self) -> Option<&str> {
Some(TEST_WORKSPACE_ID)
}
fn kind(&self) -> &str {
"signed-notification-router"
}
fn is_available(&self) -> bool {
true
}
fn execute(
&self,
request: WorkspaceRequest,
) -> std::result::Result<WorkspaceResponse, WorkspaceClientError> {
let method = match request.method {
WorkspaceRequestMethod::Get => "GET",
WorkspaceRequestMethod::Post => "POST",
_ => panic!("unexpected method"),
};
if request.path.contains("/notification-diagnostics") {
self.reads.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
let request = runtime_source_request(
&self.identity,
Some(&self.worker.worker_id),
method,
&request.path,
request.body.unwrap_or_default().into_bytes(),
);
let execute = move || {
self.handle.block_on(async {
let response = build_router(self.api.clone())
.oneshot(request)
.await
.unwrap();
let status = response.status().as_u16();
let body = to_bytes(response.into_body(), 2 * 1024 * 1024)
.await
.unwrap();
Ok(WorkspaceResponse {
status,
body: String::from_utf8(body.to_vec()).unwrap(),
})
})
};
if tokio::runtime::Handle::try_current().is_ok() {
std::thread::scope(|scope| scope.spawn(execute).join()).unwrap()
} else {
execute()
}
}
}
fn job(id: &str) -> PermissionResource {
PermissionResource::Job {
workspace_id: TEST_WORKSPACE_ID.into(),
job_id: id.into(),
}
}
fn assign(api: &WorkspaceApi, subject: &str, resources: Vec<PermissionResource>) {
let store = open_subjektiv_store(api).unwrap();
let previous = store.subject_permissions(subject).unwrap();
store
.set_subject_permissions(
subject,
&SubjectPermissionsUpdate {
operation_id: Uuid::now_v7().to_string(),
expected_change_id: previous.map(|policy| policy.change_id),
execution: resources
.into_iter()
.map(|resource| ScopedPermission {
operation: "notification.diagnostics.read".into(),
resource,
})
.collect(),
delegation: vec![],
},
)
.unwrap();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notification_source_host_binding_denies_other_jobs_and_returns_actual_scope() {
let dir = tempfile::tempdir().unwrap();
let mut api = test_api(dir.path()).await;
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
api.runtime
.register_or_replace(WorkdirlessFixtureRuntime::default());
let subject = open_subjektiv_store(&api)
.unwrap()
.create_subject(crate::subjektiv::SubjectRole::new("notification-source").unwrap())
.unwrap()
.id;
let mut workers = Vec::new();
for attached in [true, false] {
let Json(created) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(),
display_name: "Notification source reader".into(),
singleton_key: None,
profile: Some("builtin:companion".into()),
ticket_assignment: None,
initial_submit: vec![],
workdir_attachments: vec![],
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: attached.then(|| {
server_api::SubjektivWorkerConnectionRequest {
subject_id: subject.clone(),
}
}),
},
control_operation_id: None,
execution_grant: None,
}),
)
.await
.unwrap();
workers.push(RuntimeWorkerRef::new(created.runtime_id, created.worker_id));
}
let identity =
Arc::new(RuntimeIdentityMaterial::generate(&workers[0].runtime_id).unwrap());
configure_runtime_request_auth(&mut api, &identity, &workers[0].runtime_id);
let mut events = Vec::new();
for id in ["Job-A", "Job-B"] {
let request = BackendJobRequest {
job_id: id.into(),
purpose: "notification_binding".into(),
input_ref: "test://notification".into(),
input: json!({}),
instruction: "fixture".into(),
profile: "builtin:backend-job".into(),
grants: Default::default(),
serialization_key: None,
source_worker: None,
notification_target: None,
limits: Default::default(),
};
let attempt = api
.store
.reserve_backend_job(TEST_WORKSPACE_ID, &request, TEST_CREATED_AT)
.unwrap()
.attempt;
let worker =
RuntimeWorkerRef::new(&workers[0].runtime_id, format!("notification-job-{id}"));
let mut registry = api
.store
.get_worker_registry(TEST_WORKSPACE_ID, &workers[1])
.unwrap()
.unwrap();
registry.worker = worker.clone();
api.store.upsert_worker_registry(&registry).unwrap();
api.store
.claim_backend_job_attempt_dispatch(
TEST_WORKSPACE_ID,
id,
&attempt.attempt_id,
TEST_CREATED_AT,
)
.unwrap();
api.store
.bind_backend_job_attempt_worker(
TEST_WORKSPACE_ID,
id,
&attempt.attempt_id,
&worker,
Some("run"),
TEST_CREATED_AT,
)
.unwrap();
api.store
.accept_backend_job_result(
TEST_WORKSPACE_ID,
&worker,
&BackendJobResultSubmission {
job_id: id.into(),
attempt_id: attempt.attempt_id.clone(),
input_digest: request.input_digest().unwrap(),
result: json!({"protected":"not authorization metadata"}),
},
TEST_CREATED_AT,
)
.unwrap();
events.push(crate::backend_job::delivery_id(id, &attempt.attempt_id));
}
api.config_store.with_conn(|conn| {
for (event, legacy) in [(&events[0],"legacy-A"),(&events[1],"legacy-B"),(&events[1],"mr-event:legacy-B:e")] {
conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,?3,role,assignment_id,kind,facts_json FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2",rusqlite::params![TEST_WORKSPACE_ID,event,legacy])?;
}
Ok(())
}).unwrap();
let client = Arc::new(Client {
api: api.clone(),
identity: identity.clone(),
worker: workers[0].clone(),
handle: tokio::runtime::Handle::current(),
reads: Default::default(),
});
#[derive(Clone)]
struct UnusedModel;
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for UnusedModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
_: agen::llm_client::Request,
) -> std::result::Result<
agen::llm_client::ResponseStream,
agen::llm_client::ClientError,
> {
panic!("notification authorization fixture must not call a model")
}
}
let manifest = worker::WorkerManifest::from_toml("[worker]\nname='notification-reader'\n[model]\nscheme='anthropic'\nmodel_id='unused'\n[engine]\n[scope]\nallow=[]\n[feature.workdir_catalog]\nenabled=false\n[feature.ticket]\nenabled=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(dir.path().join("reader")).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(
worker::feature::builtin::ticket::TicketFeature::new(
client.clone(),
worker::feature::builtin::ticket::TicketFeatureAccess::read_only(),
),
));
assert!(!report.has_errors());
let tools = host.engine().tool_server_handle();
tools.flush_pending();
let tool = tools.get_tool("NotificationDiagnostics").unwrap().1;
let read = |event: &str, supplied: Option<&str>| {
json!({"source_event_id":event,"job_id":supplied}).to_string()
};
assert!(
tool.execute(&read(&events[0], None), ToolExecutionContext::direct())
.await
.unwrap_err()
.to_string()
.contains("403")
);
assert_eq!(client.reads.load(std::sync::atomic::Ordering::SeqCst), 0);
assign(&api, &subject, vec![job("Job-A")]);
let mut refusals = Vec::new();
for event in [&events[1], "legacy-B", "mr-event:legacy-B:e", "missing"] {
let denied = tool
.execute(&read(event, Some("Job-A")), ToolExecutionContext::direct())
.await
.unwrap_err()
.to_string();
assert!(denied.contains("403"), "{denied}");
refusals.push(denied);
}
assert!(refusals.windows(2).all(|pair| pair[0] == pair[1]));
assert_eq!(client.reads.load(std::sync::atomic::Ordering::SeqCst), 0);
for event in [&events[0], "legacy-A"] {
let output = tool
.execute(&read(event, None), ToolExecutionContext::direct())
.await
.unwrap();
let page: Value = serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
assert_eq!(page["items"][0]["job_id"], "Job-A");
assert_eq!(page["source_checks"][0]["captured"], true);
assert!(!output.content.unwrap().contains("protected"));
}
assign(&api, &subject, vec![]);
assert!(
tool.execute(&read("legacy-A", None), ToolExecutionContext::direct())
.await
.is_err()
);
assert_eq!(client.reads.load(std::sync::atomic::Ordering::SeqCst), 2);
let bound = || BoundAuthorizationTarget::NotificationSource {
source_event_id: "legacy-A".into(),
};
assign(&api, &subject, vec![job("Job-A")]);
let scope = std::thread::scope(|scope| {
scope
.spawn(|| {
client.authorize_bound_operation_scope(
"notification.diagnostics.read",
bound(),
)
})
.join()
.unwrap()
})
.unwrap();
assert_eq!(
scope,
Some(job("Job-A")),
"return the authorized exact Job target"
);
let legacy = Client {
api,
identity,
worker: workers[1].clone(),
handle: tokio::runtime::Handle::current(),
reads: Default::default(),
};
let scope = std::thread::scope(|scope| {
scope
.spawn(|| {
legacy.authorize_bound_operation_scope(
"notification.diagnostics.read",
bound(),
)
})
.join()
.unwrap()
})
.unwrap();
assert_eq!(
scope,
Some(job("Job-A")),
"non-Subject path also returns actual target"
);
}
}
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
mod subjektiv_journal_tests;
@@ -46356,6 +46960,153 @@ mod tests {
assert_eq!(restored.worker.state, "idle");
}
#[tokio::test]
async fn bound_authorization_rechecks_host_subject_and_exact_parent_without_target_lookup() {
use server_api::authorization::{
PermissionResource, ScopedPermission, SubjectPermissionsUpdate,
};
// Real signed router, Host binding and permission storage; only Runtime
// execution is substituted. No child or protected-data fixture is created.
let workspace = tempfile::tempdir().unwrap();
let mut api = test_api(workspace.path()).await;
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
api.runtime
.register_or_replace(WorkdirlessFixtureRuntime::default());
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject_with_behavior(
crate::subjektiv::SubjectRole::new("bound-authorization").unwrap(),
"Protected behavior must not appear in authorization responses.".into(),
)
.unwrap();
let Json(created) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(),
display_name: "Bound authorization source".into(),
singleton_key: None,
profile: Some("builtin:companion".into()),
ticket_assignment: None,
initial_submit: vec![],
workdir_attachments: vec![],
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
subject_id: subject.id.clone(),
}),
},
control_operation_id: None,
execution_grant: None,
}),
)
.await
.unwrap();
let identity = RuntimeIdentityMaterial::generate(&created.runtime_id).unwrap();
configure_runtime_request_auth(&mut api, &identity, &created.runtime_id);
let path = format!("/api/w/{TEST_WORKSPACE_ID}/authorization/bound-check");
let subject_request =
json!({"operation":"memory.resident.supply", "target":{"kind":"subject"}});
let check = |body: Value| {
runtime_source_request(
&identity,
Some(&created.worker_id),
"POST",
&path,
serde_json::to_vec(&body).unwrap(),
)
};
let denied = build_router(api.clone())
.oneshot(check(subject_request.clone()))
.await
.unwrap();
assert_eq!(denied.status(), StatusCode::FORBIDDEN);
let subject_resource = PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.id.clone(),
};
let child_resource = PermissionResource::SubWorker {
runtime_id: created.runtime_id.clone(),
worker_id: created.worker_id.clone(),
name: "not-started-child".into(),
};
let policy = store
.set_subject_permissions(
&subject.id,
&SubjectPermissionsUpdate {
operation_id: Uuid::now_v7().to_string(),
expected_change_id: None,
execution: vec![
ScopedPermission {
operation: "memory.resident.supply".into(),
resource: subject_resource.clone(),
},
ScopedPermission {
operation: "worker.session.read".into(),
resource: child_resource.clone(),
},
],
delegation: vec![],
},
)
.unwrap();
for (body, expected) in [
(subject_request.clone(), subject_resource),
(
json!({"operation":"worker.session.read", "target":{"kind":"sub_worker", "name":"not-started-child"}}),
child_resource,
),
] {
let response = build_router(api.clone())
.oneshot(check(body))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let bytes = to_bytes(response.into_body(), 16 * 1024).await.unwrap();
let value: Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(value, json!({"scope": expected}));
}
// A different child name cannot inherit this grant. Caller-provided Subject
// identities are rejected at the wire boundary, not used as selectors.
for body in [
json!({"operation":"worker.session.read", "target":{"kind":"sub_worker", "name":"other-child"}}),
json!({"operation":"memory.resident.supply", "target":{"kind":"subject", "subject_id":"other-subject"}}),
] {
let response = build_router(api.clone())
.oneshot(check(body))
.await
.unwrap();
assert!(response.status().is_client_error());
}
store
.set_subject_permissions(
&subject.id,
&SubjectPermissionsUpdate {
operation_id: Uuid::now_v7().to_string(),
expected_change_id: Some(policy.change_id),
execution: vec![],
delegation: vec![],
},
)
.unwrap();
let revoked = build_router(api.clone())
.oneshot(check(subject_request))
.await
.unwrap();
assert_eq!(revoked.status(), StatusCode::FORBIDDEN);
let unsigned = Request::builder()
.method("POST")
.uri(&path)
.header(CONTENT_TYPE, "application/json")
.body(Body::from(
json!({"operation":"memory.resident.supply", "target":{"kind":"subject"}})
.to_string(),
))
.unwrap();
let response = build_router(api).oneshot(unsigned).await.unwrap();
assert!(response.status().is_client_error());
}
#[tokio::test]
async fn subjektiv_subject_worker_and_staging_derive_scope_from_singleton_authority() {
let workspace = tempfile::tempdir().unwrap();
@@ -46379,6 +47130,46 @@ mod tests {
"Prefer explicit evidence.".to_string(),
)
.unwrap();
use server_api::authorization::{
PermissionResource, ScopedPermission, SubjectPermissionsUpdate,
};
let resource = PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.id.clone(),
};
let mut execution: Vec<_> = [
"memory.resident.read",
"memory.resident.supply",
"memory.staging.write",
"memory.receipt.read",
"session.record",
"session.list",
"session.search",
]
.into_iter()
.map(|operation| ScopedPermission {
operation: operation.into(),
resource: resource.clone(),
})
.collect();
execution.push(ScopedPermission {
operation: "memory.resident.read".into(),
resource: PermissionResource::Workspace {
workspace_id: TEST_WORKSPACE_ID.into(),
},
});
open_subjektiv_store(&api)
.unwrap()
.set_subject_permissions(
&subject.id,
&SubjectPermissionsUpdate {
operation_id: "staging-scope-fixture".into(),
expected_change_id: None,
execution,
delegation: vec![],
},
)
.unwrap();
let request = CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Subject Worker".to_string(),
crates/workspace-server/src/server/subject_authorization.rs
+202 -7
View File
@@ -143,12 +143,157 @@ fn live_child_grant(
))
}
/// Job domain capabilities attenuate this policy; they are never non-Subject
/// compatibility. A missing frozen origin cannot mint new operation permissions.
fn job_permissions(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> ApiResult<Option<Vec<ScopedPermission>>> {
let Some(binding) = api
.store
.worker_registry_projection(api.workspace_id(), worker)?
.and_then(|projection| projection.job)
else {
return Ok(None);
};
let job = api
.store
.get_backend_job(api.workspace_id(), &binding.job_id)?
.ok_or_else(|| denied("Job authorization binding has no intent"))?;
let subject_id = job
.request
.grants
.subjektiv_consolidation
.as_ref()
.map(|grant| &grant.subject_id)
.or_else(|| {
job.request
.grants
.subjektiv_journal
.as_ref()
.map(|grant| &grant.subject_id)
});
let Some(subject_id) = subject_id else {
// Generic Jobs have no Subject operation capability, even if their Profile
// exposes a protected Feature or their source owns a Subject body.
return Ok(Some(Vec::new()));
};
// This shared check validates current attempt, worker, input digest and deadline
// for both Job kinds, independently of its optional consolidation payload.
subjektiv_job::active_grant(api, api.workspace_id(), worker)?;
let origin = job
.request
.source_worker
.as_ref()
.ok_or_else(|| denied("saved Job has no frozen operation-authority origin"))?;
if current_body_subject(api, origin)?.as_deref() != Some(subject_id.as_str()) {
return Err(denied("Job origin is no longer the current Subject body"));
}
let permissions = policy(api, subject_id)?
.map(|policy| policy.execution)
.unwrap_or_default();
Ok(Some(permissions.into_iter().filter(|permission| {
// Domain grants are immutable capabilities, not permission-policy snapshots.
// Current policy can narrow them, but cannot turn a Journal Job into a
// consolidator or confer other Subject operations on either Job kind.
let within_job = if job.request.grants.subjektiv_consolidation.is_some() {
matches!(permission.operation.as_str(),
"memory.consolidation.supply" | "memory.staging.list" | "memory.staging.read"
| "memory.read" | "memory.query" | "memory.resident.read" | "memory.resident.supply"
| "memory.candidate.decide" | "memory.proposal.validate" | "memory.changes.list"
| "memory.receipt.read" | "memory.surface.prepare" | "memory.surface.supply"
| "memory.surface.publish" | "memory.surface.fail")
} else {
matches!(permission.operation.as_str(),
"journal.job.input.read" | "journal.job.input.supply" | "journal.state.read"
| "journal.text.read" | "journal.operation.query" | "journal.operation.list"
| "journal.operation.coverage.read" | "journal.entry.create" | "journal.entry.fix"
| "journal.entry.rewrite" | "journal.position.update" | "journal.unchanged.confirm"
| "journal.checkpoint" | "journal.finish")
};
within_job && matches!(&permission.resource, PermissionResource::Subject {workspace_id, subject_id: granted}
if workspace_id == api.workspace_id() && granted == subject_id)
}).collect()))
}
/// Resolve only Host-held Job identity; authorization still validates the live
/// attempt, frozen origin and current policy before returning the resource.
pub(super) fn job_subject(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> ApiResult<Option<String>> {
let Some(binding) = api
.store
.worker_registry_projection(api.workspace_id(), worker)?
.and_then(|projection| projection.job)
else {
return Ok(None);
};
let job = api
.store
.get_backend_job(api.workspace_id(), &binding.job_id)?
.ok_or_else(|| denied("Job authorization binding has no intent"))?;
Ok(job
.request
.grants
.subjektiv_consolidation
.map(|grant| grant.subject_id)
.or_else(|| {
job.request
.grants
.subjektiv_journal
.map(|grant| grant.subject_id)
}))
}
pub(super) fn authorize_job_input(
api: &WorkspaceApi,
request: &BackendJobRequest,
) -> ApiResult<()> {
let binding = request
.grants
.subjektiv_consolidation
.as_ref()
.map(|grant| (&grant.subject_id, "memory.consolidation.supply"))
.or_else(|| {
request
.grants
.subjektiv_journal
.as_ref()
.map(|grant| (&grant.subject_id, "journal.job.input.supply"))
});
let Some((subject_id, operation)) = binding else {
return Ok(());
};
let origin = request
.source_worker
.as_ref()
.ok_or_else(|| denied("saved Job has no frozen operation-authority origin"))?;
if current_body_subject(api, origin)?.as_deref() != Some(subject_id.as_str()) {
return Err(denied("Job origin is no longer the current Subject body"));
}
authorize(
api,
origin,
&OperationAuthorizationRequest {
operation: operation.into(),
resource: PermissionResource::Subject {
workspace_id: api.workspace_id().into(),
subject_id: subject_id.clone(),
},
},
)
}
/// None is explicitly the preexisting non-Subject contract, not an all-powerful
/// grant. Existing Workdir/provider/Ticket checks still apply to those Workers.
pub(super) fn worker_permissions(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> ApiResult<Option<Vec<ScopedPermission>>> {
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));
}
@@ -170,14 +315,57 @@ pub(super) fn authorize(
authorize_scope(api, worker, request).map(|_| ())
}
fn resources_cover(
api: &WorkspaceApi,
granted: &PermissionResource,
requested: &PermissionResource,
) -> ApiResult<bool> {
if granted.covers(requested) {
return Ok(true);
}
let (workspace_id, granted_id, requested_id, kind) = match (granted, requested) {
(
PermissionResource::Ticket {
workspace_id: a,
ticket_id: ai,
},
PermissionResource::Ticket {
workspace_id: b,
ticket_id: bi,
},
) if a == b => (a, ai, bi, WorkspaceResourceKind::Ticket),
(
PermissionResource::Objective {
workspace_id: a,
objective_id: ai,
},
PermissionResource::Objective {
workspace_id: b,
objective_id: bi,
},
) if a == b => (a, ai, bi, WorkspaceResourceKind::Objective),
_ => return Ok(false),
};
// Resolve public aliases using identity metadata only. Never fetch Ticket,
// Objective or MR bodies to answer an authorization query. Missing aliases
// and unrelated resources have the same refusal.
let granted_id = api
.store
.resolve_resource_reference(workspace_id, kind, granted_id)?;
let requested_id = api
.store
.resolve_resource_reference(workspace_id, kind, requested_id)?;
Ok(granted_id.is_some() && granted_id == requested_id)
}
pub(super) fn authorize_scope(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
request: &OperationAuthorizationRequest,
) -> ApiResult<Option<PermissionResource>> {
request.validate().map_err(denied)?;
if let PermissionResource::Workspace { workspace_id } = &request.resource {
if workspace_id != &api.config.workspace_id {
if let Some(workspace_id) = request.resource.workspace_id() {
if workspace_id != api.config.workspace_id {
return Err(denied("resource belongs to another Workspace"));
}
}
@@ -188,11 +376,18 @@ pub(super) fn authorize_scope(
let Some(permissions) = worker_permissions(api, worker)? else {
return Ok(None);
};
if let Some(permission) = permissions
.iter()
.find(|permission| permission.covers(&needed))
{
return Ok(Some(permission.resource.clone()));
for permission in &permissions {
if permission.operation == needed.operation
&& resources_cover(api, &permission.resource, &needed.resource)?
{
return Ok(Some(if permission.resource.covers(&needed.resource) {
permission.resource.clone()
} else {
// Alias resolution proved exact identity; return the spelling the
// caller requested rather than a scope it cannot lexically match.
needed.resource.clone()
}));
}
}
Err(denied(format!(
"operation `{}` is not granted for the requested resource",
crates/workspace-server/src/server/subjektiv_job.rs
+2 -1
View File
@@ -179,6 +179,7 @@ fn bounded_candidate_batch(
pub(super) fn dispatch(
api: &WorkspaceApi,
subject_id: &str,
source_worker: Option<RuntimeWorkerRef>,
candidate_count: usize,
total_bytes: u64,
) -> ApiResult<MemoryConsolidationOutput> {
@@ -206,7 +207,7 @@ pub(super) fn dispatch(
profile: format!("builtin:{SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE}"),
grants: BackendJobGrants { subjektiv_consolidation: Some(SubjektivConsolidationGrant { subject_id: subject_id.into(), candidate_ids }), ..Default::default() },
serialization_key: None,
source_worker: None,
source_worker,
notification_target: None,
limits: crate::backend_job::BackendJobLimits { timeout_seconds: 600, ..Default::default() },
};
crates/workspace-server/src/server/subjektiv_journal.rs
+83 -1
View File
@@ -91,6 +91,15 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) -
return Ok(());
};
let store = store(api)?;
let attribution = store
.session_attribution(&grant.session_id)
.map_err(|e| Error::Store(e.to_string()))?
.filter(|attr| attr.subject_id == grant.subject_id)
.ok_or_else(|| {
Error::WorkspacePermissionDenied("Journal source attribution missing".into())
})?;
authorize_source(api, &attribution, "journal.job.input.read")?;
authorize_source(api, &attribution, "journal.job.input.supply")?;
let host = host(api, &store, grant)?;
let input = saved_input(&reservation.job.request, grant)?;
let g = store
@@ -238,6 +247,49 @@ fn generation(
Ok(g)
}
fn authorize(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
subject: &str,
operation: &str,
) -> Result<()> {
let scope = authorize_subjektiv_operation(api, worker, api.workspace_id(), subject, operation)
.map_err(|error| error.error)?;
if scope.is_none() {
return Err(Error::WorkspacePermissionDenied(
"Journal requires current Subject operation authority".into(),
));
}
Ok(())
}
fn authorize_source(
api: &WorkspaceApi,
attribution: &crate::subjektiv::SubjectSessionAttribution,
operation: &str,
) -> Result<()> {
let worker = RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id);
authorize(api, &worker, &attribution.subject_id, operation)
}
fn job_operation(request: &JournalJobRequest) -> &'static str {
match request {
JournalJobRequest::Execute { operation } => match &operation.action {
JournalAction::Create { .. } => "journal.entry.create",
JournalAction::Rewrite { .. } => "journal.entry.rewrite",
JournalAction::Fix { .. } => "journal.entry.fix",
JournalAction::UpdatePosition { .. } => "journal.position.update",
JournalAction::ConfirmUnchanged => "journal.unchanged.confirm",
JournalAction::Checkpoint { .. } => "journal.checkpoint",
},
JournalJobRequest::ReadState => "journal.state.read",
JournalJobRequest::QueryOperation { .. } => "journal.operation.query",
JournalJobRequest::ReadOperationCoverage { .. } => "journal.operation.coverage.read",
JournalJobRequest::ListOperations { .. } => "journal.operation.list",
JournalJobRequest::ReadText { .. } => "journal.text.read",
}
}
pub(super) fn execute(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
@@ -251,6 +303,7 @@ pub(super) fn execute(
// Dispatch holds this same gate until matching Runtime acceptance is attested.
// Revalidate after waiting: erasure, cancellation or replacement may have won.
let (grant, job, attempt) = live(api, worker)?;
authorize(api, worker, &grant.subject_id, job_operation(&request))?;
let store = store(api)?;
let host = host(api, &store, &grant)?;
let g = generation(&store, &host, &grant, &job, &attempt)?;
@@ -296,6 +349,7 @@ pub(super) fn finish(
// Dispatch holds this same gate until matching Runtime acceptance is attested.
// Revalidate after waiting: erasure, cancellation or replacement may have won.
let (grant, job, attempt) = live(api, worker)?;
authorize(api, worker, &grant.subject_id, "journal.finish")?;
let store = store(api)?;
let host = host(api, &store, &grant)?;
let input = saved_input(&job.request, &grant)?;
@@ -583,6 +637,7 @@ fn dispatch(
let _gate = gate
.lock()
.map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
authorize_source(api, &attribution, "journal.recording.request")?;
let store = store(api)?;
let host =
JournalHostContext::validated_session(&store, attribution.clone()).map_err(domain)?;
@@ -662,6 +717,7 @@ fn dispatch(
if input.selected_ranges.is_empty() {
return Ok(());
}
authorize_source(api, &attribution, "journal.recording.request")?;
store
.accept_journal_sources(&host, "Session activity", &input.selected_ranges)
.map_err(domain)?;
@@ -699,6 +755,7 @@ fn admit_sources(
attribution: crate::subjektiv::SubjectSessionAttribution,
dispatch_jobs: bool,
) -> Result<JournalRecordingResponse> {
authorize_source(api, &attribution, "journal.recording.request")?;
let store = store(api)?;
let host =
JournalHostContext::validated_session(&store, attribution.clone()).map_err(domain)?;
@@ -715,6 +772,7 @@ fn admit_sources(
(0, 0)
};
let capture = capture_input(api, &attribution, order, offset, 48 * 1024)?;
authorize_source(api, &attribution, "journal.recording.request")?;
retain_source(
api.runtime.as_ref(),
api.workspace_id(),
@@ -724,6 +782,7 @@ fn admit_sources(
)?;
let sources = journal::recording::capture_ranges(&capture).map_err(domain)?;
for batch in sources.chunks(MAX_JOURNAL_RANGES) {
authorize_source(api, &attribution, "journal.recording.request")?;
id = Some(
store
.accept_journal_sources(&host, "Session activity", batch)
@@ -768,6 +827,13 @@ pub(super) async fn startup_route(
)
.into());
}
authorize_subjektiv_operation(
&api,
&worker,
&path.workspace_id,
subject,
"journal.startup.read",
)?;
let store = store(&api)?;
let reader = JournalHostContext::validated_reader(&store, subject).map_err(domain)?;
let mut current_writer = None;
@@ -834,6 +900,13 @@ pub(super) async fn compaction_route(
)
.into());
}
authorize_subjektiv_operation(
&api,
&worker,
&path.workspace_id,
subject,
"journal.compaction.read",
)?;
let store = store(&api)?;
let attribution = store
.session_attribution(&input.session_id)
@@ -1263,6 +1336,13 @@ pub(super) async fn recording_route(
)
.into());
}
authorize_subjektiv_operation(
&api,
&worker,
&path.workspace_id,
subject,
"journal.recording.request",
)?;
require_enabled(&api, &worker)?;
let store = store(&api)?;
let attributions = store
@@ -1527,6 +1607,7 @@ pub(super) fn recording_request(
Error::WorkspacePermissionDenied("Journal Session has no Host writer binding".into())
})?
.to_string();
let source_worker = RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id);
let grant = SubjektivJournalGrant {
binding_id,
subject_id: attribution.subject_id,
@@ -1541,7 +1622,7 @@ pub(super) fn recording_request(
instruction: "Record ONLY the immutable body-bearing capture supplied in this input. These exact canonical source bytes, identities and ranges are the whole input for this Job; do not fetch original Sessions or use ReadJournalText Source. First ReadJournalState; initial/current Journal position/open text remain separately paged by ReadJournalText. Sequentially save CreateJournalEntry/RewriteJournalEntry/FixJournalEntry, current position and receipt-backed CheckpointJournal. Input delivery is not understanding or reflection. Never infer event completion from turn/task/job/compaction boundaries. Preserve exact operation ID/full input after Unknown. SubmitBackendJobResult only after the complete selected prefix has a saved checkpoint; no aggregate JournalRecordResult.".into(),
profile: "builtin:subjektiv-journal-recording".into(),
grants: BackendJobGrants { subjektiv_journal: Some(grant), ..Default::default() },
serialization_key: Some(journal_serialization_key(journal_id)), source_worker: None, notification_target: None,
serialization_key: Some(journal_serialization_key(journal_id)), source_worker: Some(source_worker), notification_target: None,
limits: crate::backend_job::BackendJobLimits { timeout_seconds: 600, ..Default::default() },
};
saved_input(
@@ -1563,6 +1644,7 @@ pub(super) fn capture_input(
offset: u64,
budget: usize,
) -> Result<session_store::capture::CaptureDto> {
authorize_source(api, attr, "journal.recording.request")?;
match api
.runtime
.session_recording_capture(
crates/workspace-server/src/server/subjektiv_memory_journal.rs
+16
View File
@@ -72,6 +72,22 @@ pub(super) async fn route(
SubjektivWorkerAuthority::Subject,
"Memory Journal access",
)?;
let operation = match &request {
MemoryJournalRequest::Reconcile { .. }
| MemoryJournalRequest::ReconcilePage { .. } => "journal.memory.links.reconcile",
MemoryJournalRequest::Links { .. } | MemoryJournalRequest::LinksPage { .. } => {
"journal.memory.links.read"
}
MemoryJournalRequest::Read { .. } => "journal.memory.read",
MemoryJournalRequest::Related { .. } => "journal.memory.related.read",
};
authorize_subjektiv_operation(
&api,
&worker,
&path.workspace_id,
subject_id,
operation,
)?;
let store = open_subjektiv_store(&api)?;
let host = JournalHostContext::validated_reader(&store, subject_id).map_err(domain)?;
store
crates/workspace-server/src/server/tests/subject_authorization_tests.rs
+998 -54
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/server/tests/subject_spawn_tests.rs
+24
View File
@@ -18,6 +18,30 @@ async fn subject_callbacks_finish_before_waiting_host_parents_release_capacity()
crate::subjektiv::SubjectRole::new(format!("companion-{index}")).unwrap(),
)
.unwrap();
subjektiv
.set_subject_permissions(
&subject.id,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: format!("subject-callback-grant-{index}"),
expected_change_id: None,
execution: [
"memory.resident.read",
"session.record",
"journal.startup.read",
]
.into_iter()
.map(|operation| server_api::authorization::ScopedPermission {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.id.clone(),
},
})
.collect(),
delegation: vec![],
},
)
.unwrap();
let singleton_key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap();
let reserved = api
.config_store
crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs
+352 -1
View File
@@ -7,6 +7,81 @@ use memory::extract::{CandidateKind, ExtractedCandidate, StagingEvidence, Stagin
use memory::schema::SourceRef;
use memory::schema::{EvidenceKind, EvidenceOrigin, EvidenceOriginKind};
const MEMORY_JOB_OPERATIONS: &[&str] = &[
"memory.consolidation.request",
"memory.consolidation.supply",
"memory.staging.list",
"memory.staging.read",
"memory.candidate.decide",
"memory.query",
"memory.read",
"memory.changes.list",
"memory.surface.prepare",
"memory.surface.supply",
"memory.surface.publish",
"memory.surface.fail",
];
fn grant_memory_operations(api: &WorkspaceApi, subject: &str, operations: &[&str]) {
use server_api::authorization::{
PermissionResource, ScopedPermission, SubjectPermissionsUpdate,
};
let store = open_subjektiv_store(api).unwrap();
let previous = store.subject_permissions(subject).unwrap();
store
.set_subject_permissions(
subject,
&SubjectPermissionsUpdate {
operation_id: Uuid::now_v7().to_string(),
expected_change_id: previous.map(|policy| policy.change_id),
execution: operations
.iter()
.map(|operation| ScopedPermission {
operation: (*operation).into(),
resource: PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.into(),
},
})
.collect(),
delegation: vec![],
},
)
.unwrap();
}
async fn memory_job_source(api: &WorkspaceApi, subject: &str) -> RuntimeWorkerRef {
grant_memory_operations(api, subject, MEMORY_JOB_OPERATIONS);
attach_memory_job_source(api, subject).await
}
async fn attach_memory_job_source(api: &WorkspaceApi, subject: &str) -> RuntimeWorkerRef {
seed_test_delivery_binding(api);
let Json(created) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
display_name: "Memory Job authority source".into(),
singleton_key: None,
profile: Some("builtin:companion".into()),
ticket_assignment: None,
initial_submit: vec![],
workdir_attachments: vec![],
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
subject_id: subject.into(),
}),
},
control_operation_id: None,
execution_grant: None,
}),
)
.await
.unwrap();
RuntimeWorkerRef::new(created.runtime_id, created.worker_id)
}
fn stage(store: &crate::subjektiv::SubjektivStore, subject: &str, id: &str) {
let origin = EvidenceOrigin {
kind: EvidenceOriginKind::HumanInput,
@@ -105,10 +180,12 @@ async fn subjektiv_job_bounds_batch_auth_partial_receipts_surface_and_cleanup()
let subject = store
.create_subject(SubjectRole::new("test-subject").unwrap())
.unwrap();
let source = memory_job_source(&api, &subject.id).await;
stage(&store, &subject.id, "candidate-first");
let started = start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
@@ -153,6 +230,7 @@ async fn subjektiv_job_bounds_batch_auth_partial_receipts_surface_and_cleanup()
let again = start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
@@ -261,6 +339,7 @@ async fn subjektiv_job_bounds_batch_auth_partial_receipts_surface_and_cleanup()
let next = start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
@@ -285,11 +364,13 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o
let subject = store
.create_subject(SubjectRole::new("surface-only").unwrap())
.unwrap();
let source = memory_job_source(&api, &subject.id).await;
let old = store.prepare_surface_generation(&subject.id).unwrap();
assert_eq!(
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: false }
)
.await
@@ -375,6 +456,7 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true }
)
.await
@@ -502,7 +584,8 @@ async fn subjektiv_job_legacy_cutover_waits_foreground_then_removes_only_owned_c
let subject = store
.create_subject(SubjectRole::new("legacy-cutover").unwrap())
.unwrap();
let subject_body = spawn_ticket_check_source(&api, "subject-body-protected");
let source = memory_job_source(&api, &subject.id).await;
let subject_body = source.clone();
let key = format!("subjektiv-consolidation:{}", subject.id);
let legacy = api
.spawn_workspace_worker(
@@ -557,6 +640,7 @@ async fn subjektiv_job_legacy_cutover_waits_foreground_then_removes_only_owned_c
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true }
)
.await
@@ -590,6 +674,7 @@ async fn subjektiv_job_legacy_cutover_waits_foreground_then_removes_only_owned_c
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true }
)
.await
@@ -616,6 +701,7 @@ async fn subjektiv_job_long_receipts_and_escaped_ids_fit_result_budget_without_l
let subject = store
.create_subject(SubjectRole::new("result-budget").unwrap())
.unwrap();
let source = memory_job_source(&api, &subject.id).await;
for i in 0..100 {
stage(
&store,
@@ -626,6 +712,7 @@ async fn subjektiv_job_long_receipts_and_escaped_ids_fit_result_budget_without_l
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
@@ -706,6 +793,7 @@ async fn subjektiv_job_long_receipts_and_escaped_ids_fit_result_budget_without_l
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
@@ -730,3 +818,266 @@ async fn subjektiv_job_long_receipts_and_escaped_ids_fit_result_budget_without_l
> 16 * 1024
);
}
#[tokio::test]
async fn subjektiv_job_rechecks_current_policy_and_preserves_provider_batch_ceiling() {
let dir = tempfile::tempdir().unwrap();
let (api, execution) = test_api_with_recording_backend(dir.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(SubjectRole::new("live-job-policy").unwrap())
.unwrap();
let source = memory_job_source(&api, &subject.id).await;
let mut operations = MEMORY_JOB_OPERATIONS.to_vec();
operations.extend(["journal.state.read", "session.record"]);
grant_memory_operations(&api, &subject.id, &operations);
stage(&store, &subject.id, "frozen-candidate");
start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source.clone()),
MemoryConsolidateStagingOperation { force: true },
)
.await
.unwrap();
let job = resource(&api, &subject.id);
assert_eq!(job.request.source_worker.as_ref(), Some(&source));
let attempt = current_attempt(&api, &job);
let worker = attempt.worker.as_ref().unwrap();
assert_eq!(execution.take_inputs().len(), 1);
for operation in ["journal.state.read", "session.record"] {
let request = server_api::authorization::OperationAuthorizationRequest {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.id.clone(),
},
};
assert!(subject_authorization::authorize_scope(&api, &source, &request).is_ok());
assert!(
subject_authorization::authorize_scope(&api, worker, &request).is_err(),
"current body policy cannot widen a consolidation Job's domain: {operation}"
);
}
let prepare = || {
server_api::SubjektivMemoryBackendOperation::PrepareSurface(
server_api::SubjektivSurfacePrepareRequest {},
)
};
// Late arrivals never enter this attempt's frozen batch, and the pending
// granted candidate still prevents premature surface preparation.
stage(&store, &subject.id, "late-candidate");
assert!(
subjektiv_job::require_candidate_grant(&api, TEST_WORKSPACE_ID, worker, "frozen-candidate")
.is_ok()
);
assert!(
subjektiv_job::require_candidate_grant(&api, TEST_WORKSPACE_ID, worker, "late-candidate")
.is_err()
);
let read_candidate = |id: &str| {
server_api::SubjektivMemoryBackendOperation::ReadCandidate(
server_api::SubjektivMemoryCandidateReadRequest {
candidate_id: id.into(),
},
)
};
assert!(
op(&api, worker, read_candidate("frozen-candidate"))
.await
.is_ok()
);
assert!(
op(&api, worker, read_candidate("late-candidate"))
.await
.is_err()
);
grant_memory_operations(&api, &subject.id, &["memory.surface.prepare"]);
let error = op(&api, worker, read_candidate("frozen-candidate"))
.await
.unwrap_err();
assert!(
error.error.to_string().contains("memory.staging.read"),
"{error:?}"
);
let permissions = subject_authorization::worker_permissions(&api, worker)
.unwrap()
.unwrap();
assert_eq!(permissions.len(), 1);
assert_eq!(permissions[0].operation, "memory.surface.prepare");
// Provider batch checks remain active even when operation policy allows it.
assert!(op(&api, worker, prepare()).await.is_err());
grant_memory_operations(&api, &subject.id, &[]);
let error = op(&api, worker, prepare()).await.unwrap_err();
assert!(
error.error.to_string().contains("memory.surface.prepare"),
"{error:?}"
);
assert!(
subject_authorization::worker_permissions(&api, worker)
.unwrap()
.unwrap()
.is_empty()
);
assert_eq!(resource(&api, &subject.id).request, job.request);
assert!(execution.take_inputs().is_empty());
}
#[tokio::test]
async fn saved_job_without_frozen_source_cannot_use_non_subject_fallback() {
let dir = tempfile::tempdir().unwrap();
let (api, _) = test_api_with_recording_backend(dir.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(SubjectRole::new("legacy-job-origin").unwrap())
.unwrap();
let _source = memory_job_source(&api, &subject.id).await;
// Reconstruct a retained pre-migration dispatched attempt through the ledger,
// not today's dispatcher: new dispatch itself must reject a missing origin.
let request = BackendJobRequest {
job_id: "legacy-no-operation-origin".into(),
purpose: "subjektiv_consolidation".into(),
input_ref: format!("subjektiv://{}/consolidation", subject.id),
input: serde_json::json!({"subject_id":subject.id,"candidate_ids":[]}),
instruction: "retained legacy input".into(),
profile: "builtin:subjektiv-memory-consolidation".into(),
grants: crate::backend_job::BackendJobGrants {
subjektiv_consolidation: Some(crate::backend_job::SubjektivConsolidationGrant {
subject_id: subject.id.clone(),
candidate_ids: vec![],
}),
..Default::default()
},
serialization_key: None,
source_worker: None,
notification_target: None,
limits: Default::default(),
};
let error = subject_authorization::authorize_job_input(&api, &request).unwrap_err();
assert!(error.error.to_string().contains("frozen"), "{error:?}");
let now = Utc::now().to_rfc3339();
let reserved = api
.store
.reserve_backend_job(TEST_WORKSPACE_ID, &request, &now)
.unwrap();
let (_, claimed) = api
.store
.claim_backend_job_attempt_dispatch(
TEST_WORKSPACE_ID,
&request.job_id,
&reserved.attempt.attempt_id,
&now,
)
.unwrap();
assert!(claimed);
let retained_worker = spawn_ticket_check_source(&api, "retained-legacy-job");
api.store
.bind_backend_job_attempt_worker(
TEST_WORKSPACE_ID,
&request.job_id,
&reserved.attempt.attempt_id,
&retained_worker,
None,
&now,
)
.unwrap();
let job = resource(&api, &subject.id);
let attempt = current_attempt(&api, &job);
let worker = attempt.worker.as_ref().unwrap();
assert!(subjektiv_job::active_grant(&api, TEST_WORKSPACE_ID, worker).is_ok());
let error = op(
&api,
worker,
server_api::SubjektivMemoryBackendOperation::PrepareSurface(
server_api::SubjektivSurfacePrepareRequest {},
),
)
.await
.unwrap_err();
assert!(error.error.to_string().contains("frozen"), "{error:?}");
assert!(subject_authorization::worker_permissions(&api, worker).is_err());
assert_eq!(resource(&api, &subject.id).request, job.request);
}
#[tokio::test]
async fn unset_subject_consolidation_denies_before_job_dispatch() {
let dir = tempfile::tempdir().unwrap();
let (api, execution) = test_api_with_recording_backend(dir.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(SubjectRole::new("unset-job-policy").unwrap())
.unwrap();
let source = attach_memory_job_source(&api, &subject.id).await;
assert!(store.subject_permissions(&subject.id).unwrap().is_none());
stage(&store, &subject.id, "untouched");
let error = start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source),
MemoryConsolidateStagingOperation { force: true },
)
.await
.unwrap_err();
assert!(
error
.error
.to_string()
.contains("memory.consolidation.request"),
"{error:?}"
);
assert!(execution.take_inputs().is_empty());
assert!(
api.store
.find_active_backend_job_for_resource(
TEST_WORKSPACE_ID,
&crate::backend_job::subjektiv_serialization_key(&subject.id)
)
.unwrap()
.is_none()
);
assert_eq!(store.pending_staging_backlog(&subject.id).unwrap().0, 1);
}
#[tokio::test]
async fn subject_consolidation_request_without_supply_denies_before_model_input() {
let dir = tempfile::tempdir().unwrap();
let (api, execution) = test_api_with_recording_backend(dir.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(SubjectRole::new("job-supply-denied").unwrap())
.unwrap();
let source = memory_job_source(&api, &subject.id).await;
stage(&store, &subject.id, "not-supplied");
grant_memory_operations(&api, &subject.id, &["memory.consolidation.request"]);
let error = start_subjektiv_staging_consolidation(
api.clone(),
&subject.id,
Some(source),
MemoryConsolidateStagingOperation { force: true },
)
.await
.unwrap_err();
assert!(
error
.error
.to_string()
.contains("memory.consolidation.supply"),
"{error:?}"
);
assert!(
execution.take_inputs().is_empty(),
"denied frozen input must not reach the model/runtime"
);
let saved = resource(&api, &subject.id);
let attempt = current_attempt(&api, &saved);
assert_eq!(attempt.state, BackendJobAttemptState::Reserved);
assert!(
attempt.worker.is_none(),
"supply refusal must precede claim/allocation"
);
assert_eq!(
saved.request.input["candidate_ids"],
serde_json::json!(["not-supplied"])
);
assert_eq!(store.pending_staging_backlog(&subject.id).unwrap().0, 1);
}
crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs
+200 -41
View File
@@ -1,6 +1,8 @@
//! Connected Backend → production Runtime factory → real Controller → Journal → signed
//! Host startup reader → persisted first input. Only the external model stream is scripted.
//! Browser authentication is outside this regression: creation enters the real scoped handler.
//! Policy regressions reuse the parent's canonical source/SQLite RuntimeFixture with
//! recording Runtime execution; those cases do not claim real Controller/model execution.
use super::*;
use agen::llm_client::{
LlmClient, Request,
@@ -13,6 +15,40 @@ use std::sync::{Arc, Mutex};
use super::super::restored_parent_control_tests::connected_restore_tempdir;
fn connected_subject_permissions(
subject: &str,
) -> Vec<server_api::authorization::ScopedPermission> {
JOURNAL_OPERATIONS
.iter()
.copied()
.chain([
"memory.extract.read",
"memory.extract.supply",
"memory.resident.read",
"memory.resident.supply",
"memory.consolidation.request",
"memory.consolidation.supply",
"memory.surface.prepare",
"memory.surface.supply",
"memory.surface.publish",
"memory.surface.fail",
"memory.staging.list",
"memory.staging.read",
"memory.candidate.decide",
"memory.query",
"memory.read",
"memory.changes.list",
])
.map(|operation| server_api::authorization::ScopedPermission {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.into(),
},
})
.collect()
}
#[derive(Clone, Default)]
struct ConnectedJournalModel {
dialogs: Arc<Mutex<Vec<Request>>>,
@@ -399,31 +435,12 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
let owner =
seed_test_workspace_owner_token(api.store.as_ref(), "connected-journal-owner").await;
// This fixture exercises Journal continuity, not unset-policy refusal. Assign
// the companion's exact activation keys through authenticated administration;
// the companion's required Subject operations through authenticated administration;
// no filesystem, shell or delegation permission is granted to its body.
let update = server_api::authorization::SubjectPermissionsUpdate {
operation_id: "connected-journal-activation".into(),
expected_change_id: None,
execution: [
"task",
"subjektiv",
"web",
"image",
"sub_worker",
"worker",
"workspace_worker_discovery",
"objective",
"manage_workdir",
"workdir_catalog",
"drive",
"ticket",
]
.into_iter()
.map(|name| server_api::authorization::ScopedPermission {
operation: name.into(),
resource: server_api::authorization::PermissionResource::Feature,
})
.collect(),
execution: connected_subject_permissions(&subject.id),
delegation: Vec::new(),
};
let response = build_router(api.clone())
@@ -1606,26 +1623,7 @@ async fn connected_restore_fixture_with_pending_head(
let update = server_api::authorization::SubjectPermissionsUpdate {
operation_id: "restore-host-auth-permissions".into(),
expected_change_id: None,
execution: [
"task",
"subjektiv",
"web",
"image",
"sub_worker",
"worker",
"workspace_worker_discovery",
"objective",
"manage_workdir",
"workdir_catalog",
"drive",
"ticket",
]
.into_iter()
.map(|name| server_api::authorization::ScopedPermission {
operation: name.into(),
resource: server_api::authorization::PermissionResource::Feature,
})
.collect(),
execution: connected_subject_permissions(&subject.id),
delegation: vec![],
};
let response = build_router(api.clone())
@@ -2554,3 +2552,164 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) {
}
shutdown_workspace_feature_storage(api).await.unwrap();
}
fn replace_recording_fixture_policy(
f: &RuntimeFixture,
execution: Vec<server_api::authorization::ScopedPermission>,
) {
let subject = &f
.request
.grants
.subjektiv_journal
.as_ref()
.unwrap()
.subject_id;
let previous = f.store.subject_permissions(subject).unwrap().unwrap();
f.store
.set_subject_permissions(
subject,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: Uuid::now_v7().to_string(),
expected_change_id: Some(previous.change_id),
execution,
delegation: vec![],
},
)
.unwrap();
}
#[tokio::test]
async fn journal_supply_denial_precedes_generation_claim_and_dispatch_then_exact_grant_allows_processing()
{
let f = RuntimeFixture::new().await;
let grant = f.request.grants.subjektiv_journal.as_ref().unwrap();
let execution = connected_subject_permissions(&grant.subject_id);
assert!(
execution
.iter()
.any(|permission| permission.operation == "journal.job.input.supply")
);
replace_recording_fixture_policy(
&f,
execution
.iter()
.filter(|permission| permission.operation != "journal.job.input.supply")
.cloned()
.collect(),
);
assert!(f.execution.take_inputs().is_empty());
let error = f
.api
.dispatch_backend_job_reservation(f.reserved.clone())
.unwrap_err();
assert!(
error.to_string().contains("journal.job.input.supply"),
"{error:?}"
);
assert!(
f.execution.take_inputs().is_empty(),
"denied input must not reach Runtime/model dispatch"
);
assert!(
f.store
.active_journal_generation(&f.host, &f.id)
.unwrap()
.is_none()
);
let attempt = f
.api
.store
.get_backend_job_attempt(
TEST_WORKSPACE_ID,
&f.request.job_id,
&f.reserved.attempt.attempt_id,
)
.unwrap()
.unwrap();
assert_eq!(attempt.state, BackendJobAttemptState::Reserved);
assert!(attempt.worker.is_none());
let saved = f
.api
.store
.get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id)
.unwrap()
.unwrap();
assert_eq!(
saved.request, f.request,
"refusal preserves immutable capture and selection"
);
replace_recording_fixture_policy(&f, execution);
let dispatched = f
.api
.dispatch_backend_job_reservation(f.reserved.clone())
.unwrap();
let worker = dispatched.attempt.worker.as_ref().unwrap();
assert_eq!(f.execution.take_inputs().len(), 1);
let JournalJobResponse::State(state) =
subjektiv_journal::execute(&f.api, worker, JournalJobRequest::ReadState).unwrap()
else {
panic!("actual Journal state required")
};
assert_eq!(state.selected_ranges, grant.selected_ranges);
assert_eq!(dispatched.job.request, f.request);
// The body really holds Memory and session permissions. A Journal capability
// cannot inherit those merely because their Subject resource is identical.
for operation in [
"memory.read",
"memory.consolidation.supply",
"session.record",
] {
let needed = server_api::authorization::OperationAuthorizationRequest {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: grant.subject_id.clone(),
},
};
assert!(subject_authorization::authorize_scope(&f.api, &f.source, &needed).is_ok());
assert!(
subject_authorization::authorize_scope(&f.api, worker, &needed).is_err(),
"Journal domain ceiling must reject {operation}"
);
}
// Exercise the signed bound Subject API for the Job, not only the internal
// permission helper. Metadata resolution must not require body singleton ownership.
let mut api = f.api.clone();
let identity = RuntimeIdentityMaterial::generate(&worker.runtime_id).unwrap();
configure_runtime_request_auth(&mut api, &identity, &worker.runtime_id);
for (operation, expected) in [
("journal.job.input.supply", StatusCode::OK),
("memory.read", StatusCode::FORBIDDEN),
] {
let response = build_router(api.clone())
.oneshot(runtime_source_request(
&identity,
Some(&worker.worker_id),
"POST",
&format!("/api/w/{TEST_WORKSPACE_ID}/authorization/bound-check"),
serde_json::to_vec(&json!({"operation":operation,"target":{"kind":"subject"}}))
.unwrap(),
))
.await
.unwrap();
let status = response.status();
let body: Value = serde_json::from_slice(
&to_bytes(response.into_body(), 2 * 1024 * 1024)
.await
.unwrap(),
)
.unwrap();
assert_eq!(status, expected, "{operation}: {body}");
if status == StatusCode::OK {
assert_eq!(
body,
json!({"scope":{"kind":"subject","workspace_id":TEST_WORKSPACE_ID,"subject_id":grant.subject_id}})
);
} else {
assert!(body.get("scope").is_none());
}
}
}
crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs
+164 -20
View File
@@ -11,10 +11,100 @@ use subjektiv::journal::*;
#[path = "subjektiv_journal_connected_tests.rs"]
mod connected;
const JOURNAL_OPERATIONS: &[&str] = &[
"session.record",
"journal.startup.read",
"journal.startup.supply",
"journal.compaction.read",
"journal.compaction.supply",
"journal.recording.request",
"journal.job.input.read",
"journal.job.input.supply",
"journal.entry.create",
"journal.entry.rewrite",
"journal.entry.fix",
"journal.position.update",
"journal.unchanged.confirm",
"journal.checkpoint",
"journal.state.read",
"journal.operation.query",
"journal.operation.coverage.read",
"journal.operation.list",
"journal.text.read",
"journal.finish",
];
fn grant_journal_operations(store: &crate::subjektiv::SubjektivStore, subject: &str) {
use server_api::authorization::{
PermissionResource, ScopedPermission, SubjectPermissionsUpdate,
};
let previous = store.subject_permissions(subject).unwrap();
store
.set_subject_permissions(
subject,
&SubjectPermissionsUpdate {
operation_id: format!("journal-fixture-grant-{subject}"),
expected_change_id: previous.map(|policy| policy.change_id),
execution: JOURNAL_OPERATIONS
.iter()
.map(|operation| ScopedPermission {
operation: (*operation).into(),
resource: PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.into(),
},
})
.collect(),
delegation: vec![],
},
)
.unwrap();
}
fn spawn_journal_source(api: &WorkspaceApi, subject: &str, name: &str) -> RuntimeWorkerRef {
seed_test_delivery_binding(api);
api.spawn_workspace_worker(
EMBEDDED_WORKER_RUNTIME_ID,
WorkerSpawnRequest {
requested_worker_name: Some(name.into()),
intent: WorkerSpawnIntent::WorkspaceCompanion,
singleton_key: Some(crate::subjektiv::subject_worker_singleton_key(subject).unwrap()),
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
},
profile: ProfileSelector::Builtin("builtin:companion".into()),
ticket_assignment: None,
initial_submit: vec![],
workdir_attachment_requests: vec![],
resolved_workdir_attachment_requests: vec![],
resolved_workdir_attachments: vec![],
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: vec![],
resolved_workspace_api: None,
resolved_memory_settings: None,
resolved_subjektiv_attached: false,
resolved_control_operation: None,
resolved_execution_grant: None,
},
)
.unwrap()
.worker
.unwrap()
.worker
}
fn bind_fixture_session(
store: &crate::subjektiv::SubjektivStore,
attr: crate::subjektiv::SubjectSessionAttribution,
) -> JournalHostContext {
if store
.subject_permissions(&attr.subject_id)
.unwrap()
.is_none()
{
grant_journal_operations(store, &attr.subject_id);
}
let binding = store
.claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
.unwrap();
@@ -245,7 +335,7 @@ impl Fixture {
let subject = store
.create_subject(crate::subjektiv::SubjectRole::new("journal-test").unwrap())
.unwrap();
let source = spawn_ticket_check_source(&api, "journal-source");
let source = spawn_journal_source(&api, &subject.id, "journal-source");
let attr = store
.record_session_attribution(
crate::subjektiv::SubjectSessionAttribution::new(
@@ -308,7 +398,7 @@ impl Fixture {
..Default::default()
},
serialization_key: Some(journal_serialization_key(&id)),
source_worker: None,
source_worker: Some(source.clone()),
notification_target: None,
limits: Default::default(),
};
@@ -609,13 +699,13 @@ impl RuntimeFixture {
async fn with_history(history: Vec<session_store::LoggedHistoryEntry>) -> Self {
let dir = tempfile::tempdir().unwrap();
let (api, execution) = test_api_with_recording_backend(dir.path()).await;
let source = spawn_ticket_check_source(&api, "original-source");
let session = session_store::new_session_id();
seed_session_history(&session_root(dir.path(), &source), session, history);
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(crate::subjektiv::SubjectRole::new("source-reader-test").unwrap())
.unwrap();
let source = spawn_journal_source(&api, &subject.id, "original-source");
let session = session_store::new_session_id();
seed_session_history(&session_root(dir.path(), &source), session, history);
let attr = store
.record_session_attribution(
crate::subjektiv::SubjectSessionAttribution::new(
@@ -1303,7 +1393,7 @@ async fn journal_reserved_restart_dispatches_exact_intent_once_and_unbound_unkno
}
#[tokio::test]
async fn archived_original_session_remains_readable_then_releases_its_pin_after_saved_completion() {
async fn subject_source_removal_is_fenced_and_pin_releases_after_saved_completion() {
let f = RuntimeFixture::new().await;
let dispatched = f.dispatch_input();
let worker = dispatched.attempt.worker.as_ref().unwrap();
@@ -1326,18 +1416,21 @@ async fn archived_original_session_remains_readable_then_releases_its_pin_after_
.unwrap();
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
StatusCode::CONFLICT.as_u16(),
"{}",
response.body
);
let archive = f
.api
.config_store
.worker_session_archive(TEST_WORKSPACE_ID, &f.source, &f.session.to_string())
.unwrap()
.expect("durable archive binding");
assert!(!session_root(f.dir.path(), &f.source).exists());
// Archive movement neither changes nor reclaims the saved immutable body.
assert!(response.body.contains("journal_handoff_incomplete"));
assert!(session_root(f.dir.path(), &f.source).exists());
assert!(
f.api
.config_store
.worker_session_archive(TEST_WORKSPACE_ID, &f.source, &f.session.to_string())
.unwrap()
.is_none()
);
// Stopping the body is not release. Keep its source while the active
// writer/Job still depends on current ownership and saved completion.
// The Job cannot fall back to a retained or archived Session read.
let saved = f
.api
@@ -1396,19 +1489,70 @@ async fn archived_original_session_remains_readable_then_releases_its_pin_after_
BackendJobWorkerCleanupState::Completed,
)
.await;
assert!(
!retention.pins(f.session).unwrap().is_empty(),
"saved reflection alone does not release the still-owned source"
);
// This deterministic fixture has no Controller to prove a live disconnect.
// Seed an explicitly completed historical release, as in the archive-recovery
// component tests; connected release proof is exercised separately.
let lease = f
.api
.store
.require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &f.source)
.unwrap()
.unwrap();
let binding_id = f.host.writer_binding_id().unwrap();
let operation = "completed-source-release-before-archive";
f.api
.config_store
.reserve_subject_release(TEST_WORKSPACE_ID, &lease, binding_id, operation)
.unwrap();
f.store
.release_journal_writer(&JournalWriterBinding {
subject_id: f
.request
.grants
.subjektiv_journal
.as_ref()
.unwrap()
.subject_id
.clone(),
runtime_id: f.source.runtime_id.clone(),
worker_id: f.source.worker_id.clone(),
binding_id: binding_id.into(),
})
.unwrap();
f.api
.config_store
.confirm_subject_release_proof(TEST_WORKSPACE_ID, &lease, binding_id, operation, false)
.unwrap();
f.api
.config_store
.release_subject_singleton(TEST_WORKSPACE_ID, &lease, binding_id, operation)
.unwrap();
let response = WorkerRemovalService::new(&f.api)
.execute_target_removal_on_host(f.api.runtime.as_ref(), &f.source, "archive original", None)
.await
.unwrap();
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
"{}",
response.body
);
assert!(
retention.pins(f.session).unwrap().is_empty(),
"archive pin is released only after saved reflection"
"completed reflection and explicit release allow archive and source pin release"
);
assert!(!session_root(f.dir.path(), worker).exists());
assert_eq!(
assert!(
f.api
.config_store
.worker_session_archive(TEST_WORKSPACE_ID, &f.source, &f.session.to_string())
.unwrap()
.unwrap()
.archive_id,
archive.archive_id
.is_some(),
"the explicitly released source is archived after saved reflection"
);
}
crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs
+146 -23
View File
@@ -39,7 +39,39 @@ struct Fixture {
}
impl Fixture {
fn grant_permissions(store: &crate::subjektiv::SubjektivStore, subject_id: &str) {
store
.set_subject_permissions(
subject_id,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: format!("memory-journal-fixture-grant-{subject_id}"),
expected_change_id: None,
execution: [
"journal.memory.links.reconcile",
"journal.memory.links.read",
"journal.memory.read",
"journal.memory.related.read",
]
.into_iter()
.map(|operation| server_api::authorization::ScopedPermission {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject_id.into(),
},
})
.collect(),
delegation: vec![],
},
)
.unwrap();
}
async fn new() -> Self {
Self::with_permissions(true).await
}
async fn with_permissions(granted: bool) -> Self {
use session_store::{LogEntry, Store as _, WorkerSessionStore};
let dir = tempfile::tempdir().unwrap();
let mut api = test_api(dir.path()).await;
@@ -49,6 +81,9 @@ impl Fixture {
let subject = store
.create_subject(crate::subjektiv::SubjectRole::new("memory-journal").unwrap())
.unwrap();
if granted {
Self::grant_permissions(&store, &subject.id);
}
let worker = Self::body_worker(&api, &subject.id);
let source_worker = RuntimeWorkerRef::new(RUNTIME, Uuid::new_v4().to_string());
seed_worker_source_member(&api, RUNTIME, &source_worker.worker_id);
@@ -648,29 +683,38 @@ async fn memory_journal_router_ids_do_not_grant_cross_subject_reads_or_reconcili
.create_subject(crate::subjektiv::SubjectRole::new("other").unwrap())
.unwrap();
let other_worker = Fixture::body_worker(&f.api, &other.id);
for request in [
MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
},
MemoryJournalRequest::Links {
selector: f.selector.clone(),
},
f.read_request(),
MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
after: None,
limit: 32,
},
] {
let (status, bytes) = f.call_as(&other_worker, request).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"{}",
String::from_utf8_lossy(&bytes)
);
assert!(!String::from_utf8_lossy(&bytes).contains(JOURNAL_BODY));
for granted in [false, true] {
if granted {
Fixture::grant_permissions(&f.store, &other.id);
}
for request in [
MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
},
MemoryJournalRequest::Links {
selector: f.selector.clone(),
},
f.read_request(),
MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
after: None,
limit: 32,
},
] {
let (status, bytes) = f.call_as(&other_worker, request).await;
assert_eq!(
status,
if granted {
StatusCode::BAD_REQUEST
} else {
StatusCode::FORBIDDEN
},
"{}",
String::from_utf8_lossy(&bytes)
);
assert!(!String::from_utf8_lossy(&bytes).contains(JOURNAL_BODY));
}
}
// A model cannot override the authenticated Subject with an additional JSON field.
let mut body = serde_json::to_value(f.read_request()).unwrap();
@@ -733,3 +777,82 @@ async fn memory_journal_router_correction_reverse_lookup_preserves_memory_and_or
#[path = "subjektiv_journal_browser_tests.rs"]
mod browser_tests;
#[tokio::test]
async fn memory_journal_unset_or_revoked_permissions_deny_every_route_without_link_writes() {
for revoked in [false, true] {
let f = Fixture::with_permissions(revoked).await;
let lease = f
.api
.store
.require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &f.worker)
.unwrap()
.unwrap();
let (subject, _) = subjektiv_scope_from_singleton_key(&lease.key).unwrap();
if revoked {
let policy = f.store.subject_permissions(&subject).unwrap().unwrap();
f.store
.set_subject_permissions(
&subject,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: "revoke-memory-journal".into(),
expected_change_id: Some(policy.change_id),
execution: vec![],
delegation: vec![],
},
)
.unwrap();
}
let count_links = || {
f.api
.feature_storage
.open(&f.api.subjektiv_registration)
.unwrap()
.with_connection(|conn| {
Ok(
conn.query_row("SELECT COUNT(*) FROM memory_journal_links", [], |row| {
row.get::<_, i64>(0)
})?,
)
})
.unwrap()
};
let before = count_links();
let requests = [
MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
},
MemoryJournalRequest::Links {
selector: f.selector.clone(),
},
MemoryJournalRequest::ReconcilePage {
selector: f.selector.clone(),
source_offset: 0,
limit: 1,
},
MemoryJournalRequest::LinksPage {
selector: f.selector.clone(),
source_offset: 0,
limit: 1,
},
f.read_request(),
MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
after: None,
limit: 1,
},
];
for request in requests {
let (status, body) = f.call_as(&f.worker, request).await;
assert_eq!(
status,
StatusCode::FORBIDDEN,
"{}",
String::from_utf8_lossy(&body)
);
assert!(!String::from_utf8_lossy(&body).contains(JOURNAL_BODY));
assert_eq!(count_links(), before);
}
}
}
crates/workspace-server/src/server/tests/workspace_admission_tests.rs
+17
View File
@@ -821,6 +821,23 @@ async fn create_and_restore_session_callbacks_finish_while_deletion_start_waits(
let subject = subjektiv
.create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
.unwrap();
subjektiv
.set_subject_permissions(
&subject.id,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: "session-callback-grant".into(),
expected_change_id: None,
execution: vec![server_api::authorization::ScopedPermission {
operation: "session.record".into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: TEST_WORKSPACE_ID.into(),
subject_id: subject.id.clone(),
},
}],
delegation: vec![],
},
)
.unwrap();
let (entered, mut entries) = mpsc::unbounded_channel();
let (release, released) = std::sync::mpsc::channel();
let released = Arc::new(Mutex::new(released));
crates/workspace-server/src/server_db_execution_tests.rs
+24
View File
@@ -1124,6 +1124,30 @@ async fn subjektiv_runtime_waits_leave_db_and_control_free_and_cancel_retains_th
.await
.unwrap()
.1;
if let Some(operation) = match kind {
"read" => Some("session.read"),
"search" => Some("session.search"),
_ => None,
} {
open_subjektiv_store(&fixture.api)
.unwrap()
.set_subject_permissions(
&subject.id,
&server_api::authorization::SubjectPermissionsUpdate {
operation_id: format!("session-wait-grant-{kind}"),
expected_change_id: None,
execution: vec![server_api::authorization::ScopedPermission {
operation: operation.into(),
resource: server_api::authorization::PermissionResource::Subject {
workspace_id: WORKSPACE.into(),
subject_id: subject.id.clone(),
},
}],
delegation: vec![],
},
)
.unwrap();
}
let singleton = if kind == "legacy_scope" {
format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{}", subject.id)
} else {
crates/workspace-server/src/store.rs
+28
View File
@@ -2197,6 +2197,13 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
delivery_id: &str,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<()>;
/// Resolve only persisted source ownership metadata, never notification content.
/// Missing and ambiguous identities both return None.
fn notification_source_resource(
&self,
workspace_id: &str,
source_event_id: &str,
) -> Result<Option<server_api::authorization::PermissionResource>>;
fn notification_diagnostics(
&self,
workspace_id: &str,
@@ -4696,6 +4703,20 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
));
}
self.with_conn(|conn| {
if kind == WorkspaceResourceKind::Ticket {
// Identity-only projection with the same ambiguity rule as the
// Ticket provider; authorization must not load protected bodies.
let mut statement = conn.prepare(
"SELECT ticket_id FROM typed_tickets
WHERE workspace_id = ?1 AND (ticket_id = ?2 OR slug = ?2)
UNION SELECT resource_id FROM workspace_resource_keys
WHERE workspace_id = ?1 AND resource_kind = 'ticket' AND resource_key = ?2
LIMIT 2",
)?;
let ids = statement.query_map(params![workspace_id, reference], |row| row.get::<_, String>(0))?
.collect::<std::result::Result<Vec<_>, _>>()?;
return Ok(match ids.as_slice() { [id] => Some(id.clone()), _ => None });
}
if let Some(resource_id) = conn
.query_row(
"SELECT resource_id FROM workspace_resource_keys
@@ -9136,6 +9157,13 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
None,
)
}
fn notification_source_resource(
&self,
workspace_id: &str,
source_event_id: &str,
) -> Result<Option<server_api::authorization::PermissionResource>> {
notification_diagnostics::source_resource(self, workspace_id, source_event_id)
}
fn notification_diagnostics(
&self,
workspace_id: &str,