G7: make multi-tenant isolation a test, and close the two leaks it found

.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.

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.

rust/adapters/dynamodb/src/knowledge.rs

@@ -53,8 +53,10 @@ pub struct DynamoKnowledgeBase {
     table: String,
     embedder: Arc<dyn Embedder>,
     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<F, T>(&self, fut: F) -> Result<T>
     where
         F: std::future::Future<Output = Result<T>> + 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",

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(())
+}

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;
+}