security(memory): /verbatim-memory/search and session delete are scoped to the caller (#16701)

autobot-backend/api/verbatim_memory.py

@@ -14,11 +14,12 @@
 
 from typing import Any, Dict, List
 
-from fastapi import APIRouter, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 
 from auth_middleware import get_auth_middleware
 from autobot_shared.error_boundaries import with_error_handling
 from autobot_shared.logging_manager import get_logger
+from security.session_ownership import validate_session_ownership
 
 logger = get_logger(__name__)
 
@@ -48,8 +49,10 @@ async def verbatim_search(
 ) -> Dict[str, Any]:
     """Search verbatim conversation chunks.
 
-    Returns chunks ranked by cosine similarity.  When ``session_id`` is
-    provided, only chunks from that session are considered.
+    Returns chunks ranked by cosine similarity, scoped to the caller's own
+    chunks (#16701 -- this route has no admin bypass; an admin-wide search
+    would need its own explicit admin API). When ``session_id`` is provided,
+    results are additionally restricted to that session.
 
     Args:
         q: Free-text query.
@@ -60,13 +63,16 @@ async def verbatim_search(
         JSON object with ``results`` list, each item having
         ``id``, ``text``, ``score``, and ``metadata`` keys.
     """
-    _require_user(request)
+    user = _require_user(request)
 
+    from knowledge.search_filters import extract_user_context_from_request
     from memory.verbatim_store import get_verbatim_store
 
+    user_id, _, _ = extract_user_context_from_request(user)
     store = await get_verbatim_store()
     results: List[Dict[str, Any]] = await store.search(
         query=q,
+        user_id=user_id,
         session_filter=session_id,
         limit=limit,
     )
@@ -83,20 +89,21 @@ async def verbatim_search(
 async def delete_session_verbatim(
     session_id: str,
     request: Request,
+    ownership: Dict = Depends(validate_session_ownership),  # SECURITY: Validate ownership (#16701)
 ) -> Dict[str, Any]:
     """Delete all verbatim chunks for a session.
 
-    Used for user opt-out and retention enforcement.  The caller must be
-    authenticated; in production the middleware additionally enforces that
-    users can only delete their own sessions.
+    Used for user opt-out and retention enforcement. Caller must own the
+    session (#16701: this docstring previously claimed a production
+    middleware enforced that; nothing in this file did).
 
     Args:
         session_id: Session whose verbatim chunks to remove.
 
     Returns:
         JSON object with ``session_id`` and ``deleted_count``.
     """
-    _require_user(request)
+    del ownership  # dependency raises 401/403 before the handler runs
 
     from memory.verbatim_store import get_verbatim_store
 

autobot-backend/api/verbatim_memory_16701_test.py

@@ -0,0 +1,153 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""GET /verbatim-memory/search and DELETE /verbatim-memory/session/{id} are
+scoped to the caller (#16701). Search used to check only that the caller was
+signed in and never filtered by user; any signed-in user could search every
+user's verbatim conversation chunks. Delete's docstring claimed a production
+middleware enforced session ownership; nothing in this file did.
+
+Search's cross-user proof uses a real VerbatimStore over a mocked ChromaDB
+collection (not a mock asserting call args), so this proves the route's own
+scoping actually works end to end, not just that an argument was passed.
+Delete's proof mounts the real router and overrides validate_session_ownership
+the same way FastAPI would invoke it, so a refusal there is proven to happen
+before the handler body ever runs.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.testclient import TestClient
+
+from api.verbatim_memory import router as verbatim_router
+from memory.verbatim_store import VerbatimStore
+from security.session_ownership import validate_session_ownership
+
+_STORED = [
+    {"id": "a1", "text": "reset the system", "session_id": "s1", "role": "user", "user_id": "user-a"},
+    {"id": "b1", "text": "reset the router", "session_id": "s2", "role": "user", "user_id": "user-b"},
+]
+
+
+def _make_collection(stored):
+    """Minimal fake AsyncChromaCollection: $eq / $and-composite where support."""
+    collection = MagicMock()
+
+    def _conditions(where):
+        if not where:
+            return []
+        return where["$and"] if "$and" in where else [where]
+
+    async def _query(query_texts=None, n_results=10, where=None, include=None):
+        items = stored
+        for cond in _conditions(where):
+            for field, op in cond.items():
+                val = op.get("$eq")
+                if val is not None:
+                    items = [i for i in items if i.get(field) == val]
+        limit = min(n_results, len(items))
+        return {
+            "ids": [[i["id"] for i in items[:limit]]],
+            "documents": [[i["text"] for i in items[:limit]]],
+            "metadatas": [[{"user_id": i.get("user_id", "")} for i in items[:limit]]],
+            "distances": [[0.1 * j for j in range(limit)]],
+        }
+
+    collection.query = _query
+    return collection
+
+
+def _request_for(user_id: str) -> Request:
+    """A minimal Request; only used to satisfy verbatim_search's signature here."""
+    return MagicMock(spec=Request)
+
+
+def _fake_auth_middleware(user_id: str | None):
+    return SimpleNamespace(get_user_from_request=lambda _request: {"user_id": user_id} if user_id else None)
+
+
+@pytest.mark.asyncio
+async def test_search_never_returns_another_users_chunks():
+    from api.verbatim_memory import verbatim_search
+
+    store = VerbatimStore()
+    store._collection = _make_collection(_STORED)
+
+    with (
+        patch("api.verbatim_memory.get_auth_middleware", return_value=_fake_auth_middleware("user-a")),
+        patch("memory.verbatim_store.get_verbatim_store", new=AsyncMock(return_value=store)),
+    ):
+        result = await verbatim_search(_request_for("user-a"), q="reset", session_id=None, limit=10)
+
+    ids = {r["id"] for r in result["results"]}
+    assert ids == {"a1"}, f"user-a must not see user-b's chunk: {result['results']}"
+
+
+@pytest.mark.asyncio
+async def test_search_is_symmetric_for_the_other_user():
+    from api.verbatim_memory import verbatim_search
+
+    store = VerbatimStore()
+    store._collection = _make_collection(_STORED)
+
+    with (
+        patch("api.verbatim_memory.get_auth_middleware", return_value=_fake_auth_middleware("user-b")),
+        patch("memory.verbatim_store.get_verbatim_store", new=AsyncMock(return_value=store)),
+    ):
+        result = await verbatim_search(_request_for("user-b"), q="reset", session_id=None, limit=10)
+
+    ids = {r["id"] for r in result["results"]}
+    assert ids == {"b1"}, f"user-b must not see user-a's chunk: {result['results']}"
+
+
+@pytest.mark.asyncio
+async def test_search_requires_authentication():
+    from api.verbatim_memory import verbatim_search
+
+    with patch("api.verbatim_memory.get_auth_middleware", return_value=_fake_auth_middleware(None)):
+        with pytest.raises(HTTPException) as exc_info:
+            await verbatim_search(_request_for("anyone"), q="reset", session_id=None, limit=10)
+
+    assert exc_info.value.status_code == 401
+
+
+def _build_app() -> FastAPI:
+    app = FastAPI()
+    app.include_router(verbatim_router)
+    return app
+
+
+def test_delete_session_refuses_a_caller_who_does_not_own_it():
+    """#16701: the route must run validate_session_ownership, not just 'is signed in'."""
+
+    async def _refuse(session_id: str, request: Request):
+        raise HTTPException(status_code=403, detail="not your session")
+
+    app = _build_app()
+    app.dependency_overrides[validate_session_ownership] = _refuse
+
+    response = TestClient(app).delete("/verbatim-memory/session/s1")
+
+    assert response.status_code == 403
+
+
+def test_delete_session_allows_the_owner():
+    async def _allow(session_id: str, request: Request):
+        return {"user_id": "user-a"}
+
+    app = _build_app()
+    app.dependency_overrides[validate_session_ownership] = _allow
+
+    with patch("memory.verbatim_store.get_verbatim_store") as get_store:
+        store = AsyncMock()
+        store.delete_session = AsyncMock(return_value=2)
+        get_store.return_value = store
+
+        response = TestClient(app).delete("/verbatim-memory/session/s1")
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"session_id": "s1", "deleted_count": 2}

autobot-backend/mcp/autobot_server.py

@@ -832,11 +832,11 @@ async def _memory_path(
 
     async def _memory_verbatim_search(self, query: str, session_filter: str | None = None) -> Any:
         from api.schemas_knowledge import MemoryVerbatimSearchRequest
-        from memory.verbatim_store import VerbatimStore
+        from memory.verbatim_store import UNSCOPED_ALL_USERS, VerbatimStore
 
         args = MemoryVerbatimSearchRequest(query=query, session_filter=session_filter)
         store = VerbatimStore()
-        results = await store.search(args.query, session_filter=args.session_filter)
+        results = await store.search(args.query, user_id=UNSCOPED_ALL_USERS, session_filter=args.session_filter)
         return {"results": results, "count": len(results)}
 
     # ------------------------------------------------------------------

autobot-backend/memory/verbatim_recency_test.py

@@ -16,7 +16,7 @@
 import pytest
 
 from memory import verbatim_store
-from memory.verbatim_store import VerbatimStore, _recency_factor
+from memory.verbatim_store import UNSCOPED_ALL_USERS, VerbatimStore, _recency_factor
 
 
 def _make_collection(items: List[Dict[str, Any]]) -> MagicMock:
@@ -85,7 +85,7 @@ async def test_search_recent_beats_equally_similar_stale(monkeypatch):
             {"id": "fresh", "text": "new", "distance": 0.1, "timestamp": now.isoformat()},
         ]
     )
-    results = await store.search("q")
+    results = await store.search("q", user_id=UNSCOPED_ALL_USERS)
     assert [r["id"] for r in results] == ["fresh", "stale"]
     assert results[0]["score"] > results[1]["score"]
 
@@ -101,7 +101,7 @@ async def test_search_weight_zero_preserves_semantic_order(monkeypatch):
             {"id": "farther_new", "text": "b", "distance": 0.4, "timestamp": now.isoformat()},
         ]
     )
-    results = await store.search("q")
+    results = await store.search("q", user_id=UNSCOPED_ALL_USERS)
     # Weight 0 → pure semantic: the closer (lower distance) chunk stays first.
     assert [r["id"] for r in results] == ["closer_old", "farther_new"]
     assert results[0]["score"] == pytest.approx(0.9)

autobot-backend/memory/verbatim_store.py

@@ -35,6 +35,32 @@
 _DEFAULT_LIMIT = 10
 _DEFAULT_RETENTION_DAYS: int = 90  # override via memory.verbatim.retention_days config
 
+
+class _UnscopedAllUsersSentinel:
+    """Type of :data:`UNSCOPED_ALL_USERS` -- deliberately not ``str``.
+
+    #16732 review: the sentinel was originally the plain string
+    ``"__unscoped_all_users__"``, compared with ``==``. A caller's
+    ``user_id`` falls back to their username when a token carries no
+    ``user_id`` claim, and usernames are validated against
+    ``^[a-zA-Z0-9_]+$`` -- a string the old sentinel text satisfies. A user
+    actually named ``__unscoped_all_users__`` would have matched the
+    sentinel and searched every user's chunks. A non-``str`` singleton,
+    compared by identity (``is``), can never equal a string any caller
+    supplies, however it is spelled.
+    """
+
+    def __repr__(self) -> str:
+        return "UNSCOPED_ALL_USERS"
+
+
+#: Issue #16701: search()'s user_id is required precisely so a caller can't
+#: get every user's chunks by omitting an argument. The one legitimate
+#: unscoped caller (mcp/autobot_server.py's memory.verbatim_search -- the
+#: stdio MCP transport has no per-user identity at all, tracked as #16727)
+#: must say so explicitly by passing this sentinel, not None or "".
+UNSCOPED_ALL_USERS = _UnscopedAllUsersSentinel()
+
 # B1 (#12555): optional MemPalace-style symbolic "drawer" index. An inverted
 # term -> chunk_id index in Redis lets an entity/keyword query resolve candidate
 # chunks in one lookup and rank them lexically, skipping the ANN embed+search on
@@ -284,6 +310,7 @@ async def _deindex_symbolic(self, chunk_ids: List[str]) -> None:
     async def search_symbolic(
         self,
         query: str,
+        user_id: "str | _UnscopedAllUsersSentinel",
         session_filter: str | None = None,
         limit: int = _DEFAULT_LIMIT,
     ) -> List[Dict[str, Any]] | None:
@@ -293,6 +320,10 @@ async def search_symbolic(
         ranks by query-term overlap blended with recency. Returns ``None`` (not
         ``[]``) when the index is disabled, the query has no salient terms, or no
         candidate matched — so the caller can fall back to semantic ``search``.
+
+        Issue #16701: ``user_id`` is required for the same reason as ``search``'s
+        -- the term index has no per-user structure, so scoping happens when
+        candidates are fetched and ranked (:meth:`_rank_symbolic_candidates`).
         """
         if not _SYMBOLIC_INDEX_ENABLED:
             return None
@@ -319,10 +350,15 @@ async def search_symbolic(
                 _SYM_MAX_CANDIDATES,
             )
             return None
-        return await self._rank_symbolic_candidates(candidate_ids, terms, session_filter, limit)
+        return await self._rank_symbolic_candidates(candidate_ids, terms, user_id, session_filter, limit)
 
     async def _rank_symbolic_candidates(
-        self, candidate_ids: List[str], query_terms: set, session_filter: str | None, limit: int
+        self,
+        candidate_ids: List[str],
+        query_terms: set,
+        user_id: "str | _UnscopedAllUsersSentinel",
+        session_filter: str | None,
+        limit: int,
     ) -> List[Dict[str, Any]]:
         """Fetch candidate chunks and rank by term-overlap blended with recency."""
         collection = await self._get_collection()
@@ -334,6 +370,8 @@ async def _rank_symbolic_candidates(
         ranked: List[Dict[str, Any]] = []
         for cid, doc, meta in zip(ids, docs, metas):
             meta = meta or {}
+            if user_id is not UNSCOPED_ALL_USERS and meta.get("user_id") != user_id:
+                continue
             if session_filter and meta.get("session_id") != session_filter:
                 continue
             doc_terms = _extract_terms(doc)
@@ -352,19 +390,30 @@ async def _rank_symbolic_candidates(
     async def search(
         self,
         query: str,
+        user_id: "str | _UnscopedAllUsersSentinel",
         session_filter: str | None = None,
         limit: int = _DEFAULT_LIMIT,
     ) -> List[Dict[str, Any]]:
         """Search verbatim chunks via ChromaDB's built-in vector + BM25 path.
 
-        When ``session_filter`` is provided, results are restricted to that
-        session.  ChromaDB ``query_texts`` triggers its internal embedding +
-        ANN search; callers that need BM25 re-ranking should apply it on the
-        returned list.
+        When ``session_filter`` is provided, results are additionally
+        restricted to that session.  ChromaDB ``query_texts`` triggers its
+        internal embedding + ANN search; callers that need BM25 re-ranking
+        should apply it on the returned list.
+
+        Issue #16701: ``user_id`` is required, not optional -- results are
+        restricted to chunks stored with that ``metadata.user_id``, the same
+        ``where`` filter ``memory/transparency.py``'s ``_list_verbatim``
+        already applies, in the ChromaDB query itself, not a post-filter.
+        Requiring it (rather than defaulting to unscoped) means a caller
+        cannot get every user's chunks by simply omitting an argument; the
+        one legitimate unscoped caller passes :data:`UNSCOPED_ALL_USERS`
+        explicitly (see its docstring for why).
 
         Args:
             query: Free-text search query.
-            session_filter: Optional session_id to scope results.
+            user_id: The caller's user_id, or :data:`UNSCOPED_ALL_USERS`.
+            session_filter: Optional session_id to further scope results.
             limit: Maximum number of chunks to return.
 
         Returns:
@@ -375,9 +424,16 @@ async def search(
         if limit <= 0:
             raise ValueError("limit must be positive")
 
-        where: Dict[str, Any] | None = None
+        where_conditions: List[Dict[str, Any]] = []
         if session_filter:
-            where = {"session_id": {"$eq": session_filter}}
+            where_conditions.append({"session_id": {"$eq": session_filter}})
+        if user_id is not UNSCOPED_ALL_USERS:
+            where_conditions.append({"user_id": {"$eq": user_id}})
+        where: Dict[str, Any] | None = None
+        if len(where_conditions) == 1:
+            where = where_conditions[0]
+        elif len(where_conditions) > 1:
+            where = {"$and": where_conditions}
 
         collection = await self._get_collection()
         try: