diff --git a/.changeset/g7-multitenancy-isolation.md b/.changeset/g7-multitenancy-isolation.md new file mode 100644 index 00000000..174965c1 --- /dev/null +++ b/.changeset/g7-multitenancy-isolation.md @@ -0,0 +1,62 @@ +--- +"@smooai/smooth-operator": minor +--- + +fix(security): enforce tenant isolation on the by-id session paths and the knowledge store (feature gap G7) + +Closes G7 with a **shared** conformance suite — `rust/adapters/multitenancy_suite.rs`, +one body run by the in-memory, Postgres and DynamoDB adapters — plus a +server-level suite driving the real `handle_frame` from an attacker in another +org. Writing it found two live cross-tenant holes. + +**1. Cross-tenant session access on every by-id path (WS server + Lambda).** +The connection's org was resolved only to *stamp* newly created sessions. Every +by-id action — `get_session`, `get_conversation_messages`, `send_message`, +`confirm_tool_action`, `submit_interaction`, `verify_otp`, `rename_conversation`, +and conversation resume — went through `may_read_conversation`, which checks the +**owner email** and never the org. Its deliberate ownerless-is-open rule (a +conversation with no `user` participant carrying an email stays readable, so +anonymous principals keep their own sessions) is exactly the embeddable widget's +default state, so an attacker authenticated to org B who learned an org-A session +id could read that session, replay its whole history through a turn, retitle its +conversation, and resume it (minting a session bound to the victim's org, which +then flows into the turn's `ToolProviderContext`). The Lambda transport had **no** +check at all — `dispatch::get_session` / `send_message` acted on whatever +`storage.get_session` returned. + +Fixed at the chokepoints: `scoped_session` and `may_read_conversation` now take +the connection's `auth_org` and refuse a row belonging to another tenant +(indistinguishably from not-found), and the Lambda gained the same check off the +frame's verified principal. A connection with **no** verified org (anonymous / +tokenless — the widget's normal state) is unchanged. + +**2. Knowledge was not tenant-isolated on the in-memory adapter, and the admin +connector-index path ingested org-blind.** `AclKnowledgeStore` filtered by +user/group only, on the assumption that the wrapped store was already +org-partitioned — true for Postgres/DynamoDB, false for the in-memory adapter and +for any third-party adapter using the `knowledge_for_access` trait default. And +`POST /admin/connectors/{id}/index` ingested through the org-blind `knowledge()` +handle for every tenant: Postgres wrote `organization_id = NULL` (which the +org-filtered read can never match, so connector-ingested knowledge silently +returned nothing) and DynamoDB wrote whichever partition the adapter was +constructed for. + +- `AclKnowledgeStore` now records each document's owning org (from the + `org_id` metadata the ingestion pipeline stamps, falling back to the org the + ingesting handle is bound to) and enforces the tenant boundary **before** the + ACL. +- `DynamoKnowledgeBase` honours `AccessContext::organization_id` for the query + partition and the document's own `org_id` for the ingest partition, mirroring + what `PgKnowledgeBase::with_access` already did. +- `PgKnowledgeBase::ingest` prefers the document's `org_id` over the handle's, so + the org-blind handle still lands rows in the right tenant. +- The admin index run ingests through `knowledge_for_access`. + +**Behavior change worth reading before upgrading.** A retrieval whose +`AccessContext` carries an org now sees **only** documents recorded as that org's +— matching the Postgres backend's existing SQL pre-filter, so all three backends +finally agree. A document ingested through the raw `knowledge()` handle with no +`org_id` metadata belongs to no tenant and is therefore invisible to a turn that +has one. If you seed knowledge directly, either stamp `org_id` on the document or +ingest through `storage.knowledge_for_access(&AccessContext::default().with_organization_id(org))` +— which is what the reference server's seeding and the admin index path do. diff --git a/docs/Planning/Feature Gaps.md b/docs/Planning/Feature Gaps.md index 9326c34b..d4ac164c 100644 --- a/docs/Planning/Feature Gaps.md +++ b/docs/Planning/Feature Gaps.md @@ -74,9 +74,55 @@ Mature platforms ship extensive web + Playwright suites. - ✅ **Done.** `.github/workflows/pr-kind-deploy-smoke.yml` runs the planned `kind` job on every PR: `helm install` into an ephemeral cluster, then the protocol smoke against the live pod. It is a required-looking check on current PRs (observed green on #526, 2026-08-22). - This entry read "we only `helm lint`/`helm template`" for some time after the job shipped. A gap doc that is stale in the CLOSED direction is worse than one that is merely incomplete: it argues for work that already exists. When you close a gap here, edit this file in the same PR. -### G7. Multi-tenancy +### G7. Multi-tenancy — ✅ isolation is now a test, and closing two live leaks Mature knowledge platforms support multi-tenant schemas. Our org scoping is row-level only. - **TDD**: `tests/multitenancy.rs` first — two orgs, assert full isolation across conversations/knowledge/checkpoints on both adapters. (Likely already passes for OLTP via `organizationId`; the test makes it a guarantee and covers the knowledge/S3-Vectors index-per-org path.) +- ✅ **Done — and the "likely already passes" framing above was wrong twice.** One + **shared** suite (`rust/adapters/multitenancy_suite.rs`, `#[path]`-included by + each adapter's `tests/multitenancy.rs`) runs the same body against **in-memory, + Postgres and DynamoDB**, with a positive control on every isolation assertion so + a backend that returns nothing can't pass vacuously. A second suite + (`smooth-operator-server/tests/multitenancy.rs`) drives the real + `handler::handle_frame` from an attacker authenticated to another org. + + **Now guaranteed by a test, on all three backends:** + - conversation listings are org-partitioned, by-org and by-org-and-user — asserted with the **same user email owning one conversation in each org**, so the isolation cannot be incidentally coming from a differing owner; + - the **idempotency claim is per-org** (`(organization_id, idempotency_key)`): two orgs using the same key get two distinct conversations, where an org-blind claim would have handed org B **org A's conversation row**; + - message pages, participants and sessions ride their own org's conversation, and a session update in one org does not touch the other's row; + - knowledge retrieval bound to org B never returns org A's document — *including* documents ingested through the org-blind `knowledge()` handle, which must land in the tenant their `org_id` metadata names; + - checkpoints saved under one agent id are invisible under another. + + **What it found (both fixed in the same PR, each proven by reverting the fix and re-running):** + 1. 🚨 **Cross-tenant session access on every by-id path.** Org was resolved per + connection only to *stamp* new sessions; `may_read_conversation` checked the + **owner email** and never the org, and its deliberate ownerless-is-open rule + is exactly the widget's default state. An attacker authenticated to org B who + learned an org-A session id could read the session, replay its history through + a turn, retitle the conversation, and resume it — minting a session bound to + the victim's org, which flows into the turn's `ToolProviderContext`. The + **Lambda transport had no check at all**. Fixed at the `scoped_session` / + `may_read_conversation` chokepoints (and the Lambda's `get_session` / + `send_message`), denying indistinguishably from not-found. + 2. 🚨 **Knowledge was not tenant-isolated where the backend isn't + org-partitioned.** `AclKnowledgeStore` filtered by user/group only, assuming + the wrapped store had already done the org filter — true for Postgres/DynamoDB, + false for the in-memory adapter and any adapter using the + `knowledge_for_access` trait default. Compounding it, `POST + /admin/connectors/{id}/index` ingested through the org-blind `knowledge()` + handle for every tenant (same shape as G3: the seam existed, one caller went + around it), so on Postgres every connector document was written with + `organization_id = NULL` — invisible to every org-scoped read. The ACL store now + records each document's org and enforces the tenant boundary before the ACL; + DynamoDB honours `AccessContext::organization_id` for its query partition + (Postgres already did); both backends prefer the document's own `org_id` at + ingest; and the admin run goes through the org-bound seam. + + **Still NOT guaranteed — merely true today (residuals, deliberately not dressed up):** + - `StorageAdapter`'s by-id reads (`get_conversation`, `get_message`, `get_session`, `list_participants_by_conversation`) take **no org** and are not org-checked at the adapter. Enforcement lives at the caller; a new caller can still forget. + - A connection with **no** verified org (anonymous / tokenless — the widget's normal state) is not org-checked at all. Fail-closing it would deny the widget its own session; a deployment needing hard isolation must require auth (`strict_auth`). + - A conversation with **no participants yet** (the create→first-frame race) has no derivable org, so the conversation-id path falls through to the ownership check. The session-id path is unaffected — a `Session` always carries its org. + - `CheckpointStore` has **no org dimension** at all: isolation rests entirely on agent-id uniqueness (the server mints a fresh UUID per `Agent`, and never reads checkpoints back). A host that reuses a stable agent id across tenants would commingle conversation state. + - **S3 Vectors is UNVERIFIED.** The index-per-org path is behind the `s3-vectors` feature and needs real AWS; the suite exercises the brute-force DynamoDB backend only. ### G8. Model-server parity (embedding/rerank) — ✅ rerank stage shipped Mature knowledge platforms have a dedicated, tested model server (embeddings + rerank + intent). We have a pluggable `Embedder` + RRF; the rerank stage is now implemented as a pluggable seam mirroring the `Embedder` pattern. diff --git a/rust/adapters/dynamodb/src/knowledge.rs b/rust/adapters/dynamodb/src/knowledge.rs index 518f728c..3e66e6a3 100644 --- a/rust/adapters/dynamodb/src/knowledge.rs +++ b/rust/adapters/dynamodb/src/knowledge.rs @@ -53,8 +53,10 @@ pub struct DynamoKnowledgeBase { table: String, embedder: Arc, handle: Handle, - /// Org partition for ingest/query. The engine is single-org per agent, so a - /// fixed org keeps the brute-force scan scoped to one partition. + /// Construction-time org partition — the single-tenant default, and the + /// fallback when neither the document nor the requester names an org. A + /// multi-tenant host overrides it per turn: see [`Self::ingest_org`] and + /// [`Self::query_org`]. organization_id: String, backend: KnowledgeBackend, /// Optional document-level access control (feature gap G3). When set, the @@ -110,6 +112,37 @@ impl DynamoKnowledgeBase { } } + /// The org partition `doc` is written to: the document's own `org_id` + /// metadata (the ingestion pipeline stamps it on every chunk) when present, + /// else the access-bound org, else the construction-time org. + /// + /// Without this, every tenant's documents landed in the construction-time + /// partition — one adapter instance could not ingest for more than one org, + /// and a multi-tenant host that threaded the turn's org on the + /// `AccessContext` (as Postgres already honoured) silently wrote the wrong + /// tenant's partition. + fn ingest_org<'a>(&'a self, doc: &'a Document) -> &'a str { + doc.metadata + .get(smooth_operator::access_control::ORG_METADATA_KEY) + .map(String::as_str) + .or_else(|| { + self.access + .as_ref() + .and_then(|a| a.organization_id.as_deref()) + }) + .unwrap_or(&self.organization_id) + } + + /// The org partition a query reads from: the requester's org when the + /// handle is access-bound (multi-tenant host), else the construction-time + /// org (single-tenant). Mirrors `PgKnowledgeBase::with_access`. + fn query_org(&self) -> &str { + self.access + .as_ref() + .and_then(|a| a.organization_id.as_deref()) + .unwrap_or(&self.organization_id) + } + fn run_blocking(&self, fut: F) -> Result where F: std::future::Future> + Send + 'static, @@ -156,14 +189,12 @@ impl DynamoKnowledgeBase { .map(|f| AttributeValue::N(f.to_string())) .collect(), ); + let ingest_org = self.ingest_org(&doc).to_string(); let mut put = self .client .put_item() .table_name(&self.table) - .item( - attr::PK, - AttributeValue::S(keys::knowledge_pk(&self.organization_id)), - ) + .item(attr::PK, AttributeValue::S(keys::knowledge_pk(&ingest_org))) .item(attr::SK, AttributeValue::S(keys::knowledge_sk(&doc.id))) .item(attr::ENTITY, AttributeValue::S("knowledge".to_string())) .item("documentId", AttributeValue::S(doc.id.clone())) @@ -181,9 +212,7 @@ impl DynamoKnowledgeBase { // S3 Vectors path additionally writes the embedding to its index. #[cfg(feature = "s3-vectors")] if let Some(store) = &self.s3vectors { - store - .upsert(&self.organization_id, &doc, &embedding) - .await?; + store.upsert(&ingest_org, &doc, &embedding).await?; } Ok(()) @@ -206,7 +235,7 @@ impl DynamoKnowledgeBase { .s3vectors .as_ref() .ok_or_else(|| anyhow!("s3 vectors store not initialized"))?; - store.query(&self.organization_id, &query_vec, limit).await + store.query(self.query_org(), &query_vec, limit).await } } } @@ -237,7 +266,7 @@ impl DynamoKnowledgeBase { .expression_attribute_names("#sk", attr::SK) .expression_attribute_values( ":pk", - AttributeValue::S(keys::knowledge_pk(&self.organization_id)), + AttributeValue::S(keys::knowledge_pk(self.query_org())), ) .expression_attribute_values( ":skp", diff --git a/rust/adapters/dynamodb/tests/multitenancy.rs b/rust/adapters/dynamodb/tests/multitenancy.rs new file mode 100644 index 00000000..4fabad31 --- /dev/null +++ b/rust/adapters/dynamodb/tests/multitenancy.rs @@ -0,0 +1,23 @@ +//! Multi-tenancy conformance (feature gap G7) for the DynamoDB single-table +//! `StorageAdapter`, against a real `amazon/dynamodb-local` container. +//! +//! ONE adapter instance serving TWO orgs — the multi-tenant pod shape. The +//! knowledge slice partitions per tenant: ingest writes the document's own +//! `org_id`, and a query reads the requester's org partition. +//! +//! The suite body is shared with the in-memory and Postgres adapters — see +//! `rust/adapters/multitenancy_suite.rs`. Skip policy: `tests/common/mod.rs`. + +mod common; + +#[path = "../../multitenancy_suite.rs"] +mod suite; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_orgs_are_isolated_on_one_dynamodb_adapter() -> anyhow::Result<()> { + let Some((_node, store)) = common::start().await? else { + return Ok(()); // Docker unavailable or port unreachable — skip, don't fail. + }; + suite::assert_multitenancy(&store, &store, "ddb").await; + Ok(()) +} diff --git a/rust/adapters/in-memory/tests/multitenancy.rs b/rust/adapters/in-memory/tests/multitenancy.rs new file mode 100644 index 00000000..489e2cba --- /dev/null +++ b/rust/adapters/in-memory/tests/multitenancy.rs @@ -0,0 +1,18 @@ +//! Multi-tenancy conformance (feature gap G7) for the in-memory `StorageAdapter`. +//! +//! ONE adapter instance serving TWO orgs — the shared-pod shape, and the harshest +//! variant: every slice (tables, knowledge side table, checkpoints) is literally +//! the same object for both tenants, so any missing org partition shows up +//! immediately rather than being hidden behind two separate instances. +//! +//! The suite body is shared with the Postgres and DynamoDB adapters — see +//! `rust/adapters/multitenancy_suite.rs`. + +#[path = "../../multitenancy_suite.rs"] +mod suite; + +#[tokio::test] +async fn two_orgs_are_isolated_on_one_in_memory_adapter() { + let store = smooth_operator_adapter_memory::InMemoryStorageAdapter::new(); + suite::assert_multitenancy(&store, &store, "mem").await; +} diff --git a/rust/adapters/multitenancy_suite.rs b/rust/adapters/multitenancy_suite.rs new file mode 100644 index 00000000..de1a69ba --- /dev/null +++ b/rust/adapters/multitenancy_suite.rs @@ -0,0 +1,432 @@ +//! **Shared** multi-tenancy conformance suite (feature gap G7) — one suite, every +//! storage backend. +//! +//! Included verbatim by each adapter's `tests/multitenancy.rs` via +//! `#[path = "../../multitenancy_suite.rs"] mod suite;`, so the isolation +//! property is asserted from exactly ONE source. A backend that drifts fails +//! here, not in a per-adapter copy nobody kept in sync. +//! +//! ## The shape it tests +//! +//! One process, ONE adapter instance, two tenants (`ORG_A` / `ORG_B`) — the +//! multi-tenant pod shape, and the harshest variant: every slice is literally +//! the same object for both tenants, so a missing org partition shows up +//! immediately instead of hiding behind two separate instances. The two +//! `&dyn StorageAdapter` parameters are each org's *view* of that one store; +//! every caller passes the same handle twice today. +//! +//! ## What it guarantees +//! +//! 1. **Conversations** — org A's conversations never appear in an org B listing, +//! by-org or by-org-and-user, even when the SAME user email owns one in each. +//! 2. **Idempotency keys are per-org** — the same `idempotency_key` used by two +//! orgs yields two distinct conversations. If the idempotency claim were +//! org-blind, org B's create would hand back **org A's conversation row**. +//! 3. **Messages / participants / sessions** — reachable only through their own +//! org's conversation; a cross-org listing never sees them, and an update in +//! one org does not touch the other's row. +//! 4. **Knowledge** — a document ingested for org A is never returned to a +//! retrieval bound to org B's [`AccessContext`], on every backend; and a +//! document ingested through the **org-blind** `knowledge()` handle still +//! lands in the tenant its `org_id` metadata names (neither lost nor shared). +//! 5. **Checkpoints** — a checkpoint saved under one agent id is not visible +//! under another. +//! +//! ## What it deliberately does NOT claim +//! +//! `StorageAdapter`'s by-id reads (`get_conversation`, `get_message`, +//! `get_session`, `list_participants_by_conversation`) take no org and are +//! **not** org-checked at the adapter — enforcement lives at the caller +//! (`smooth-operator-server`'s `scoped_session`). The suite asserts the *listing* +//! boundary, which is the one the adapter owns. + +#![allow(dead_code)] + +use chrono::Utc; + +use smooth_operator::access_control::AccessContext; +use smooth_operator::adapter::{MessageQuery, SessionUpdate, StorageAdapter}; +use smooth_operator::domain::{ + Conversation, Direction, Message, MessageContent, Participant, ParticipantType, Platform, + Session, SessionStatus, +}; +use smooth_operator_core::{ + Checkpoint, Conversation as EngineConversation, Document, DocumentType, +}; + +pub const ORG_A: &str = "org-alpha"; +pub const ORG_B: &str = "org-beta"; + +/// The metadata key every ingested chunk carries naming its owning org — set by +/// the ingestion pipeline (`ingestion::pipeline`) on every document. Kept in +/// sync with `smooth_operator::access_control::ORG_METADATA_KEY`. +pub const ORG_METADATA_KEY: &str = "org_id"; + +/// A shared user email — the SAME person in both orgs. Proves isolation is by +/// org, not incidentally by a differing owner email. +pub const SHARED_EMAIL: &str = "shared.person@example.test"; + +/// The idempotency key both orgs use, verbatim. The collision is the point. +const SHARED_IDEMPOTENCY_KEY: &str = "idem-shared-across-tenants"; + +fn conversation(id: &str, org: &str, idempotency_key: &str) -> Conversation { + Conversation { + id: id.into(), + platform: Platform::Web, + name: format!("{org} chat"), + organization_id: org.into(), + idempotency_key: idempotency_key.into(), + metadata_json: None, + analytics_json: None, + created_at: Utc::now(), + updated_at: Utc::now(), + } +} + +fn owner(id: &str, conv: &str, org: &str) -> Participant { + Participant { + id: id.into(), + conversation_id: conv.into(), + organization_id: org.into(), + participant_type: ParticipantType::User, + external_id: None, + internal_id: None, + browser_fingerprint: None, + browser_info: None, + name: id.into(), + email: Some(SHARED_EMAIL.into()), + phone: None, + crm_contact_id: None, + metadata_json: None, + created_at: Utc::now(), + updated_at: Utc::now(), + } +} + +fn message(id: &str, conv: &str, org: &str, text: &str) -> Message { + Message { + id: id.into(), + external_id: None, + organization_id: Some(org.into()), + conversation_id: Some(conv.into()), + direction: Direction::Inbound, + content: MessageContent::from_text(text), + from: None, + to: None, + metadata_json: None, + analytics_json: None, + created_at: Utc::now(), + updated_at: None, + } +} + +fn session(id: &str, conv: &str, org: &str) -> Session { + Session { + session_id: id.into(), + conversation_id: conv.into(), + organization_id: org.into(), + agent_id: Some(format!("agent-{org}")), + agent_name: "Smantha".into(), + user_participant_id: format!("owner-{org}"), + agent_participant_id: format!("bot-{org}"), + thread_id: format!("thread-{org}"), + status: Some(SessionStatus::Active), + token_count: Some(0), + message_count: Some(0), + metadata: None, + created_at: Some(Utc::now()), + updated_at: Some(Utc::now()), + ended_at: None, + last_activity_at: Some(Utc::now()), + } +} + +/// A document owned by `org`, stamped the way the ingestion pipeline stamps it. +/// `marker` is a distinctive term so retrieval can be asserted by content. +fn org_document(id: &str, org: &str, marker: &str) -> Document { + let mut doc = Document::new( + format!("The {marker} escalation code for this tenant is {marker}-9137."), + format!("policies/{org}.md"), + DocumentType::Documentation, + ) + .with_metadata(ORG_METADATA_KEY, org) + // The pipeline also stamps `document_id`; the Postgres backend stores it as + // the result's `document_id`, which is the key the org/ACL side table uses. + .with_metadata("document_id", id); + // Pin the id so `KnowledgeResult::document_id` is assertable (the in-memory + // backend reports the document's own id; `Document::new` would randomise it). + doc.id = id.into(); + doc +} + +/// Run the whole suite. `a` is the org-A-scoped handle, `b` the org-B-scoped +/// handle, both over the same backing store. `suffix` disambiguates row ids so +/// the suite can run repeatedly against a shared (containerised) database. +pub async fn assert_multitenancy(a: &dyn StorageAdapter, b: &dyn StorageAdapter, suffix: &str) { + let conv_a = format!("conv-a-{suffix}"); + let conv_b = format!("conv-b-{suffix}"); + + // ---- 1. conversations are listed per org ------------------------------ + a.create_conversation(conversation(&conv_a, ORG_A, &format!("idem-a-{suffix}"))) + .await + .expect("create org-A conversation"); + b.create_conversation(conversation(&conv_b, ORG_B, &format!("idem-b-{suffix}"))) + .await + .expect("create org-B conversation"); + + let listed_a = a + .list_conversations_by_org(ORG_A) + .await + .expect("list org A"); + let listed_b = b + .list_conversations_by_org(ORG_B) + .await + .expect("list org B"); + + assert!( + listed_a.iter().any(|c| c.id == conv_a), + "org A must see its own conversation" + ); + assert!( + !listed_a.iter().any(|c| c.id == conv_b), + "CROSS-TENANT LEAK: org B's conversation {conv_b} appeared in org A's listing" + ); + assert!( + !listed_b.iter().any(|c| c.id == conv_a), + "CROSS-TENANT LEAK: org A's conversation {conv_a} appeared in org B's listing" + ); + assert!( + listed_a.iter().all(|c| c.organization_id == ORG_A), + "org A's listing contained a foreign organization_id" + ); + + // ---- 2. idempotency keys do not collide across tenants ----------------- + // Org A claims the key first; org B then creates with the SAME key. An + // org-blind idempotency claim would return org A's row to org B — handing a + // whole conversation (and every message on it) to the wrong tenant. + let claimed_a = a + .create_conversation(conversation( + &format!("idem-a-{suffix}"), + ORG_A, + SHARED_IDEMPOTENCY_KEY, + )) + .await + .expect("org A claims the shared idempotency key"); + let claimed_b = b + .create_conversation(conversation( + &format!("idem-b-{suffix}"), + ORG_B, + SHARED_IDEMPOTENCY_KEY, + )) + .await + .expect("org B claims the same idempotency key"); + assert_ne!( + claimed_a.id, claimed_b.id, + "CROSS-TENANT LEAK: the idempotency claim is not org-scoped — org B was handed org A's conversation" + ); + assert_eq!(claimed_b.organization_id, ORG_B); + + // ---- 3. participants + messages ride their own org's conversation ------ + a.add_participant(owner(&format!("owner-a-{suffix}"), &conv_a, ORG_A)) + .await + .expect("org A owner"); + b.add_participant(owner(&format!("owner-b-{suffix}"), &conv_b, ORG_B)) + .await + .expect("org B owner"); + + a.append_message(message( + &format!("msg-a-{suffix}"), + &conv_a, + ORG_A, + "org A private message", + )) + .await + .expect("org A message"); + b.append_message(message( + &format!("msg-b-{suffix}"), + &conv_b, + ORG_B, + "org B private message", + )) + .await + .expect("org B message"); + + // The SAME user email exists in both orgs. The per-user listing must still + // be org-partitioned — org is the outer boundary, ownership the inner one. + let owned_a = a + .list_conversations_by_org_and_user(ORG_A, SHARED_EMAIL) + .await + .expect("org A owned listing"); + assert!( + owned_a.iter().any(|c| c.id == conv_a), + "the shared user must see their own org-A conversation" + ); + assert!( + !owned_a.iter().any(|c| c.id == conv_b), + "CROSS-TENANT LEAK: the shared user saw org B's conversation while scoped to org A" + ); + + let page_a = a + .list_messages_by_conversation(MessageQuery::new(&conv_a, 50)) + .await + .expect("org A messages"); + assert!( + page_a + .messages + .iter() + .all(|m| m.organization_id.as_deref() == Some(ORG_A)), + "CROSS-TENANT LEAK: org A's message page carried a foreign org's message" + ); + + // ---- 4. sessions ------------------------------------------------------ + let sess_a = format!("sess-a-{suffix}"); + let sess_b = format!("sess-b-{suffix}"); + a.create_session(session(&sess_a, &conv_a, ORG_A)) + .await + .expect("org A session"); + b.create_session(session(&sess_b, &conv_b, ORG_B)) + .await + .expect("org B session"); + + let sessions_a = a + .list_sessions_by_conversation(&conv_a) + .await + .expect("org A sessions"); + assert_eq!(sessions_a.len(), 1, "org A's conversation has one session"); + assert_eq!(sessions_a[0].organization_id, ORG_A); + assert!( + !sessions_a.iter().any(|s| s.session_id == sess_b), + "CROSS-TENANT LEAK: org B's session listed under org A's conversation" + ); + + // Updating org B's session must not disturb org A's. + b.update_session( + &sess_b, + SessionUpdate { + token_count: Some(4242), + ..Default::default() + }, + ) + .await + .expect("update org B session"); + let after = a + .list_sessions_by_conversation(&conv_a) + .await + .expect("org A sessions after org B update"); + assert_eq!( + after[0].token_count, + Some(0), + "org B's session update bled into org A's session" + ); + + // ---- 5. knowledge ----------------------------------------------------- + // Each org ingests through ITS OWN access-bound handle — the seam the + // ingestion path uses (`knowledge_for_access`), which is what stamps the + // owning org on the stored row. + let doc_a = format!("doc-a-{suffix}"); + let doc_b = format!("doc-b-{suffix}"); + let marker_a = format!("alphamarker{suffix}"); + let marker_b = format!("betamarker{suffix}"); + + let access_a = AccessContext::default().with_organization_id(ORG_A); + let access_b = AccessContext::default().with_organization_id(ORG_B); + + a.knowledge_for_access(&access_a) + .ingest(org_document(&doc_a, ORG_A, &marker_a)) + .expect("ingest org A doc"); + b.knowledge_for_access(&access_b) + .ingest(org_document(&doc_b, ORG_B, &marker_b)) + .expect("ingest org B doc"); + + // Positive control: org A finds its own document. Without this, a backend + // that returns NOTHING would pass the isolation assertion vacuously. + let hits_a = a + .knowledge_for_access(&access_a) + .query(&format!("{marker_a} escalation code"), 10) + .expect("org A retrieval"); + assert!( + hits_a.iter().any(|r| r.chunk.contains(&marker_a)), + "org A could not retrieve its OWN document — the isolation assertion below would be vacuous" + ); + + // The leak assertion: org B's retrieval, using ORG A's distinctive query, + // must not surface org A's document. + let hits_b = b + .knowledge_for_access(&access_b) + .query(&format!("{marker_a} escalation code"), 10) + .expect("org B retrieval"); + assert!( + !hits_b.iter().any(|r| r.chunk.contains(&marker_a)), + "CROSS-TENANT LEAK: org B's knowledge retrieval returned org A's document: {hits_b:?}" + ); + assert!( + !hits_b.iter().any(|r| r.document_id == doc_a), + "CROSS-TENANT LEAK: org B's knowledge retrieval returned org A's document id" + ); + + // ---- 5b. ingest through the ORG-BLIND `knowledge()` handle ------------- + // `knowledge()` takes no org, and the admin connector-index path used it for + // every tenant. The document's own `org_id` metadata (which the ingestion + // pipeline stamps on every chunk) must therefore be enough to place the row + // in the right tenant — otherwise a connector run either lands in whichever + // org the handle happened to be built for (DynamoDB) or in no org at all + // (Postgres wrote `organization_id = NULL`, which the org-filtered read can + // never match, so retrieval silently returned nothing). + let blind_doc = format!("doc-blind-{suffix}"); + let blind_marker = format!("blindmarker{suffix}"); + a.knowledge() + .ingest(org_document(&blind_doc, ORG_A, &blind_marker)) + .expect("org-blind ingest of an org-A-stamped document"); + + let blind_hits_a = a + .knowledge_for_access(&access_a) + .query(&format!("{blind_marker} escalation code"), 10) + .expect("org A retrieval of the blind-ingested doc"); + assert!( + blind_hits_a.iter().any(|r| r.chunk.contains(&blind_marker)), + "a document ingested through the org-blind handle but stamped with org A's \ + `{ORG_METADATA_KEY}` must be retrievable by org A — it was lost instead" + ); + let blind_hits_b = b + .knowledge_for_access(&access_b) + .query(&format!("{blind_marker} escalation code"), 10) + .expect("org B retrieval of the blind-ingested doc"); + assert!( + !blind_hits_b.iter().any(|r| r.chunk.contains(&blind_marker)), + "CROSS-TENANT LEAK: a doc ingested through the org-blind handle reached org B: {blind_hits_b:?}" + ); + + // ---- 6. checkpoints --------------------------------------------------- + // `CheckpointStore` has no org dimension — it is keyed by agent id, and the + // server mints a fresh per-turn agent id. Assert the key boundary holds: + // one agent's checkpoint is invisible under another's id. + let engine_conv = EngineConversation::new(100_000).with_system_prompt("tenant A only"); + let agent_a = format!("agent-a-{suffix}"); + let agent_b = format!("agent-b-{suffix}"); + a.checkpoints() + .save(&Checkpoint::new(&agent_a, &engine_conv, 1)) + .expect("save org A checkpoint"); + + assert!( + b.checkpoints() + .load_latest(&agent_b) + .expect("load org B checkpoint") + .is_none(), + "CROSS-TENANT LEAK: org B loaded a checkpoint it never wrote" + ); + assert!( + b.checkpoints() + .list(&agent_b) + .expect("list org B checkpoints") + .is_empty(), + "CROSS-TENANT LEAK: org A's checkpoint listed under org B's agent id" + ); + assert_eq!( + a.checkpoints() + .list(&agent_a) + .expect("list org A checkpoints") + .len(), + 1, + "org A must still see its own checkpoint" + ); +} diff --git a/rust/adapters/postgres/src/knowledge.rs b/rust/adapters/postgres/src/knowledge.rs index f897f7f2..18f1008f 100644 --- a/rust/adapters/postgres/src/knowledge.rs +++ b/rust/adapters/postgres/src/knowledge.rs @@ -131,6 +131,17 @@ impl PgKnowledgeBase { // Stable per-chunk id: the document is stored as a single chunk keyed by // its document id, so re-ingesting the same doc upserts in place. let row_id = doc.id.clone(); + // Which tenant owns this row. The document's own `org_id` metadata (the + // ingestion pipeline stamps it on every chunk) wins over the handle's + // org, so a multi-tenant caller holding the org-blind `knowledge()` + // handle still lands the row in the right tenant instead of writing + // `organization_id = NULL` — which the org-filtered read + // (`WHERE organization_id = $1`) can then never match. + let ingest_org = doc + .metadata + .get(smooth_operator::access_control::ORG_METADATA_KEY) + .cloned() + .or_else(|| self.organization_id.clone()); let client = self.pool.get().await?; client @@ -149,7 +160,7 @@ impl PgKnowledgeBase { &[ &row_id, &doc.id, - &self.organization_id, + &ingest_org, &doc.source, &doc.content, &literal, diff --git a/rust/adapters/postgres/tests/multitenancy.rs b/rust/adapters/postgres/tests/multitenancy.rs new file mode 100644 index 00000000..df0e9304 --- /dev/null +++ b/rust/adapters/postgres/tests/multitenancy.rs @@ -0,0 +1,58 @@ +//! Multi-tenancy conformance (feature gap G7) for the Postgres + pgvector +//! `StorageAdapter`, against a real pgvector container via testcontainers. +//! +//! ONE adapter instance serving TWO orgs — the multi-tenant pod shape. The +//! knowledge slice is per-turn tenanted through `knowledge_for_access`, which is +//! what `PgKnowledgeBase::with_access` overrides the org from. +//! +//! The suite body is shared with the in-memory and DynamoDB adapters — see +//! `rust/adapters/multitenancy_suite.rs`. +//! +//! Skips (does not fail) when Docker is unavailable, matching +//! `tests/conformance.rs`. + +use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::runners::AsyncRunner; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +use smooth_operator_adapter_postgres::PostgresAdapter; + +#[path = "../../multitenancy_suite.rs"] +mod suite; + +/// Spin up a throwaway `pgvector/pgvector:pg16` container. `Ok(None)` ⇒ Docker +/// unavailable ⇒ skip. +async fn start_pgvector() -> anyhow::Result, String)>> { + let image = GenericImage::new("pgvector/pgvector", "pg16") + .with_wait_for(WaitFor::message_on_stderr( + "database system is ready to accept connections", + )) + .with_exposed_port(5432.tcp()) + .with_env_var("POSTGRES_USER", "postgres") + .with_env_var("POSTGRES_PASSWORD", "postgres") + .with_env_var("POSTGRES_DB", "postgres"); + + match image.start().await { + Ok(node) => { + let host = node.get_host().await?; + let port = node.get_host_port_ipv4(5432).await?; + let conn_str = + format!("host={host} port={port} user=postgres password=postgres dbname=postgres"); + Ok(Some((node, conn_str))) + } + Err(e) => { + eprintln!("SKIP: could not start pgvector container (Docker unavailable?): {e}"); + Ok(None) + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn two_orgs_are_isolated_on_one_postgres_adapter() -> anyhow::Result<()> { + let Some((_node, conn_str)) = start_pgvector().await? else { + return Ok(()); // Docker unavailable — skip, don't fail. + }; + let store = PostgresAdapter::connect(&conn_str).await?; + suite::assert_multitenancy(&store, &store, "pg").await; + Ok(()) +} diff --git a/rust/smooth-operator-lambda/src/dispatch.rs b/rust/smooth-operator-lambda/src/dispatch.rs index 06589d46..376db9eb 100644 --- a/rust/smooth-operator-lambda/src/dispatch.rs +++ b/rust/smooth-operator-lambda/src/dispatch.rs @@ -84,7 +84,7 @@ pub async fn handle_frame( create_session(storage, config, auth, poster, &parsed, request_id).await?; } Some("get_session") => { - get_session(storage, poster, &parsed, request_id).await?; + get_session(storage, auth, poster, &parsed, request_id).await?; } Some("send_message") => { send_message(storage, config, auth, poster, &parsed, request_id).await?; @@ -316,6 +316,7 @@ async fn create_session( /// `get_session` — read the session snapshot straight from DynamoDB. async fn get_session( storage: &Arc, + auth: &Arc, poster: &ConnectionPoster, parsed: &Value, request_id: Option<&str>, @@ -332,6 +333,17 @@ async fn get_session( }; match storage.get_session(session_id).await { + Ok(Some(s)) if !same_org(frame_org(auth, parsed).as_deref(), &s.organization_id) => { + // Another tenant's session — answer exactly as for an unknown id so + // the caller cannot distinguish "not yours" from "never existed". + poster + .post(&protocol::error( + request_id, + "SESSION_NOT_FOUND", + &format!("session '{session_id}' not found"), + )) + .await?; + } Ok(Some(s)) => { let data = json!({ "sessionId": s.session_id, @@ -395,6 +407,25 @@ async fn get_session( /// that fails to verify — so the caller falls back to the configured org and the /// no-auth/dev behavior is unchanged. Verification failures are logged (never the /// token). +/// The org a frame is authenticated to, when it carries a verifying token. +/// `None` for no token / an unconfigured verifier / a token that fails to +/// verify — which keeps the no-auth and dev paths unchanged. +fn frame_org(auth: &Arc, parsed: &Value) -> Option { + resolve_frame_principal(auth, parsed).map(|p| p.org_id) +} + +/// Whether a frame authenticated to `auth_org` may touch a row owned by +/// `row_org`. `None` (unauthenticated) keeps the pre-existing behavior; an +/// authenticated frame is pinned to its principal's tenant. +/// +/// Without this the lambda transport had **no** check at all on `sessionId`: +/// `get_session` and `send_message` acted on whatever `get_session` returned, so +/// any caller who knew or guessed a session id could read another org's session +/// snapshot and drive turns in its conversation. Feature gap G7. +fn same_org(auth_org: Option<&str>, row_org: &str) -> bool { + auth_org.is_none_or(|o| o == row_org) +} + fn resolve_frame_principal( auth: &Arc, parsed: &Value, @@ -485,6 +516,17 @@ async fn send_message( }; let session = match storage.get_session(session_id).await { + Ok(Some(s)) if !same_org(frame_org(auth, parsed).as_deref(), &s.organization_id) => { + // Another tenant's session — indistinguishable from an unknown id. + poster + .post(&protocol::error( + Some(request_id), + "SESSION_NOT_FOUND", + &format!("session '{session_id}' not found"), + )) + .await?; + return Ok(()); + } Ok(Some(s)) => s, Ok(None) => { poster diff --git a/rust/smooth-operator-server/src/admin.rs b/rust/smooth-operator-server/src/admin.rs index f0f5a24e..8164d53a 100644 --- a/rust/smooth-operator-server/src/admin.rs +++ b/rust/smooth-operator-server/src/admin.rs @@ -792,7 +792,17 @@ async fn index_connector( // created with this same embedder's dim by the storage-backend wiring // (`build_state_from_env_async`), so document and query vectors agree. let embedder = build_embedder(&EmbedderConfig::from_server_config(&state.config)); - let knowledge = state.storage.knowledge(); + // Ingest through the ORG-BOUND knowledge seam, never the raw `knowledge()` + // handle. `knowledge()` performs no org scoping (see the trait docs), so on a + // multi-tenant backend every org's connector documents landed in the same + // unscoped bucket — Postgres wrote `organization_id = NULL`, which the + // org-filtered retrieval path (`WHERE organization_id = $1`) can never match, + // and DynamoDB wrote the adapter's construction-time partition. Same shape as + // the G3 finding: the seam existed and one caller went around it. + let knowledge = state.storage.knowledge_for_access( + &smooth_operator::access_control::AccessContext::default() + .with_organization_id(principal.org_id.clone()), + ); let run = service .run_once( diff --git a/rust/smooth-operator-server/src/handler.rs b/rust/smooth-operator-server/src/handler.rs index dd5f1321..d3b71136 100644 --- a/rust/smooth-operator-server/src/handler.rs +++ b/rust/smooth-operator-server/src/handler.rs @@ -72,9 +72,32 @@ impl UserScope { } } +/// Whether a connection authenticated to `auth_org` may touch a row owned by +/// `row_org`. +/// +/// `auth_org == None` — an anonymous / tokenless connection, the embeddable +/// widget's normal state — keeps the pre-existing behavior: there is no org to +/// compare against, and the widget must still reach the session it just created. +/// A connection that DID present a verified principal, however, is pinned to +/// that principal's org: it may not read, drive, or mutate another tenant's +/// conversation even if it learns the id. Before this, org was resolved only to +/// *stamp* new sessions — every by-id path (`get_session`, `send_message`, +/// `confirm_tool_action`, `submit_interaction`, `verify_otp`, +/// `rename_conversation`, and conversation resume) was checked for per-user +/// ownership and never for tenant, so an ownerless conversation (the widget +/// default — no `user` participant carrying an email) was reachable by any +/// authenticated user in any org. Feature gap G7. +fn same_org(auth_org: Option<&str>, row_org: &str) -> bool { + auth_org.is_none_or(|o| o == row_org) +} + /// Whether this connection may read `conversation_id`. /// -/// `Unscoped` sees everything (auth not configured). Otherwise the conversation +/// Two boundaries, outer first. **Tenant**: a connection carrying a verified org +/// may only reach that org's conversations — see [`same_org`] — and that applies +/// to `Unscoped` too, since a single-user flavor still must not reach another +/// tenant. **Ownership**: `Unscoped` (auth not configured) then sees +/// everything. Otherwise the conversation /// is owner-checked **only if it has an owner** — a `user` participant carrying /// a non-blank email. A conversation with no such participant (created by an /// anonymous or emailless principal, or predating ownership) stays readable by @@ -85,8 +108,15 @@ impl UserScope { /// /// A storage error is a denial — an owner check that can't be completed must not /// pass. -async fn may_read_conversation(state: &AppState, conversation_id: &str, scope: &UserScope) -> bool { - if matches!(scope, UserScope::Unscoped) { +async fn may_read_conversation( + state: &AppState, + conversation_id: &str, + auth_org: Option<&str>, + scope: &UserScope, +) -> bool { + // Nothing to check on either axis — skip the participant read entirely, as + // this function did for `Unscoped` before the tenant check existed. + if auth_org.is_none() && matches!(scope, UserScope::Unscoped) { return true; } match state @@ -95,6 +125,21 @@ async fn may_read_conversation(state: &AppState, conversation_id: &str, scope: & .await { Ok(participants) => { + // Tenant boundary first — it outranks the per-user one, and applies + // even to `Unscoped` (a single-user flavor still must not reach + // another tenant). The conversation's org is read off its + // participants, which all carry it, so this costs no extra query. + // A conversation with no participants yet (the create→first-frame + // race) has no derivable org and is left to the ownership check + // below, exactly as before. + if let Some(row_org) = participants.first().map(|p| p.organization_id.as_str()) { + if !same_org(auth_org, row_org) { + return false; + } + } + if matches!(scope, UserScope::Unscoped) { + return true; + } let owned = participants.iter().any(|p| { p.participant_type == smooth_operator::domain::ParticipantType::User && p.email.as_deref().is_some_and(|e| !e.trim().is_empty()) @@ -129,12 +174,22 @@ async fn may_read_conversation(state: &AppState, conversation_id: &str, scope: & /// each used to load a session by raw id, so any authenticated user who knew or /// guessed another user's session id could drive a turn in it (and read the /// replayed history back through their own stream). th-1b7ed0. -async fn scoped_session(state: &AppState, session_id: &str, scope: &UserScope) -> Option { +async fn scoped_session( + state: &AppState, + session_id: &str, + auth_org: Option<&str>, + scope: &UserScope, +) -> Option { // th-ca579c: hydrate from storage on a local miss. `get_session` here would // report "not found" for a session this pod simply has not seen — which with // 2+ replicas is most returning visitors. let session = state.load_session(session_id).await?; - may_read_conversation(state, &session.conversation_id, scope) + // The session carries its own org, so the tenant check needs no extra read + // and covers every session-id-taking handler at this one chokepoint. + if !same_org(auth_org, &session.organization_id) { + return None; + } + may_read_conversation(state, &session.conversation_id, auth_org, scope) .await .then_some(session) } @@ -192,11 +247,12 @@ pub async fn handle_frame( None } Some("get_session") => { - handle_get_session(state, scope, &parsed, request_id, sink).await; + handle_get_session(state, auth_org, scope, &parsed, request_id, sink).await; None } Some("get_conversation_messages") => { - handle_get_conversation_messages(state, scope, &parsed, request_id, sink).await; + handle_get_conversation_messages(state, auth_org, scope, &parsed, request_id, sink) + .await; None } Some("list_conversations") => { @@ -204,24 +260,24 @@ pub async fn handle_frame( None } Some("rename_conversation") => { - handle_rename_conversation(state, scope, &parsed, request_id, sink).await; + handle_rename_conversation(state, auth_org, scope, &parsed, request_id, sink).await; None } // The only action that spawns a turn — its handle flows back to the reader // loop so a later `cancel` (or a disconnect) can abort it. Some("send_message") => { - handle_send_message(state, access, scope, &parsed, request_id, sink).await + handle_send_message(state, auth_org, access, scope, &parsed, request_id, sink).await } Some("confirm_tool_action") => { - handle_confirm_tool_action(state, scope, &parsed, request_id, sink).await; + handle_confirm_tool_action(state, auth_org, scope, &parsed, request_id, sink).await; None } Some("verify_otp") => { - handle_verify_otp(state, scope, &parsed, request_id, sink).await; + handle_verify_otp(state, auth_org, scope, &parsed, request_id, sink).await; None } Some("submit_interaction") => { - handle_submit_interaction(state, scope, &parsed, request_id, sink).await; + handle_submit_interaction(state, auth_org, scope, &parsed, request_id, sink).await; None } Some(other) => { @@ -408,7 +464,9 @@ async fn handle_create_session( // conversation ids are real, letting a caller enumerate other users' // conversations by their ids alone. let resume = match parsed.get("conversationId").and_then(Value::as_str) { - Some(cid) if !cid.is_empty() && may_read_conversation(state, cid, scope).await => { + Some(cid) + if !cid.is_empty() && may_read_conversation(state, cid, auth_org, scope).await => + { state.storage.get_conversation(cid).await.ok().flatten() } _ => None, @@ -661,6 +719,7 @@ async fn handle_create_session( /// `get_session` — return the session snapshot (per `get-session.schema.json`). async fn handle_get_session( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -675,7 +734,7 @@ async fn handle_get_session( return; }; - match scoped_session(state, session_id, scope).await { + match scoped_session(state, session_id, auth_org, scope).await { Some(s) => { let data = json!({ "sessionId": s.session_id, @@ -714,6 +773,7 @@ async fn handle_get_session( /// page's `nextCursor`. Newest-first (the common "recent history" read). async fn handle_get_conversation_messages( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -732,7 +792,7 @@ async fn handle_get_conversation_messages( // same message, same shape. A distinct "forbidden" would be an existence // oracle: it would tell a caller which session ids are real, which is all // an attacker needs to enumerate other users' conversations. - let Some(session) = scoped_session(state, session_id, scope).await else { + let Some(session) = scoped_session(state, session_id, auth_org, scope).await else { let _ = sink.send(protocol::error( request_id, "SESSION_NOT_FOUND", @@ -838,7 +898,7 @@ async fn handle_list_conversations( const MSG_CAP: usize = 200; let mut rows: Vec<(i64, Value)> = Vec::new(); for conv in conversations { - if !may_read_conversation(state, &conv.id, scope).await { + if !may_read_conversation(state, &conv.id, auth_org, scope).await { continue; } let mut query = smooth_operator::adapter::MessageQuery::new(&conv.id, MSG_CAP); @@ -933,6 +993,7 @@ const TITLE_MAX: usize = 60; /// (200) carrying `{ conversationId, title }`. async fn handle_rename_conversation( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -967,7 +1028,7 @@ async fn handle_rename_conversation( // retitle another user's conversation. Not-ours is reported exactly as // never-existed. match state.storage.get_conversation(conversation_id).await { - Ok(Some(_)) if may_read_conversation(state, conversation_id, scope).await => {} + Ok(Some(_)) if may_read_conversation(state, conversation_id, auth_org, scope).await => {} Ok(_) => { let _ = sink.send(protocol::error( request_id, @@ -1324,6 +1385,7 @@ async fn persist_workflow_step( /// failure (an `error` event was already emitted) — no turn was spawned. async fn handle_send_message( state: &AppState, + auth_org: Option<&str>, access: &AccessContext, scope: &UserScope, parsed: &Value, @@ -1411,7 +1473,7 @@ async fn handle_send_message( // sending into another user's session would replay their history as context // and stream the reply back to the sender, so an unscoped write here is also // a read of their conversation. - let Some(session) = scoped_session(state, session_id, scope).await else { + let Some(session) = scoped_session(state, session_id, auth_org, scope).await else { let _ = sink.send(protocol::error( Some(request_id), "SESSION_NOT_FOUND", @@ -1913,6 +1975,7 @@ async fn handle_send_message( /// duplicate confirm a no-op (`NO_PENDING_CONFIRMATION`). async fn handle_confirm_tool_action( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -1941,7 +2004,9 @@ async fn handle_confirm_tool_action( // Approving a write parked in ANOTHER user's turn is the same class of hole // as writing into their session. A session we may not read is reported with // the identical event an id with no pending confirmation produces. - let owned = scoped_session(state, session_id, scope).await.is_some(); + let owned = scoped_session(state, session_id, auth_org, scope) + .await + .is_some(); let Some(responder) = owned.then(|| state.take_confirmation(session_id)).flatten() else { let _ = sink.send(protocol::error( request_id, @@ -2134,6 +2199,7 @@ fn attach_interaction_effect(state: &AppState, session_id: &str, kind: &str, val /// duplicate submit a no-op (`NO_PENDING_INTERACTION`). async fn handle_submit_interaction( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -2164,7 +2230,9 @@ async fn handle_submit_interaction( // read reports the identical event an id with no pending park produces (the // submitted values would otherwise land in another user's turn, and its // identity-attach effect on their session). - let owned = scoped_session(state, session_id, scope).await.is_some(); + let owned = scoped_session(state, session_id, auth_org, scope) + .await + .is_some(); let Some(pending) = owned .then(|| state.pending_interaction(session_id)) .flatten() @@ -2344,6 +2412,7 @@ fn resolve_interaction( /// (`NOT_FOUND`, 0 attempts). async fn handle_verify_otp( state: &AppState, + auth_org: Option<&str>, scope: &UserScope, parsed: &Value, request_id: Option<&str>, @@ -2380,7 +2449,10 @@ async fn handle_verify_otp( // The session must exist AND be ours (a code can't verify — or brute-force — // a session we don't track, nor one belonging to another user). - if scoped_session(state, session_id, scope).await.is_none() { + if scoped_session(state, session_id, auth_org, scope) + .await + .is_none() + { let _ = sink.send(protocol::error( Some(request_id), "SESSION_NOT_FOUND", diff --git a/rust/smooth-operator-server/tests/acl_trusted_mode.rs b/rust/smooth-operator-server/tests/acl_trusted_mode.rs index b79f8ffd..ec7ccc93 100644 --- a/rust/smooth-operator-server/tests/acl_trusted_mode.rs +++ b/rust/smooth-operator-server/tests/acl_trusted_mode.rs @@ -36,6 +36,12 @@ use smooth_operator_server::runner::{self, TurnRequest, TurnResult}; /// stamps for a private repo (`github:owner/repo`). const PRIVATE_GROUP: &str = "github:acme/secret"; +/// The org the forwarded identities below claim. Knowledge is seeded *as* this +/// org because a turn's `AccessContext` carries the principal's org, and the +/// knowledge reader now enforces the tenant boundary before the ACL (feature gap +/// G7) — so an unstamped document belongs to no tenant and is invisible to one. +const ORG: &str = "acme"; + fn mock_llm() -> LlmConfig { LlmConfig::openrouter("not-a-real-key").with_model("openai/gpt-4o") } @@ -65,7 +71,9 @@ fn resolve_access(verifier: &dyn AuthVerifier, forwarded: Option<&str>) -> Acces /// [`PRIVATE_GROUP`]. Mirrors `acl_chat_leak::seeded_storage`. fn seeded_storage() -> Arc { let storage = Arc::new(InMemoryStorageAdapter::new()); - let kb = storage.knowledge(); + // Ingest through the ORG-BOUND seam, exactly as `server::seed_knowledge` and + // the admin connector-index path do: it stamps the owning tenant on each doc. + let kb = storage.knowledge_for_access(&AccessContext::default().with_organization_id(ORG)); let mut public = Document::new( "The alpha office hours are open to the whole organization.", diff --git a/rust/smooth-operator-server/tests/multitenancy.rs b/rust/smooth-operator-server/tests/multitenancy.rs new file mode 100644 index 00000000..b6f2646b --- /dev/null +++ b/rust/smooth-operator-server/tests/multitenancy.rs @@ -0,0 +1,284 @@ +//! Cross-**tenant** session access on the live WS path (feature gap G7) — SECURITY. +//! +//! `user_scoping.rs` covers two users in the SAME org. This covers the boundary +//! outside that one: two ORGS on one pod. +//! +//! Before the fix, org was resolved per connection only to *stamp* newly created +//! sessions. Every by-id path — `get_session`, `get_conversation_messages`, +//! `send_message`, `confirm_tool_action`, `submit_interaction`, `verify_otp`, +//! `rename_conversation`, and conversation resume — went through +//! `may_read_conversation`, which checks the **owner email** and never the org. +//! Its documented ownerless-is-open rule (a conversation with no `user` +//! participant carrying an email stays readable, so anonymous principals keep +//! their own sessions) is exactly the widget's default state — so an attacker +//! authenticated to org B who learned an org-A session id could read that +//! session, replay its whole history through a turn, and retitle its +//! conversation. +//! +//! These tests drive the real `handler::handle_frame` from the attacker's side +//! and assert every one of those actions is **byte-identical to not-found**. + +use std::sync::Arc; +use std::time::Duration; + +use serde_json::{json, Value}; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver}; + +use smooth_operator::access_control::AccessContext; +use smooth_operator::adapter::StorageAdapter; +use smooth_operator_adapter_memory::InMemoryStorageAdapter; + +use smooth_operator_server::config::{ServerConfig, StorageBackend}; +use smooth_operator_server::handler::{self, UserScope}; +use smooth_operator_server::state::AppState; + +const ORG_A: &str = "org-alpha"; +const ORG_B: &str = "org-beta"; + +fn base_config() -> ServerConfig { + ServerConfig { + bind: "127.0.0.1".into(), + port: 0, + gateway_url: "https://example.invalid/v1".into(), + gateway_key: None, + model: "claude-haiku-4-5".into(), + seed_kb: false, + max_iterations: 4, + max_tokens: 128, + storage: StorageBackend::Memory, + widget_auth_strict: false, + confirm_tools: Vec::new(), + judge_model: "claude-haiku-4-5".to_string(), + } +} + +/// Drive one frame as a connection authenticated to `auth_org` with `scope`, +/// returning the first emitted event. +async fn drive(state: &AppState, auth_org: &str, scope: &UserScope, frame: &Value) -> Value { + let (tx, mut rx) = unbounded_channel::(); + handler::handle_frame( + state, + &AccessContext::default().with_organization_id(auth_org), + "conn-test", + None, + Some(auth_org), + scope, + &frame.to_string(), + &tx, + ) + .await; + recv(&mut rx).await +} + +async fn recv(rx: &mut UnboundedReceiver) -> Value { + tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .expect("an event should be emitted") + .expect("sink open") +} + +/// An **ownerless** session in `ORG_A` — the embeddable widget's default state +/// (no `user` participant carrying an email), which is precisely the case +/// `may_read_conversation` deliberately leaves open to every principal. That is +/// what made this a cross-tenant hole rather than a theoretical one. +async fn victim_session(state: &AppState, storage: &InMemoryStorageAdapter) -> (String, String) { + let ev = drive( + state, + ORG_A, + &UserScope::Unscoped, + &json!({ + "action": "create_conversation_session", + "requestId": "cs", + "agentId": uuid::Uuid::new_v4().to_string(), + }), + ) + .await; + assert_eq!(ev["type"], "immediate_response", "got: {ev}"); + let session_id: String = ev["data"]["sessionId"].as_str().expect("sessionId").into(); + let conversation_id: String = ev["data"]["conversationId"] + .as_str() + .expect("conversationId") + .into(); + + // create-session persists participants in a spawned task; wait for a settled + // world so the ownership check sees what production would. + for _ in 0..100 { + let participants = storage + .list_participants_by_conversation(&conversation_id) + .await + .expect("list participants"); + if participants.len() >= 2 && state.get_session(&session_id).is_some() { + assert!( + participants.iter().all(|p| p + .email + .as_deref() + .unwrap_or_default() + .trim() + .is_empty()), + "the victim conversation must be OWNERLESS for this test to exercise \ + the open branch of may_read_conversation" + ); + let conv = storage + .get_conversation(&conversation_id) + .await + .expect("get conversation") + .expect("exists"); + assert_eq!(conv.organization_id, ORG_A, "victim must be org A's"); + return (session_id, conversation_id); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("create-session never persisted"); +} + +async fn state_with_victim() -> (AppState, Arc, String, String) { + let storage = Arc::new(InMemoryStorageAdapter::new()); + let state = AppState::new(storage.clone(), base_config()); + let (session_id, conversation_id) = victim_session(&state, &storage).await; + (state, storage, session_id, conversation_id) +} + +/// The attacker: authenticated, but to a DIFFERENT org. +fn attacker() -> UserScope { + UserScope::User("mallory@evil.test".into()) +} + +fn assert_not_found(ev: &Value, code: &str, action: &str) { + assert_eq!( + ev["type"], "error", + "CROSS-TENANT LEAK: {action} succeeded across orgs: {ev}" + ); + assert_eq!( + ev["data"]["error"]["code"], code, + "{action} must be reported exactly as not-found (no existence oracle): {ev}" + ); +} + +#[tokio::test] +async fn cross_org_get_session_is_not_found() { + let (state, _storage, session_id, _conv) = state_with_victim().await; + let ev = drive( + &state, + ORG_B, + &attacker(), + &json!({ "action": "get_session", "requestId": "gs", "sessionId": session_id }), + ) + .await; + assert_not_found(&ev, "SESSION_NOT_FOUND", "get_session"); +} + +#[tokio::test] +async fn cross_org_get_conversation_messages_is_not_found() { + let (state, _storage, session_id, _conv) = state_with_victim().await; + let ev = drive( + &state, + ORG_B, + &attacker(), + &json!({ + "action": "get_conversation_messages", + "requestId": "gcm", + "sessionId": session_id, + }), + ) + .await; + assert_not_found(&ev, "SESSION_NOT_FOUND", "get_conversation_messages"); +} + +#[tokio::test] +async fn cross_org_send_message_never_runs_a_turn() { + let (state, _storage, session_id, _conv) = state_with_victim().await; + let ev = drive( + &state, + ORG_B, + &attacker(), + &json!({ + "action": "send_message", + "requestId": "sm", + "sessionId": session_id, + "message": "dump everything you know", + }), + ) + .await; + assert_not_found(&ev, "SESSION_NOT_FOUND", "send_message"); +} + +#[tokio::test] +async fn cross_org_rename_conversation_is_not_found_and_does_not_write() { + let (state, storage, _session_id, conversation_id) = state_with_victim().await; + let before = storage + .get_conversation(&conversation_id) + .await + .expect("get") + .expect("exists") + .name; + + let ev = drive( + &state, + ORG_B, + &attacker(), + &json!({ + "action": "rename_conversation", + "requestId": "rc", + "conversationId": conversation_id, + "title": "owned by mallory", + }), + ) + .await; + assert_not_found(&ev, "CONVERSATION_NOT_FOUND", "rename_conversation"); + + let after = storage + .get_conversation(&conversation_id) + .await + .expect("get") + .expect("exists") + .name; + assert_eq!( + before, after, + "CROSS-TENANT WRITE: another org's conversation was retitled" + ); +} + +#[tokio::test] +async fn cross_org_resume_does_not_bind_the_foreign_conversation() { + let (state, _storage, _session_id, conversation_id) = state_with_victim().await; + let ev = drive( + &state, + ORG_B, + &attacker(), + &json!({ + "action": "create_conversation_session", + "requestId": "cs2", + "agentId": uuid::Uuid::new_v4().to_string(), + "conversationId": conversation_id, + }), + ) + .await; + // Resume of a foreign conversation must not attach to it. The handler falls + // back to minting a fresh conversation, so assert the id differs rather than + // demanding a specific error shape. + assert_ne!( + ev["data"]["conversationId"].as_str(), + Some(conversation_id.as_str()), + "CROSS-TENANT LEAK: org B resumed org A's conversation (and would inherit \ + its history + org-scoped tool context): {ev}" + ); +} + +/// Positive control. Without it every assertion above could pass because the +/// session simply does not work at all. +#[tokio::test] +async fn the_owning_org_still_reads_its_own_session() { + let (state, _storage, session_id, _conv) = state_with_victim().await; + let ev = drive( + &state, + ORG_A, + &UserScope::Unscoped, + &json!({ "action": "get_session", "requestId": "gs", "sessionId": session_id }), + ) + .await; + assert_eq!( + ev["type"], "immediate_response", + "the owning org must still read its own session: {ev}" + ); + assert_eq!(ev["data"]["sessionId"], session_id); +} diff --git a/rust/smooth-operator-server/tests/scenario_parity.rs b/rust/smooth-operator-server/tests/scenario_parity.rs index 428586ff..30f52b69 100644 --- a/rust/smooth-operator-server/tests/scenario_parity.rs +++ b/rust/smooth-operator-server/tests/scenario_parity.rs @@ -198,7 +198,14 @@ fn build_state_with_knowledge(config: ServerConfig, docs: &[Document]) -> AppSta return build_state(config); } let storage = Arc::new(InMemoryStorageAdapter::new()); - let kb = storage.knowledge(); + // Seed through the ORG-BOUND seam (as `server::seed_knowledge` does), stamping + // the org a no-auth turn resolves to. The knowledge reader enforces the tenant + // boundary before the ACL (feature gap G7), so a doc belonging to no tenant is + // invisible to a turn that has one. + let kb = storage.knowledge_for_access( + &smooth_operator::access_control::AccessContext::default() + .with_organization_id(smooth_operator_server::server::SEED_ORG_ID), + ); for doc in docs { kb.ingest(doc.clone()).expect("ingest server.knowledge doc"); } diff --git a/rust/smooth-operator/src/access_control.rs b/rust/smooth-operator/src/access_control.rs index e90eb0e7..60056a72 100644 --- a/rust/smooth-operator/src/access_control.rs +++ b/rust/smooth-operator/src/access_control.rs @@ -29,6 +29,18 @@ //! Postgres, or DynamoDB knowledge base identically (the post-filter happens in //! our layer, after the backend's own org-scoped query). //! +//! ## Two boundaries, outer first +//! +//! The reader enforces the **tenant** boundary before the within-org ACL. A +//! requester whose [`AccessContext`] carries an `organization_id` sees only +//! documents the store recorded as that org's — recorded from the document's +//! [`ORG_METADATA_KEY`] metadata at ingest, or from the org the ingesting handle +//! was bound to. A requester with no org resolved (single-tenant / anonymous) +//! gets no tenant filter, exactly as before. This matters because the wrapped +//! inner store is not necessarily org-partitioned: it is for Postgres and +//! DynamoDB, and is NOT for the in-memory adapter or any adapter relying on the +//! `knowledge_for_access` trait default (feature gap G7). +//! //! ## No-ACL default semantics — **no-acl ⇒ org-public** //! //! A document ingested **without** an ACL (the legacy / existing-seed path) has @@ -182,11 +194,13 @@ impl DocAcl { /// to scope RAG to that tenant's documents — its /// [`StorageAdapter::knowledge_for_access`](crate::adapter::StorageAdapter::knowledge_for_access) /// reads `access.organization_id` to pick the right tenant before any -/// user/group filtering. So the org rides on the `AccessContext` purely to be -/// **available** to a host adapter; the built-in ACL path ignores it (org -/// isolation already happened upstream — every knowledge row carries an -/// `organizationId` the backend filters on). `None` ⇒ "no org resolved", which a -/// single-tenant adapter treats exactly as today. +/// user/group filtering. The built-in [`AclKnowledgeStore`] reader enforces it +/// too (feature gap G7): a requester with a resolved org sees only documents the +/// store recorded as that org's, so a backend whose own storage is *not* +/// org-partitioned (the in-memory adapter, and any third-party adapter using the +/// `knowledge_for_access` trait default) is tenant-isolated by construction +/// rather than by assumption. `None` ⇒ "no org resolved" ⇒ no tenant filter, +/// which a single-tenant adapter treats exactly as before. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct AccessContext { /// The requester's user id, if authenticated as a user. `None` for an @@ -269,14 +283,36 @@ impl AccessContext { } } -/// Side table mapping a stored `document_id` to its [`DocAcl`]. Shared (`Arc`) -/// between the ingest handle that populates it and every per-request reader that -/// consults it. Documents absent from the table are org-public (no-ACL default). -type AclTable = Arc>>; +/// The document-metadata key naming the organization a document belongs to. +/// Stamped on every chunk by the ingestion pipeline +/// (`smooai-smooth-operator-ingestion`), and read back here so a store shared by +/// more than one tenant can enforce the org boundary at retrieval. +pub const ORG_METADATA_KEY: &str = "org_id"; + +/// What the side table remembers about one stored document: which org owns it +/// (for the tenant boundary) and its within-org [`DocAcl`] (for the user/group +/// boundary). Either may be absent — see [`AclReader`] for how each absence is +/// resolved. +#[derive(Debug, Clone, Default)] +struct DocEntry { + /// The owning organization, from the document's + /// [`ORG_METADATA_KEY`] metadata at ingest, falling back to the org the + /// ingesting handle was bound to. `None` ⇒ no org was ever recorded. + org: Option, + /// The within-org allow-list. `None` ⇒ no ACL recorded ⇒ org-public. + acl: Option, +} + +/// Side table mapping a stored `document_id` to its recorded org + [`DocAcl`]. +/// Shared (`Arc`) between the ingest handle that populates it and every +/// per-request reader that consults it. Documents absent from the table have +/// neither an org nor an ACL recorded. +type AclTable = Arc>>; /// An ACL-aware knowledge store: wraps any inner -/// [`KnowledgeBase`](smooth_operator_core::KnowledgeBase) and records document ACLs -/// at ingest so retrieval can be filtered per requester. +/// [`KnowledgeBase`](smooth_operator_core::KnowledgeBase) and records each +/// document's owning org + ACL at ingest so retrieval can be filtered per +/// requester (tenant boundary first, then the within-org allow-list). /// /// Construction does **not** itself implement `KnowledgeBase` for reading, /// because reads must be bound to a requester. Instead: @@ -339,11 +375,41 @@ impl AclKnowledgeStore { .acls .write() .map_err(|e| anyhow::anyhow!("acl table lock poisoned: {e}"))?; - table.insert(document_id.into(), acl); + table.entry(document_id.into()).or_default().acl = Some(acl); Ok(()) } } +/// Record a document's org + ACL into the shared side table at ingest. +/// +/// `bound_org` is the org the ingesting handle is bound to (an [`AclReader`]'s +/// requester org); it is the fallback when the document carries no +/// [`ORG_METADATA_KEY`] of its own — mirroring the Postgres backend, whose +/// ingest stamps the access-bound org onto the row. +fn record_document(acls: &AclTable, doc: &Document, bound_org: Option<&str>) -> anyhow::Result<()> { + let org = doc + .metadata + .get(ORG_METADATA_KEY) + .map(String::as_str) + .or(bound_org) + .map(str::to_string); + let acl = DocAcl::from_metadata(&doc.metadata); + if org.is_none() && acl.is_none() { + return Ok(()); + } + let mut table = acls + .write() + .map_err(|e| anyhow::anyhow!("acl table lock poisoned: {e}"))?; + let entry = table.entry(doc.id.clone()).or_default(); + if org.is_some() { + entry.org = org; + } + if acl.is_some() { + entry.acl = acl; + } + Ok(()) +} + /// Records ACLs at ingest, forwarding documents to the inner backend. struct AclIngestHandle { inner: Arc, @@ -352,15 +418,11 @@ struct AclIngestHandle { impl KnowledgeBase for AclIngestHandle { fn ingest(&self, doc: Document) -> anyhow::Result<()> { - // Record the ACL (if the document carries one) keyed by document id, so - // a later query result with that document_id can be access-checked. - if let Some(acl) = DocAcl::from_metadata(&doc.metadata) { - let mut table = self - .acls - .write() - .map_err(|e| anyhow::anyhow!("acl table lock poisoned: {e}"))?; - table.insert(doc.id.clone(), acl); - } + // Record the owning org + the ACL (whichever the document carries) keyed + // by document id, so a later query result with that document_id can be + // access-checked. This handle is bound to no requester, so there is no + // fallback org — an unstamped document records none. + record_document(&self.acls, &doc, None)?; self.inner.ingest(doc) } @@ -380,15 +442,12 @@ struct AclReader { impl KnowledgeBase for AclReader { fn ingest(&self, doc: Document) -> anyhow::Result<()> { - // A reader can still ingest (recording ACLs), so the same handle is - // usable end to end in tests — but production ingest uses ingest_handle. - if let Some(acl) = DocAcl::from_metadata(&doc.metadata) { - let mut table = self - .acls - .write() - .map_err(|e| anyhow::anyhow!("acl table lock poisoned: {e}"))?; - table.insert(doc.id.clone(), acl); - } + // A reader can still ingest, so the same handle is usable end to end — + // and this is the seam the org-scoped ingest paths use (the reference + // server's knowledge seeding, the admin connector index run). A document + // that carries no org of its own inherits the requester's, exactly as + // the Postgres backend stamps its access-bound org onto the row. + record_document(&self.acls, &doc, self.ctx.organization_id.as_deref())?; self.inner.ingest(doc) } @@ -405,11 +464,28 @@ impl KnowledgeBase for AclReader { let mut out = Vec::with_capacity(limit.min(candidates.len())); for result in candidates { - // No recorded ACL ⇒ org-public (backward-compatible default). - let allowed = match table.get(&result.document_id) { - Some(acl) => self.ctx.can_access(acl), + let entry = table.get(&result.document_id); + // Tenant boundary first — it is the outer one. A requester with a + // resolved org sees ONLY documents recorded as that org's; a + // document with no recorded org is not known to belong to the + // tenant, so it is dropped rather than assumed shared. This + // deliberately mirrors the Postgres backend's SQL pre-filter + // (`WHERE organization_id = $1`, which NULL rows also fail), so the + // shared multi-tenancy conformance suite holds identically on every + // backend. A requester with NO org (single-tenant / anonymous) keeps + // the pre-existing unfiltered behavior. + let same_tenant = match &self.ctx.organization_id { + Some(requester_org) => entry + .and_then(|e| e.org.as_deref()) + .is_some_and(|doc_org| doc_org == requester_org), None => true, }; + // Then the within-org ACL. No recorded ACL ⇒ org-public. + let allowed = same_tenant + && match entry.and_then(|e| e.acl.as_ref()) { + Some(acl) => self.ctx.can_access(acl), + None => true, + }; if allowed { out.push(result); if out.len() == limit { diff --git a/rust/smooth-operator/src/adapter.rs b/rust/smooth-operator/src/adapter.rs index d7eb293e..05d77d19 100644 --- a/rust/smooth-operator/src/adapter.rs +++ b/rust/smooth-operator/src/adapter.rs @@ -231,16 +231,18 @@ pub trait StorageAdapter: Send + Sync { /// ## Default — **fail closed for ACL'd content** /// /// The default implementation wraps [`knowledge`](Self::knowledge) in an - /// [`AclKnowledgeStore`](crate::access_control::AclKnowledgeStore) reader. - /// Because that wrapper's ACL side table starts empty (the documents were - /// ingested through a different store instance), every document it sees is - /// treated as org-public — which is the *raw* `knowledge()` behavior and is - /// therefore **not** a regression, but also offers no within-org protection. - /// Backends that can persist + read back a document's ACL (the in-memory - /// adapter via a shared store; Postgres / DynamoDB via a stored ACL column) - /// **override** this method to enforce the ACL durably, so restricted docs - /// are dropped for unentitled requesters even across the ingest→serve - /// process boundary. + /// [`AclKnowledgeStore`](crate::access_control::AclKnowledgeStore) reader, + /// which enforces **both** boundaries from its side table: the tenant + /// boundary (the document's recorded org vs `access.organization_id` — + /// feature gap G7) and the within-org user/group ACL. Documents ingested + /// through a *different* store instance are absent from that side table, so + /// they are dropped for a requester carrying an org and treated as org-public + /// for one carrying none. + /// + /// Backends that can persist + read back a document's org and ACL (the + /// in-memory adapter via a shared store; Postgres / DynamoDB via a stored + /// column / partition key) **override** this method to enforce both durably, + /// so the filter survives the ingest→serve process boundary. fn knowledge_for_access(&self, access: &AccessContext) -> Arc { crate::access_control::AclKnowledgeStore::new(self.knowledge()).reader(access.clone()) }