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

feat: integrate operation-boundary authorization

Browse Source 
Merge reviewed source a35d9a3d45 into develop after Subject deletion. Preserve deletion fencing before Job authorization, retain both test modules, and regenerate combined API artifacts.
This commit is contained in:
Hare
2026-10-28 13:38:47 +09:00
parent 0710738173 a35d9a3d45
commit f598a36ac0
78 changed files with 14305 additions and 1145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/worker_backend.rs
+187 -37
View File
@@ -7103,63 +7103,175 @@ mod tests {
const SAVED_SUBJEKTIV_SYSTEM_PROMPT: &str = "Committed saved Subject Worker system prompt";
// Substituted Host transport for restore/controller component tests. This
// fixture has exactly the two activation grants requested by the saved
// Manifest; it grants no protected operations or resources. It does not
// exercise authenticated source derivation, Subject policy, or revocation
// in the real Workspace Host.
// Substituted Host transport for restore/controller component tests. Only
// resident Memory read/supply, empty startup Journal read and Session record
// are granted, on the saved Subject alone. No Feature activation or unrelated
// operations are granted. This does not exercise authenticated source
// derivation, Subject policy or revocation in the real Workspace Host.
fn saved_subjektiv_authorization_fixture(
input: server_api::authorization::OperationAuthorizationRequest,
) -> Result<server_api::authorization::OperationAuthorizationResponse, &'static str> {
use server_api::authorization::{OperationAuthorizationResponse, PermissionResource};
if input.validate().is_ok()
&& matches!(input.operation.as_str(), "subjektiv" | "workdir_catalog")
&& input.resource == PermissionResource::Feature
&& matches!(
input.operation.as_str(),
"memory.resident.read"
| "memory.resident.supply"
| "journal.startup.read"
| "session.record"
)
&& input.resource
== (PermissionResource::Subject {
workspace_id: "workspace-saved-subjektiv".into(),
subject_id: "saved-subject".into(),
})
{
Ok(OperationAuthorizationResponse {
scope: Some(PermissionResource::Feature),
scope: Some(input.resource),
})
} else {
Err("saved Subject fixture has no matching exact Feature grant")
Err("saved Subject fixture has no matching exact operation/resource grant")
}
}
fn saved_subjektiv_bound_authorization_fixture(
input: server_api::authorization::BoundOperationAuthorizationRequest,
) -> Result<server_api::authorization::OperationAuthorizationResponse, &'static str> {
use server_api::authorization::{
BoundAuthorizationTarget, OperationAuthorizationRequest, PermissionResource,
};
if input.target != (BoundAuthorizationTarget::Subject {}) {
return Err("saved Subject fixture has no matching bound target");
}
saved_subjektiv_authorization_fixture(OperationAuthorizationRequest {
operation: input.operation,
resource: PermissionResource::Subject {
workspace_id: "workspace-saved-subjektiv".into(),
subject_id: "saved-subject".into(),
},
})
}
#[test]
fn saved_subjektiv_authorization_fixture_grants_only_exact_feature_activation() {
use server_api::authorization::{OperationAuthorizationRequest, PermissionResource};
for operation in ["subjektiv", "workdir_catalog"] {
fn saved_subjektiv_authorization_fixture_grants_only_exact_restore_operations() {
use server_api::authorization::{
BoundAuthorizationTarget, BoundOperationAuthorizationRequest,
OperationAuthorizationRequest, PermissionResource,
};
let subject = PermissionResource::Subject {
workspace_id: "workspace-saved-subjektiv".into(),
subject_id: "saved-subject".into(),
};
for operation in [
"memory.resident.read",
"memory.resident.supply",
"journal.startup.read",
"session.record",
] {
let response = saved_subjektiv_authorization_fixture(OperationAuthorizationRequest {
operation: operation.into(),
resource: PermissionResource::Feature,
resource: subject.clone(),
})
.unwrap();
assert_eq!(response.scope, Some(PermissionResource::Feature));
}
for (operation, resource) in [
("ticket", PermissionResource::Feature),
("subjektiv.journal", PermissionResource::Feature),
("subjektiv.*", PermissionResource::Feature),
(
"subjektiv",
PermissionResource::Workspace {
workspace_id: "workspace-saved-subjektiv".into(),
},
),
(
"subjektiv.journal",
PermissionResource::Workspace {
workspace_id: "workspace-saved-subjektiv".into(),
},
),
] {
assert!(
saved_subjektiv_authorization_fixture(OperationAuthorizationRequest {
assert_eq!(response.scope, Some(subject.clone()));
let bound =
saved_subjektiv_bound_authorization_fixture(BoundOperationAuthorizationRequest {
operation: operation.into(),
resource,
target: BoundAuthorizationTarget::Subject {},
})
.unwrap();
assert_eq!(bound.scope, Some(subject.clone()));
for resource in [
PermissionResource::Feature,
PermissionResource::Workspace {
workspace_id: "workspace-saved-subjektiv".into(),
},
PermissionResource::Subject {
workspace_id: "other-workspace".into(),
subject_id: "saved-subject".into(),
},
PermissionResource::Subject {
workspace_id: "workspace-saved-subjektiv".into(),
subject_id: "other-subject".into(),
},
] {
assert!(
saved_subjektiv_authorization_fixture(OperationAuthorizationRequest {
operation: operation.into(),
resource,
})
.is_err(),
"out-of-scope grant for {operation}"
);
}
for target in [
BoundAuthorizationTarget::WorkdirAttachment {
alias: "checkout".into(),
expected_connection_id: None,
},
BoundAuthorizationTarget::SubWorker {
name: "child".into(),
},
BoundAuthorizationTarget::NotificationSource {
source_event_id: "event".into(),
},
] {
assert!(
saved_subjektiv_bound_authorization_fixture(
BoundOperationAuthorizationRequest {
operation: operation.into(),
target,
}
)
.is_err()
);
}
}
for operation in [
"",
"ticket",
"subjektiv",
"workdir_catalog",
"subjektiv.journal",
"subjektiv.*",
"memory.resident",
"memory.resident.*",
"memory.resident.write",
"memory.staging.write",
"memory.consolidation.request",
"journal.startup.supply",
"journal.compaction.read",
] {
for resource in [PermissionResource::Feature, subject.clone()] {
assert!(
saved_subjektiv_authorization_fixture(OperationAuthorizationRequest {
operation: operation.into(),
resource,
})
.is_err(),
"unexpected grant for {operation}"
);
}
assert!(
saved_subjektiv_bound_authorization_fixture(BoundOperationAuthorizationRequest {
operation: operation.into(),
target: BoundAuthorizationTarget::Subject {},
})
.is_err()
);
}
assert!(
serde_json::from_value::<OperationAuthorizationRequest>(serde_json::json!({
"resource": subject,
}))
.is_err()
);
assert!(
serde_json::from_value::<BoundOperationAuthorizationRequest>(serde_json::json!({
"target": {"kind": "subject"},
}))
.is_err()
);
}
fn saved_subjektiv_policy(
@@ -7253,14 +7365,14 @@ mod tests {
worker.materialize_durable_session_head().await.unwrap();
let entries = session_store.read_all(session_id, segment_id).unwrap();
assert!(
matches!(entries.as_slice(), [LogEntry::AnnotatedSegmentStart {
matches!(entries.first(), Some(LogEntry::AnnotatedSegmentStart {
session_id: committed_session_id,
system_prompt: Some(prompt),
history,
forked_from: None,
compacted_from: None,
..
}] if *committed_session_id == session_id
}) if *committed_session_id == session_id
&& prompt == SAVED_SUBJEKTIV_SYSTEM_PROMPT
&& history.is_empty())
);
@@ -7331,6 +7443,43 @@ mod tests {
}))
}),
);
let app = app.route(
"/api/w/workspace-saved-subjektiv/subjektiv/memory",
axum::routing::post(
|axum::Json(input): axum::Json<server_api::SubjektivMemoryBackendRequest>| async move {
assert!(matches!(
input.operation,
server_api::SubjektivMemoryBackendOperation::ResidentContext(_)
));
axum::Json(serde_json::json!({
"result": "resident_context",
"data": {
"behavior_md": "Saved Subject restore behavior",
"memory_surface": {"availability": "ungenerated", "content": null},
},
}))
},
),
);
let app = app.route(
"/api/w/workspace-saved-subjektiv/authorization/bound-check",
axum::routing::post(
|axum::Json(input): axum::Json<
server_api::authorization::BoundOperationAuthorizationRequest,
>| async move {
match saved_subjektiv_bound_authorization_fixture(input) {
Ok(response) => (
axum::http::StatusCode::OK,
axum::Json(serde_json::to_value(response).unwrap()),
),
Err(message) => (
axum::http::StatusCode::FORBIDDEN,
axum::Json(serde_json::json!({ "message": message })),
),
}
},
),
);
let app = app.route(
"/api/w/workspace-saved-subjektiv/authorization/check",
axum::routing::post(
@@ -8817,6 +8966,7 @@ mod tests {
..
}) if *session_id == pending_session_id
&& prompt.contains("OPERATION-OWNED-PENDING-SYSTEM-PROMPT")
&& prompt.contains("Saved Subject restore behavior")
&& history.is_empty())
);
assert!(
crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs
+95 -20
View File
@@ -232,17 +232,16 @@ mod restore_candidate {
calls.lock().unwrap().push((
"POST".into(), authorization_path.into(),
));
let response = if input.operation == "task"
&& input.resource == server_api::authorization::PermissionResource::Feature
{
server_api::authorization::OperationAuthorizationResponse {
scope: Some(server_api::authorization::PermissionResource::Feature),
}
} else {
saved_subjektiv_authorization_fixture(input)
.expect("only the saved fixture Feature grants are available")
};
axum::Json(response)
match saved_subjektiv_authorization_fixture(input) {
Ok(response) => (
axum::http::StatusCode::OK,
axum::Json(serde_json::to_value(response).unwrap()),
),
Err(message) => (
axum::http::StatusCode::FORBIDDEN,
axum::Json(serde_json::json!({ "message": message })),
),
}
}
}),
)
@@ -1410,8 +1409,8 @@ mod restore_candidate {
.unwrap();
assert_eq!(fixture.model.call_count.load(Ordering::SeqCst), 4);
assert!(
!fixture.host.calls.lock().unwrap().is_empty(),
"activation must reach the configured Host transport"
fixture.host.calls.lock().unwrap().is_empty(),
"local TaskCreate, FIFO drain and receipt replay must not call the Host"
);
fixture.assert_no_modifying_host_calls();
assert!(
@@ -1471,12 +1470,45 @@ mod restore_candidate {
let record_calls = calls.clone();
let startup_calls = Arc::new(AtomicUsize::new(0));
let startups = startup_calls.clone();
let memory_calls = Arc::new(AtomicUsize::new(0));
let memories = memory_calls.clone();
let authorization_operations = Arc::new(Mutex::new(Vec::new()));
let operations = authorization_operations.clone();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
request.request.workspace_api.as_mut().unwrap().base_url = format!("http://{}", listener.local_addr().unwrap());
let app = axum::Router::new()
.route("/api/w/workspace-saved-subjektiv/authorization/check", axum::routing::post(
|axum::Json(input): axum::Json<server_api::authorization::OperationAuthorizationRequest>| async move {
axum::Json(saved_subjektiv_authorization_fixture(input).unwrap())
.route("/api/w/workspace-saved-subjektiv/authorization/bound-check", axum::routing::post(
move |axum::Json(input): axum::Json<server_api::authorization::BoundOperationAuthorizationRequest>| {
let operations = operations.clone();
async move {
operations.lock().unwrap().push(input.operation.clone());
match saved_subjektiv_bound_authorization_fixture(input) {
Ok(response) => (
axum::http::StatusCode::OK,
axum::Json(serde_json::to_value(response).unwrap()),
),
Err(message) => (
axum::http::StatusCode::FORBIDDEN,
axum::Json(serde_json::json!({ "message": message })),
),
}
}
}))
.route("/api/w/workspace-saved-subjektiv/subjektiv/memory", axum::routing::post(
move |axum::Json(input): axum::Json<server_api::SubjektivMemoryBackendRequest>| {
let memories = memories.clone();
async move {
assert!(matches!(input.operation,
server_api::SubjektivMemoryBackendOperation::ResidentContext(_)));
memories.fetch_add(1, Ordering::SeqCst);
axum::Json(serde_json::json!({
"result": "resident_context",
"data": {
"behavior_md": "Repaired Subject resident behavior",
"memory_surface": { "availability": "ungenerated", "content": null },
},
}))
}
}))
.route("/api/w/workspace-saved-subjektiv/subjektiv/sessions", axum::routing::post(
move |axum::Json(input): axum::Json<server_api::SubjektivRecordSessionRequest>| {
@@ -1512,26 +1544,69 @@ mod restore_candidate {
assert!(matches!(failed.subjektiv_session_attribution,
Some(session_store::SubjektivSessionAttributionState::OutcomeUnknown { session_id: id, .. }) if id == session_id));
assert_eq!(startup_calls.load(Ordering::SeqCst), 0);
assert_eq!(memory_calls.load(Ordering::SeqCst), 0);
assert!(authorization_operations.lock().unwrap().is_empty());
factory.cleanup_failed_restore(&request).await.unwrap();
assert!(factory.recover_saved_restore_source(&request).await.unwrap());
let repaired = metadata.read_by_name(&name).unwrap().unwrap();
assert_eq!(repaired.active.as_ref().unwrap().session_id, session_id);
assert!(matches!(repaired.subjektiv_session_attribution,
Some(session_store::SubjektivSessionAttributionState::Confirmed { session_id: id, .. }) if id == session_id));
let segments = WorkerSessionStore::new(aggregate.join("session")).unwrap()
.list_segments(session_id).unwrap();
let session = WorkerSessionStore::new(aggregate.join("session")).unwrap();
let segments = session.list_segments(session_id).unwrap();
assert_eq!(segments.len(), 1, "retry must retain exactly one seed");
let entries = session.read_all(session_id,
repaired.active.as_ref().unwrap().segment_id.unwrap()).unwrap();
assert_eq!(entries.len(), 2, "one head plus its protected-input checkpoint");
let prompt = match &entries[0] {
LogEntry::AnnotatedSegmentStart {
session_id: id, system_prompt: Some(prompt), history,
forked_from: None, compacted_from: None, ..
} => {
assert_eq!(*id, session_id);
assert!(history.is_empty(), "repair must not admit model input");
prompt
}
other => panic!("repair must retain a complete initial head: {other:?}"),
};
let checkpoint = match &entries[1] {
LogEntry::Extension { domain, payload, .. }
if domain == "worker.protected_inputs.v1" => payload,
other => panic!("missing protected-input checkpoint: {other:?}"),
};
let expected_startups = usize::from(!materialized);
assert_eq!(startup_calls.load(Ordering::SeqCst), expected_startups);
assert_eq!(memory_calls.load(Ordering::SeqCst), expected_startups);
assert_eq!(calls.load(Ordering::SeqCst), 2 + expected_startups,
"lost receipt retry plus publication verification only for a pending head");
let mut operations = authorization_operations.lock().unwrap().clone();
operations.sort();
if materialized {
assert_eq!(session_files(), original);
assert_eq!(startup_calls.load(Ordering::SeqCst), 0);
assert_eq!(prompt, SAVED_SUBJEKTIV_SYSTEM_PROMPT);
assert_eq!(checkpoint, &serde_json::json!({ "version": 1, "kinds": [] }));
assert!(operations.is_empty(), "existing head repair must not refetch protected context");
} else {
assert_eq!(startup_calls.load(Ordering::SeqCst), 1);
assert!(prompt.contains("Repaired Subject resident behavior"));
assert_eq!(checkpoint, &serde_json::json!({
"version": 1, "kinds": ["subject-resident"],
}));
assert_eq!(operations, [
"journal.startup.read",
// Both the resident source and its Host adapter authorize the read.
"memory.resident.read", "memory.resident.read",
"memory.resident.supply", "memory.resident.supply",
], "empty Journal startup must not require a supply grant");
}
let after_repair = persisted_files(&aggregate);
let calls_after_repair = calls.load(Ordering::SeqCst);
let authorization_count = authorization_operations.lock().unwrap().len();
assert!(!factory.recover_saved_restore_source(&request).await.unwrap());
assert_eq!(persisted_files(&aggregate), after_repair);
assert_eq!(calls.load(Ordering::SeqCst), calls_after_repair);
assert_eq!(startup_calls.load(Ordering::SeqCst), expected_startups);
assert_eq!(memory_calls.load(Ordering::SeqCst), expected_startups);
assert_eq!(authorization_operations.lock().unwrap().len(), authorization_count);
assert!(factory.failed_restore_resources.lock().unwrap().is_empty());
drop(host);
Ok(())