From 87a1b6f1eb34ef3eef2913d5c1c7425b44c55e5f Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 00:00:05 +0900 Subject: [PATCH] fix: authorize bound standalone Subject body and Job operations --- crates/standalone/src/subjektiv.rs | 66 +++++++++++ crates/standalone/src/subjektiv/tests.rs | 133 +++++++++++++++++++++++ 2 files changed, 199 insertions(+) diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 353538072..371ae8495 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -803,6 +803,41 @@ impl LocalSubjectHost { } } impl SubjektivHost for LocalSubjectHost { + fn authorize_operation( + &self, + context: &SubjektivHostContext, + operation: &str, + ) -> Result<(), SubjektivHostError> { + // An explicit operator connection issues this local body capability. + // Revalidate its live writer on every access/supply, not Feature enablement. + // Job decisions and Backend conversations are not local body capabilities. + self.body(context)?; + match operation { + "session.record" + | "memory.resident.read" + | "memory.resident.supply" + | "memory.surface.supply" + | "memory.query" + | "memory.read" + | "memory.changes.list" + | "memory.proposal.validate" + | "memory.receipt.read" + | "memory.staging.write" + | "memory.extract.read" + | "memory.extract.supply" + | "memory.extract.commit" + | "memory.consolidation.request" + | "journal.startup.read" + | "journal.startup.supply" + | "journal.compaction.read" + | "journal.compaction.supply" + | "journal.recording.request" => Ok(()), + _ => Err(SubjektivHostError::PermissionDenied(format!( + "operation `{operation}` is not granted by the local Subject body capability" + ))), + } + } + fn memory_journal( &self, context: &SubjektivHostContext, @@ -1176,6 +1211,37 @@ impl LocalConsolidationHost { } } impl SubjektivHost for LocalConsolidationHost { + fn authorize_operation( + &self, + context: &SubjektivHostContext, + operation: &str, + ) -> Result<(), SubjektivHostError> { + self.live()?; + let attempt = format!("job:{}", self.attempt_id); + if context.worker_id != attempt || context.session_id != attempt { + return Err(SubjektivHostError::PermissionDenied( + "context is not bound to this local consolidation attempt".into(), + )); + } + // The Host-issued batch/attempt grant is independent of body permission. + // Domain operations still enforce candidate membership and saved receipts. + match operation { + "memory.query" + | "memory.read" + | "memory.changes.list" + | "memory.staging.list" + | "memory.staging.read" + | "memory.candidate.decide" + | "memory.surface.prepare" + | "memory.surface.supply" + | "memory.surface.publish" + | "memory.surface.fail" => Ok(()), + _ => Err(SubjektivHostError::PermissionDenied(format!( + "operation `{operation}` is not granted by the local consolidation attempt" + ))), + } + } + fn settings(&self) -> SubjektivHostSettings { self.host.settings() } diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index c0a17ad63..9104f0795 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -86,11 +86,56 @@ fn closed_capabilities_and_foreign_worker_context_are_rejected() { session_id: session_store::SessionId::now_v7().to_string(), }; assert!(port.memory(&owner, read.clone()).is_ok()); + let local_operations = [ + "memory.resident.read", + "memory.resident.supply", + "memory.staging.write", + "journal.startup.read", + "journal.startup.supply", + "journal.compaction.read", + "journal.compaction.supply", + "journal.recording.request", + ]; + for operation in local_operations { + assert!( + port.authorize_operation(&owner, operation).is_ok(), + "{operation}" + ); + assert!( + port.authorize_operation(&foreign, operation).is_err(), + "{operation}" + ); + } + for operation in [ + "chat.read", + "chat.send", + "worker.messaging", + "memory.candidate.decide", + "memory.surface.publish", + "memory.staging.list", + "journal", + "journal.startup", + "journal.startup.read.future", + ] { + assert!( + matches!( + port.authorize_operation(&owner, operation), + Err(SubjektivHostError::PermissionDenied(_)) + ), + "local body must not grant {operation}" + ); + } assert!( port.record_session(&owner, false).is_err(), "unattributed Session is not read authority" ); connection.close().unwrap(); + for operation in local_operations { + assert!( + port.authorize_operation(&owner, operation).is_err(), + "released capability: {operation}" + ); + } assert!(port.memory(&owner, read).is_err()); } @@ -134,3 +179,91 @@ fn different_local_roots_do_not_authorize_shared_subject_ids() { .is_err() ); } + +#[test] +fn consolidation_operation_policy_is_attempt_bound_and_does_not_inherit_body_rights() { + struct Fence(AtomicBool); + impl crate::jobs::JobAttemptFence for Fence { + fn ensure_live(&self) -> Result<(), String> { + self.0 + .load(Ordering::Acquire) + .then_some(()) + .ok_or("attempt revoked".into()) + } + } + let temp = tempfile::tempdir().unwrap(); + let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap(); + let subject = catalog.create("owner", None).unwrap(); + let mut body = catalog + .connect(&subject.id, protocol::WorkerId::now_v7()) + .unwrap(); + let fence = Arc::new(Fence(AtomicBool::new(true))); + let job = LocalConsolidationHost { + host: body.host.clone(), + grant: LocalConsolidationGrant { + binding: body.binding.clone(), + candidate_ids: vec![], + }, + job_id: "job-1".into(), + attempt_id: "attempt-1".into(), + fence: fence.clone(), + }; + let context = SubjektivHostContext { + worker_id: "job:attempt-1".into(), + session_id: "job:attempt-1".into(), + }; + assert!( + job.authorize_operation(&context, "memory.staging.list") + .is_ok() + ); + assert!( + job.authorize_operation(&context, "memory.surface.supply") + .is_ok() + ); + let foreign = SubjektivHostContext { + worker_id: "job:old-attempt".into(), + session_id: "job:old-attempt".into(), + }; + assert!( + job.authorize_operation(&foreign, "memory.staging.list") + .is_err() + ); + for op in [ + "session.record", + "memory.resident.read", + "memory.resident.supply", + "memory.extract.read", + "memory.staging.write", + "journal.startup.read", + "chat.send", + "memory", + ] { + assert!( + matches!( + job.authorize_operation(&context, op), + Err(SubjektivHostError::PermissionDenied(_)) + ), + "Job must not gain {op}" + ); + } + assert!( + job.memory( + &context, + SubjektivMemoryBackendOperation::ReadCandidate(SubjektivMemoryCandidateReadRequest { + candidate_id: "outside-batch".into() + }) + ) + .is_err() + ); + fence.0.store(false, Ordering::Release); + assert!( + job.authorize_operation(&context, "memory.staging.list") + .is_err() + ); + fence.0.store(true, Ordering::Release); + body.close().unwrap(); + assert!( + job.authorize_operation(&context, "memory.staging.list") + .is_err() + ); +}