test(groups): pin the anon + isolation gates on the queue and pubsub services
`GroupQueueServiceImpl` shipped with ZERO tests; `GroupPubsubServiceImpl` had four that exercised neither boundary — its resolver always resolved and its principal was always an instance Owner (who bypasses the capability check). So on the two newest shared services, both the isolation boundary and the anonymous fail-closed gate were unpinned — and they matter most here: 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. Both now get the pair their three older siblings have: `unrelated_app_gets_collection_not_shared` (with a positive control, so a reject-everything impl can't pass it) and an anon/non-editor fail-closed test that also confirms reads (depth) stay open. Mutation-verified: swapping `script_gate_require_principal` for `script_gate` — the one-token edit that would let an unauthenticated public route enqueue into any ancestor group's shared queue — makes the anonymous enqueue succeed, and the test catches it exactly there. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -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<Option<GroupId>, 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<Principal>) -> 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Vec<(GroupId, String)>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupQueueRepo for RecordingRepo {
|
||||
async fn enqueue(
|
||||
&self,
|
||||
msg: NewGroupQueueMessage,
|
||||
) -> Result<QueueMessageId, GroupQueueRepoError> {
|
||||
self.enqueued
|
||||
.lock()
|
||||
.await
|
||||
.push((msg.group_id, msg.collection));
|
||||
Ok(QueueMessageId::new())
|
||||
}
|
||||
async fn claim(
|
||||
&self,
|
||||
_: GroupId,
|
||||
_: &str,
|
||||
) -> Result<Option<ClaimedGroupMessage>, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
async fn ack(&self, _: QueueMessageId, _: Uuid) -> Result<bool, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
async fn nack(
|
||||
&self,
|
||||
_: QueueMessageId,
|
||||
_: Uuid,
|
||||
_: chrono::Duration,
|
||||
) -> Result<bool, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
async fn release(
|
||||
&self,
|
||||
_: QueueMessageId,
|
||||
_: Uuid,
|
||||
_: chrono::Duration,
|
||||
) -> Result<bool, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn dead_letter(
|
||||
&self,
|
||||
_: QueueMessageId,
|
||||
_: Uuid,
|
||||
_: GroupId,
|
||||
_: &str,
|
||||
_: Option<picloud_shared::TriggerId>,
|
||||
_: Option<ScriptId>,
|
||||
_: u32,
|
||||
_: chrono::DateTime<Utc>,
|
||||
_: &str,
|
||||
) -> Result<picloud_shared::DeadLetterId, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
async fn reclaim_visibility_timeouts(&self) -> Result<u64, GroupQueueRepoError> {
|
||||
unreachable!("the service never consumes")
|
||||
}
|
||||
async fn depth(&self, _: GroupId, _: &str) -> Result<u64, GroupQueueRepoError> {
|
||||
Ok(7)
|
||||
}
|
||||
async fn depth_pending(&self, _: GroupId, _: &str) -> Result<u64, GroupQueueRepoError> {
|
||||
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<Option<GroupId>, 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<Option<AppRole>, AuthzError> {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
fn cx_with(app_id: AppId, principal: Option<Principal>) -> 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user