fix(memory): validate session owner on manual add (#3807)

routes/memory_routes.py

@@ -105,6 +105,13 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
         if memory_manager.find_duplicates(text, user_mem):
             return {"ok": True, "count": len(user_mem), "message": "Memory already exists"}
 
+        if memory_data.session_id:
+            try:
+                session_obj = session_manager.get_session(memory_data.session_id)
+            except KeyError:
+                raise HTTPException(404, "Session not found")
+            _assert_session_owner(session_obj, user)
+
         new_entry = memory_manager.add_entry(text, memory_data.source, memory_data.category, owner=user)
         if memory_data.session_id:
             new_entry["session_id"] = memory_data.session_id
@@ -163,8 +170,17 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
 
             session_id = memory.get("session_id")
             if session_id and session_id in session_manager.sessions:
-                session = session_manager.get_session(session_id)
-                memory["session_name"] = session.name if session else f"Session {session_id[:6]}"
+                try:
+                    session = session_manager.get_session(session_id)
+                    if session:
+                        _assert_session_owner(session, user)
+                    memory["session_name"] = session.name if session else f"Session {session_id[:6]}"
+                except KeyError:
+                    memory["session_name"] = "Unknown"
+                except HTTPException as exc:
+                    if exc.status_code != 404:
+                        raise
+                    memory["session_name"] = "Unknown"
             else:
                 memory["session_name"] = "Unknown"
 

tests/test_memory_routes_session_owner.py

@@ -14,6 +14,7 @@ import pytest
 from fastapi import HTTPException
 
 import routes.memory_routes as mr
+from src.request_models import MemoryAddRequest
 
 
 def _route(router, path, method):
@@ -38,6 +39,13 @@ def _router(monkeypatch, caller):
     return mr.setup_memory_routes(mem, sm)
 
 
+def _request(user):
+    return SimpleNamespace(
+        state=SimpleNamespace(current_user=user),
+        app=SimpleNamespace(state=SimpleNamespace(auth_manager=None)),
+    )
+
+
 def test_extract_rejects_other_users_session(monkeypatch):
     router = _router(monkeypatch, caller="bob")
     extract = _route(router, "/api/memory/extract", "POST")
@@ -59,3 +67,61 @@ def test_owner_can_access_own_session(monkeypatch):
     gbs = _route(router, "/api/memory/by-session/{session_id}", "GET")
     out = gbs(request=None, session_id="alice-sess")
     assert out["session_name"] == "Secret project"
+
+
+def test_add_memory_rejects_other_users_session(monkeypatch):
+    memory_manager = MagicMock()
+    session_manager = MagicMock()
+    memory_vector = MagicMock(healthy=True)
+    router = mr.setup_memory_routes(
+        memory_manager=memory_manager,
+        session_manager=session_manager,
+        memory_vector=memory_vector,
+    )
+    add_memory = _route(router, "/api/memory/add", "POST")
+
+    memory_manager.load.return_value = []
+    memory_manager.find_duplicates.return_value = False
+    session_manager.get_session.return_value = SimpleNamespace(owner="bob", name="Bob session")
+
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(
+            add_memory(
+                request=_request("alice"),
+                memory_data=MemoryAddRequest(
+                    text="Alice note",
+                    category="fact",
+                    source="user",
+                    session_id="bob-session",
+                ),
+            )
+        )
+
+    assert exc.value.status_code == 404
+    assert exc.value.detail == "Session not found"
+    session_manager.get_session.assert_called_once_with("bob-session")
+    memory_manager.add_entry.assert_not_called()
+    memory_manager.save.assert_not_called()
+    memory_vector.add.assert_not_called()
+
+
+def test_timeline_does_not_expose_other_users_session_name():
+    memory_manager = MagicMock()
+    session_manager = MagicMock()
+    session_manager.sessions = {"bob-session": object()}
+    session_manager.get_session.return_value = SimpleNamespace(owner="bob", name="Bob roadmap")
+    memory_manager.load.return_value = [
+        {
+            "id": "m1",
+            "text": "Alice note",
+            "owner": "alice",
+            "session_id": "bob-session",
+            "timestamp": 1,
+        }
+    ]
+    router = mr.setup_memory_routes(memory_manager, session_manager)
+    timeline = _route(router, "/api/memory/timeline", "GET")
+
+    out = timeline(request=_request("alice"))
+
+    assert out["timeline"][0]["session_name"] == "Unknown"