diff --git a/crates/manager-core/src/group_pubsub_service.rs b/crates/manager-core/src/group_pubsub_service.rs index 9c3b56e..c79457d 100644 --- a/crates/manager-core/src/group_pubsub_service.rs +++ b/crates/manager-core/src/group_pubsub_service.rs @@ -317,4 +317,137 @@ mod tests { .unwrap(); assert_eq!(count, 2); } + + // ------------------------------------------------------------------ + // The two invariants the three older sibling services pin, and this one + // did not. Both existing tests above use `FixedResolver` (always resolves) + // and `owner_cx` (an instance Owner, who bypasses the capability check) — + // so neither the isolation boundary nor the anon gate was exercised at all. + // ------------------------------------------------------------------ + + /// Resolves a topic only for apps whose chain declares it — the fake stand-in + /// for the ancestor walk. The real walk is pinned against Postgres in + /// `tests/group_collection_isolation.rs`. + #[derive(Default)] + struct ChainResolver { + map: std::collections::HashMap<(AppId, String), GroupId>, + } + #[async_trait] + impl GroupCollectionResolver for ChainResolver { + async fn resolve_owning_group( + &self, + app_id: AppId, + name: &str, + kind: &str, + ) -> Result, sqlx::Error> { + if kind != KIND_TOPIC { + return Ok(None); + } + Ok(self.map.get(&(app_id, name.to_lowercase())).copied()) + } + } + + fn cx(app_id: AppId, principal: Option) -> SdkCallCx { + SdkCallCx { + app_id, + script_id: ScriptId::new(), + principal, + execution_id: ExecutionId::new(), + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: ExecutionId::new(), + is_dead_letter_handler: false, + event: None, + } + } + + fn member_no_role() -> Principal { + Principal { + user_id: AdminUserId::new(), + instance_role: InstanceRole::Member, + scopes: None, + app_binding: None, + } + } + + fn owner_principal() -> Principal { + Principal { + user_id: AdminUserId::new(), + instance_role: InstanceRole::Owner, + scopes: None, + app_binding: None, + } + } + + #[tokio::test] + async fn unrelated_app_gets_collection_not_shared() { + let (app_a, app_b, group) = (AppId::new(), AppId::new(), GroupId::new()); + let mut resolver = ChainResolver::default(); + resolver.map.insert((app_a, "events".into()), group); + let svc = GroupPubsubServiceImpl::new( + Arc::new(FakeRepo), + Arc::new(resolver), + Arc::new(DenyAuthz), + ); + + // Positive control — app_a, on the chain, publishes fine. + svc.publish(&owner_cx(app_a), "events", "created", serde_json::json!({})) + .await + .expect("an app on the chain can publish"); + + // THE BOUNDARY — app_b is in another subtree. + let err = svc + .publish( + &cx(app_b, Some(owner_principal())), + "events", + "created", + serde_json::json!({}), + ) + .await + .unwrap_err(); + assert!( + matches!(&err, GroupPubsubError::CollectionNotShared(t) if t == "events"), + "a foreign app must not publish into another subtree's shared topic, got {err:?}" + ); + } + + #[tokio::test] + async fn publish_fails_closed_for_anon() { + let (app, group) = (AppId::new(), GroupId::new()); + let mut resolver = ChainResolver::default(); + resolver.map.insert((app, "events".into()), group); + let svc = GroupPubsubServiceImpl::new( + Arc::new(FakeRepo), + Arc::new(resolver), + Arc::new(DenyAuthz), + ); + + // A shared publish fans out to every `shared = true` trigger on the group, + // each handler running under the writer's app_id. An anonymous public route + // must not be able to drive that. This is the assertion that fires if + // `script_gate_require_principal` is swapped for `script_gate`. + let err = svc + .publish(&cx(app, None), "events", "created", serde_json::json!({})) + .await + .unwrap_err(); + assert!( + matches!(err, GroupPubsubError::Forbidden), + "an ANONYMOUS shared publish must fail closed, got {err:?}" + ); + + // Authenticated but with no editor+ on the owning group: also denied. + let err = svc + .publish( + &cx(app, Some(member_no_role())), + "events", + "created", + serde_json::json!({}), + ) + .await + .unwrap_err(); + assert!( + matches!(err, GroupPubsubError::Forbidden), + "authentication alone is not authorization — editor+ on the owning group is required" + ); + } } diff --git a/crates/manager-core/src/group_queue_service.rs b/crates/manager-core/src/group_queue_service.rs index 0e28180..823c731 100644 --- a/crates/manager-core/src/group_queue_service.rs +++ b/crates/manager-core/src/group_queue_service.rs @@ -150,3 +150,251 @@ impl GroupQueueService for GroupQueueServiceImpl { .map_err(|e| GroupQueueError::Backend(e.to_string())) } } + +/// This service shipped with NO tests at all, while its three older siblings +/// (`group_kv_service`, `group_docs_service`, `group_files_service`) each pin the +/// same two invariants. Both matter more here than anywhere else: a shared-queue +/// consumer is materialized into EVERY descendant app, so an unauthorized enqueue +/// is attacker-controlled input fanned out as script execution across a whole +/// subtree. +/// +/// The gap was concrete, not theoretical: swapping `script_gate_require_principal` +/// for `script_gate` (a one-token edit; the latter returns `Ok(())` for an +/// anonymous principal) would let an unauthenticated public HTTP route enqueue +/// into any ancestor group's shared queue — and nothing would have failed. +#[cfg(test)] +mod tests { + use super::*; + use crate::authz::AuthzError; + use crate::group_queue_repo::{ClaimedGroupMessage, GroupQueueRepoError}; + use picloud_shared::{ + AdminUserId, AppId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId, + UserId, + }; + use std::collections::HashMap; + use uuid::Uuid; + + /// Only `enqueue` / `depth` / `depth_pending` are reachable from the service — + /// consumption is the dispatcher's job — so the claim side is deliberately + /// unreachable rather than faked. + #[derive(Default)] + struct RecordingRepo { + enqueued: tokio::sync::Mutex>, + } + + #[async_trait] + impl GroupQueueRepo for RecordingRepo { + async fn enqueue( + &self, + msg: NewGroupQueueMessage, + ) -> Result { + self.enqueued + .lock() + .await + .push((msg.group_id, msg.collection)); + Ok(QueueMessageId::new()) + } + async fn claim( + &self, + _: GroupId, + _: &str, + ) -> Result, GroupQueueRepoError> { + unreachable!("the service never consumes") + } + async fn ack(&self, _: QueueMessageId, _: Uuid) -> Result { + unreachable!("the service never consumes") + } + async fn nack( + &self, + _: QueueMessageId, + _: Uuid, + _: chrono::Duration, + ) -> Result { + unreachable!("the service never consumes") + } + async fn release( + &self, + _: QueueMessageId, + _: Uuid, + _: chrono::Duration, + ) -> Result { + unreachable!("the service never consumes") + } + #[allow(clippy::too_many_arguments)] + async fn dead_letter( + &self, + _: QueueMessageId, + _: Uuid, + _: GroupId, + _: &str, + _: Option, + _: Option, + _: u32, + _: chrono::DateTime, + _: &str, + ) -> Result { + unreachable!("the service never consumes") + } + async fn reclaim_visibility_timeouts(&self) -> Result { + unreachable!("the service never consumes") + } + async fn depth(&self, _: GroupId, _: &str) -> Result { + Ok(7) + } + async fn depth_pending(&self, _: GroupId, _: &str) -> Result { + Ok(3) + } + } + + /// Models the ancestor-chain walk: a queue resolves only for apps whose chain + /// declares it, and only under `kind = "queue"`. The REAL walk is pinned + /// against Postgres in `tests/group_collection_isolation.rs`; this stands in + /// for it so the service's own plumbing can be tested without a DB. + #[derive(Default)] + struct FakeResolver { + map: HashMap<(AppId, String), GroupId>, + } + + #[async_trait] + impl GroupCollectionResolver for FakeResolver { + async fn resolve_owning_group( + &self, + app_id: AppId, + name: &str, + kind: &str, + ) -> Result, sqlx::Error> { + if kind != KIND_QUEUE { + return Ok(None); + } + Ok(self.map.get(&(app_id, name.to_lowercase())).copied()) + } + } + + struct DenyingAuthzRepo; + #[async_trait] + impl AuthzRepo for DenyingAuthzRepo { + async fn membership(&self, _: UserId, _: AppId) -> Result, AuthzError> { + Ok(None) + } + } + + fn cx_with(app_id: AppId, principal: Option) -> SdkCallCx { + SdkCallCx { + app_id, + script_id: ScriptId::new(), + principal, + execution_id: ExecutionId::new(), + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: ExecutionId::new(), + is_dead_letter_handler: false, + event: None, + } + } + + fn owner() -> Principal { + Principal { + user_id: AdminUserId::new(), + instance_role: InstanceRole::Owner, + scopes: None, + app_binding: None, + } + } + + fn member_no_role() -> Principal { + Principal { + user_id: AdminUserId::new(), + instance_role: InstanceRole::Member, + scopes: None, + app_binding: None, + } + } + + fn svc(resolver: FakeResolver) -> GroupQueueServiceImpl { + GroupQueueServiceImpl::new( + Arc::new(RecordingRepo::default()), + Arc::new(resolver), + Arc::new(DenyingAuthzRepo), + ) + } + + #[tokio::test] + async fn unrelated_app_gets_collection_not_shared() { + // app_a's chain declares "jobs"; app_b's does not. + let (app_a, app_b, group) = (AppId::new(), AppId::new(), GroupId::new()); + let mut resolver = FakeResolver::default(); + resolver.map.insert((app_a, "jobs".into()), group); + let q = svc(resolver); + + // Positive control: app_a really can enqueue and read the depth. Without + // this, a service that rejected EVERYTHING would pass the negative below. + let cx_a = cx_with(app_a, Some(owner())); + q.enqueue( + &cx_a, + "jobs", + serde_json::json!({"n": 1}), + EnqueueOpts::default(), + ) + .await + .expect("an app on the chain can enqueue"); + assert_eq!(q.depth(&cx_a, "jobs").await.unwrap(), 7); + + // THE BOUNDARY: app_b is off-chain, so the name does not resolve — on the + // write path and on the read path alike. + let cx_b = cx_with(app_b, Some(owner())); + let err = q + .enqueue(&cx_b, "jobs", serde_json::json!({}), EnqueueOpts::default()) + .await + .unwrap_err(); + assert!( + matches!(&err, GroupQueueError::CollectionNotShared(c) if c == "jobs"), + "a foreign app must not reach another subtree's shared queue, got {err:?}" + ); + let err = q.depth(&cx_b, "jobs").await.unwrap_err(); + assert!(matches!(&err, GroupQueueError::CollectionNotShared(c) if c == "jobs")); + } + + #[tokio::test] + async fn enqueue_fails_closed_for_anon_while_depth_stays_open() { + let (app, group) = (AppId::new(), GroupId::new()); + let mut resolver = FakeResolver::default(); + resolver.map.insert((app, "jobs".into()), group); + let q = svc(resolver); + + // Reads are open — the declaration IS the grant, anonymous included. + let anon = cx_with(app, None); + assert_eq!( + q.depth(&anon, "jobs").await.unwrap(), + 7, + "depth is a read; reads are open to any subtree script" + ); + + // Writes fail closed. This is the assertion that fires if + // `script_gate_require_principal` is ever swapped for `script_gate`. + let err = q + .enqueue(&anon, "jobs", serde_json::json!({}), EnqueueOpts::default()) + .await + .unwrap_err(); + assert!( + matches!(err, GroupQueueError::Forbidden), + "an ANONYMOUS enqueue must fail closed — a public route must not be able to \ + inject work that every descendant app's consumer will execute; got {err:?}" + ); + + // Authenticated but without an editor+ role on the owning group: also denied. + let member = cx_with(app, Some(member_no_role())); + let err = q + .enqueue( + &member, + "jobs", + serde_json::json!({}), + EnqueueOpts::default(), + ) + .await + .unwrap_err(); + assert!( + matches!(err, GroupQueueError::Forbidden), + "authentication alone is not authorization — editor+ on the owning group is required" + ); + } +}