fix(security): hosted multi-tenancy hardening — session REST owner-scoping + TOCTOU path revalidation (#704) (#707)

* fix(security): owner-scope session REST endpoints + revalidate workspace path at WS connect (#704)

Hosted multi-tenancy hardening (follow-up to #655):

1. Owner-scope session REST endpoints. list/get/delete/messages queried
   without an owner filter, so an authed tenant could enumerate/read/modify/
   end another tenant's sessions. Thread require_auth through all five handlers;
   scope repo.list() by user_id and gate per-id handlers via _get_owned_session
   (404 on missing-or-not-yours so IDs can't be probed). No-auth mode
   (user_id=None) skips scoping, unchanged.

2. TOCTOU symlink revalidation. The stored workspace_path cleared the allowlist
   at create time but terminal_ws/session_chat_ws used the raw string later as
   cwd/tool scope. A tenant could swap a dir for a symlink outside its root in
   between. Re-resolve + re-check (revalidate_workspace_path) at WS connect;
   close on reject; use the freshly resolved path.

Known limitation: a sub-ms race remains between revalidation and shell spawn;
true TOCTOU-proof isolation needs per-tenant container/chroot or
openat2(RESOLVE_NO_SYMLINKS) — infra-level, deferred.

* refactor: address review feedback on #704

- Reword 'ponytail:' docstring label to 'Note:' in revalidate_workspace_path
- Hoist revalidate_workspace_path imports to module top (no circular import)
- Strengthen terminal symlink test to assert close code 4008

Author: frankbria

codeframe/platform_store/repositories/interactive_sessions.py

@@ -51,9 +51,15 @@ def list(
         workspace_path: Optional[str] = None,
         state: Optional[str] = None,
         limit: int = 50,
+        user_id: Optional[int] = None,
     ) -> list[dict]:
         query = "SELECT * FROM interactive_sessions WHERE 1=1"
         params: list = []
+        # Owner-scoping (#704): filter only when a user_id is supplied. None means
+        # no-auth mode (CODEFRAME_AUTH_REQUIRED=false) → return all, unchanged.
+        if user_id is not None:
+            query += " AND user_id = ?"
+            params.append(user_id)
         if workspace_path is not None:
             query += " AND workspace_path = ?"
             params.append(workspace_path)

codeframe/ui/dependencies.py

@@ -79,6 +79,26 @@ def enforce_workspace_allowlist(path: Path, user_id: Optional[int]) -> Path:
     return path
 
 
+def revalidate_workspace_path(workspace_path: str, user_id: Optional[int]) -> Optional[Path]:
+    """Re-check a stored session workspace path against the allowlist at use time (#704).
+
+    ``create_session`` validates the path once, but the terminal/chat WebSockets
+    open later — a tenant could swap a dir (or ancestor) for a symlink pointing
+    outside its allowed root in between (TOCTOU). ``enforce_workspace_allowlist``
+    calls ``.resolve()``, which follows symlinks, so a swapped-in escape is caught
+    here. Returns the freshly resolved path, or ``None`` if it no longer passes
+    (the WS caller closes the socket instead of raising HTTP).
+
+    Note: this closes the practical window; a sub-millisecond race remains between
+    this check and the shell spawn. True TOCTOU-proof isolation needs a per-tenant
+    container/chroot or openat2(RESOLVE_NO_SYMLINKS) — infra-level, deferred.
+    """
+    try:
+        return enforce_workspace_allowlist(Path(workspace_path), user_id)
+    except HTTPException:
+        return None
+
+
 def get_workspace_manager(request: Request) -> WorkspaceManager:
     """Get workspace manager from application state.
 

codeframe/ui/routers/interactive_sessions_v2.py

@@ -115,6 +115,22 @@ def _get_repo(request: Request):
     return db.interactive_sessions
 
 
+def _get_owned_session(repo, session_id: str, auth: Dict[str, Any]) -> dict:
+    """Fetch a session, enforcing owner-scoping (#704).
+
+    Returns the row, or raises 404 if it is missing OR owned by another user.
+    404 (not 403) so a tenant can't probe which session IDs exist. In no-auth
+    mode ``user_id`` is None → ownership is not enforced (matches REST/#655).
+    """
+    session = repo.get(session_id)
+    user_id = auth.get("user_id")
+    if session is None or (
+        user_id is not None and session.get("user_id") != user_id
+    ):
+        raise HTTPException(status_code=404, detail="Session not found")
+    return session
+
+
 def _session_to_response(row: dict) -> SessionResponse:
     """Convert a DB row dict to a SessionResponse."""
     return SessionResponse(
@@ -172,6 +188,7 @@ def list_sessions(
     workspace_path: Optional[str] = Query(None),
     state: Optional[str] = Query(None),
     limit: int = Query(50, ge=1, le=200),
+    auth: Dict[str, Any] = Depends(require_auth),
 ):
     """List sessions with optional filters by workspace_path and state."""
     if state is not None and state not in VALID_STATES:
@@ -180,39 +197,55 @@ def list_sessions(
             detail=f"Invalid state '{state}'. Must be one of {sorted(VALID_STATES)}",
         )
     repo = _get_repo(request)
-    sessions = repo.list(workspace_path=workspace_path, state=state, limit=limit)
+    sessions = repo.list(
+        workspace_path=workspace_path,
+        state=state,
+        limit=limit,
+        user_id=auth.get("user_id"),
+    )
     return [_session_to_response(s) for s in sessions]
 
 
 @router.get("/{session_id}", response_model=SessionResponse)
 @rate_limit_standard()
-def get_session(session_id: str, request: Request):
+def get_session(
+    session_id: str,
+    request: Request,
+    auth: Dict[str, Any] = Depends(require_auth),
+):
     """Get a session by ID."""
     repo = _get_repo(request)
-    session = repo.get(session_id)
-    if session is None:
-        raise HTTPException(status_code=404, detail="Session not found")
-    return _session_to_response(session)
+    return _session_to_response(_get_owned_session(repo, session_id, auth))
 
 
 @router.delete("/{session_id}", response_model=SessionResponse)
 @rate_limit_standard()
-def end_session(session_id: str, request: Request):
+def end_session(
+    session_id: str,
+    request: Request,
+    auth: Dict[str, Any] = Depends(require_auth),
+):
     """End a session (sets state to 'ended' and records ended_at)."""
     repo = _get_repo(request)
+    _get_owned_session(repo, session_id, auth)
     updated = repo.end(session_id)
     if updated is None:
+        # Vanished between the ownership check and end() (concurrent delete).
         raise HTTPException(status_code=404, detail="Session not found")
     return _session_to_response(updated)
 
 
 @router.post("/{session_id}/messages", status_code=201, response_model=MessageResponse)
 @rate_limit_standard()
-def add_message(session_id: str, body: MessageCreate, request: Request):
+def add_message(
+    session_id: str,
+    body: MessageCreate,
+    request: Request,
+    auth: Dict[str, Any] = Depends(require_auth),
+):
     """Add a message to a session's history."""
     repo = _get_repo(request)
-    if repo.get(session_id) is None:
-        raise HTTPException(status_code=404, detail="Session not found")
+    _get_owned_session(repo, session_id, auth)
     message = repo.add_message(
         session_id=session_id,
         role=body.role,
@@ -236,11 +269,11 @@ def get_messages(
     request: Request,
     limit: int = Query(100, ge=1, le=500),
     offset: int = Query(0, ge=0),
+    auth: Dict[str, Any] = Depends(require_auth),
 ):
     """Get paginated message history for a session."""
     repo = _get_repo(request)
-    if repo.get(session_id) is None:
-        raise HTTPException(status_code=404, detail="Session not found")
+    _get_owned_session(repo, session_id, auth)
     messages = repo.get_messages(session_id, limit=limit, offset=offset)
     return [
         MessageResponse(

codeframe/ui/routers/session_chat_ws.py

@@ -40,6 +40,7 @@
 
 from codeframe.auth.dependencies import authenticate_websocket
 from codeframe.core.adapters.streaming_chat import StreamingChatAdapter
+from codeframe.ui.dependencies import revalidate_workspace_path
 from codeframe.ui.shared import session_chat_manager
 
 logger = logging.getLogger(__name__)
@@ -136,6 +137,21 @@ async def session_chat_ws(session_id: str, websocket: WebSocket) -> None:
         await websocket.close(code=1008, reason="Forbidden: session belongs to another user")
         return
 
+    # --- Revalidate the stored path against the allowlist (TOCTOU, #704) ---
+    # The chat agent runs file-system tools scoped to this path. It cleared the
+    # allowlist at create time, but a tenant could have swapped a dir for a
+    # symlink pointing outside its root before connecting. Re-resolve and
+    # re-check now; use the freshly resolved path for the rest of the session.
+    raw_workspace = session.get("workspace_path")
+    validated_workspace: Optional[Path] = None
+    if raw_workspace:
+        validated_workspace = await asyncio.to_thread(
+            revalidate_workspace_path, raw_workspace, user_id
+        )
+        if validated_workspace is None:
+            await websocket.close(code=1008, reason="Workspace path no longer permitted")
+            return
+
     # --- Accept connection; everything after this point must run inside the
     #     try/finally so unregister() and close() always execute even if
     #     update_state, register, or get_token_queue raises. ---
@@ -282,14 +298,13 @@ async def _receive() -> None:
                             break
 
                     interrupt_event = await session_chat_manager.get_interrupt_event(session_id)
-                    raw_workspace = session.get("workspace_path")
-                    if not raw_workspace:
+                    if validated_workspace is None:
                         logger.warning(
                             "session_id=%s has no workspace_path set; "
                             "file tools will scope to server CWD",
                             session_id,
                         )
-                    workspace_path = Path(raw_workspace or ".")
+                    workspace_path = validated_workspace or Path(".")
                     adapter_task[0] = asyncio.create_task(
                         _run_streaming_adapter(
                             session_id,

codeframe/ui/routers/terminal_ws.py

@@ -25,6 +25,7 @@
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 
 from codeframe.auth.dependencies import authenticate_websocket
+from codeframe.ui.dependencies import revalidate_workspace_path
 
 logger = logging.getLogger(__name__)
 
@@ -92,6 +93,22 @@ async def session_terminal_ws(session_id: str, websocket: WebSocket) -> None:
         await websocket.close(code=4008, reason="Session has no workspace configured")
         return
 
+    # --- Revalidate the stored path against the allowlist (TOCTOU, #704) ---
+    # The path cleared the allowlist at create time, but a tenant could have
+    # swapped a dir for a symlink pointing outside its root before connecting.
+    # Re-resolve and re-check now; spawn with the freshly resolved path.
+    revalidated = await asyncio.to_thread(
+        revalidate_workspace_path, workspace_path, user_id
+    )
+    if revalidated is None:
+        logger.error(
+            "session_id=%s workspace_path no longer within allowlist; refusing terminal spawn",
+            session_id,
+        )
+        await websocket.close(code=4008, reason="Workspace path no longer permitted")
+        return
+    workspace_path = str(revalidated)
+
     # --- Per-user connection cap ---
     current = _user_terminal_counts.get(user_id, 0)
     if current >= _MAX_TERMINALS_PER_USER:

tests/api/test_session_chat_ws.py

@@ -77,6 +77,45 @@ def test_chat_ws_ownership_mismatch_closes():
         assert exc.value.code == 1008
 
 
+def test_chat_ws_symlink_escape_rejected(tmp_path, monkeypatch):
+    """TOCTOU: stored path swapped to a symlink outside the root → WS closes (#704)."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    from codeframe.ui.routers.session_chat_ws import router
+
+    base = tmp_path / "allowed"
+    base.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    swapped = base / "proj"
+    swapped.symlink_to(outside)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(base))
+    monkeypatch.setenv("CODEFRAME_DEPLOYMENT_MODE", "self_hosted")
+
+    app = FastAPI()
+    app.include_router(router)
+    fake_db = MagicMock()
+    fake_db.interactive_sessions.get.return_value = {
+        "state": "active",
+        "workspace_path": str(swapped),
+        "user_id": 1,
+    }
+    app.state.db = fake_db
+    client = TestClient(app)
+
+    with patch(
+        "codeframe.ui.routers.session_chat_ws._authenticate_websocket",
+        new=AsyncMock(return_value=(True, 1)),
+    ):
+        with pytest.raises(WebSocketDisconnect) as exc:
+            with client.websocket_connect("/ws/sessions/s1/chat?token=x"):
+                pass
+        assert exc.value.code == 1008
+
+
 # ---------------------------------------------------------------------------
 # Auth tests
 # ---------------------------------------------------------------------------

tests/ui/test_session_owner_scoping.py

@@ -0,0 +1,101 @@
+"""Session REST endpoints must be scoped by owner (issue #704).
+
+`POST /api/v2/sessions` persists ``user_id`` and the terminal/chat WebSockets
+enforce ownership, but the REST list/get/delete/messages endpoints queried
+without an owner filter — in an authenticated multi-user deployment one tenant
+could enumerate/read/modify/end another tenant's sessions. These tests pin the
+owner-scoping (and the no-auth passthrough).
+"""
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from codeframe.auth.dependencies import require_auth
+from codeframe.platform_store.database import Database
+from codeframe.ui.routers.interactive_sessions_v2 import router
+
+pytestmark = pytest.mark.v2
+
+
+@pytest.fixture
+def app(monkeypatch):
+    # Owner-scoping is independent of the workspace allowlist; keep it off.
+    monkeypatch.delenv("WORKSPACE_ROOT", raising=False)
+    monkeypatch.setenv("CODEFRAME_DEPLOYMENT_MODE", "self_hosted")
+    application = FastAPI()
+    application.include_router(router)
+    db = Database(":memory:")
+    db.initialize()
+    # Seed the two users that sessions are owned by (FK: sessions.user_id → users.id).
+    for uid in (1, 2):
+        db.conn.execute(
+            "INSERT OR IGNORE INTO users (id, email, hashed_password) VALUES (?, ?, 'x')",
+            (uid, f"u{uid}@example.com"),
+        )
+    db.conn.commit()
+    application.state.db = db
+    return application
+
+
+def _as_user(app, user_id):
+    """Override auth so requests act as ``user_id`` (None = no-auth mode)."""
+    app.dependency_overrides[require_auth] = lambda: {"user_id": user_id}
+    return TestClient(app, raise_server_exceptions=True)
+
+
+def _create(client, path="/tmp/ws"):
+    resp = client.post("/api/v2/sessions", json={"workspace_path": path})
+    assert resp.status_code == 201, resp.text
+    return resp.json()["id"]
+
+
+class TestOwnerScoping:
+    def test_list_only_returns_own_sessions(self, app):
+        s1 = _create(_as_user(app, 1), "/tmp/a")
+        _create(_as_user(app, 2), "/tmp/b")
+
+        ids = [s["id"] for s in _as_user(app, 1).get("/api/v2/sessions").json()]
+        assert ids == [s1]
+
+    def test_get_other_users_session_404(self, app):
+        other = _create(_as_user(app, 2))
+        assert _as_user(app, 1).get(f"/api/v2/sessions/{other}").status_code == 404
+
+    def test_get_own_session_200(self, app):
+        mine = _create(_as_user(app, 1))
+        assert _as_user(app, 1).get(f"/api/v2/sessions/{mine}").status_code == 200
+
+    def test_delete_other_users_session_404(self, app):
+        other = _create(_as_user(app, 2))
+        client = _as_user(app, 1)
+        assert client.delete(f"/api/v2/sessions/{other}").status_code == 404
+        # And it is still active for its real owner.
+        assert _as_user(app, 2).get(f"/api/v2/sessions/{other}").json()["state"] == "active"
+
+    def test_post_message_to_other_users_session_404(self, app):
+        other = _create(_as_user(app, 2))
+        resp = _as_user(app, 1).post(
+            f"/api/v2/sessions/{other}/messages",
+            json={"role": "user", "content": "hi"},
+        )
+        assert resp.status_code == 404
+
+    def test_get_messages_of_other_users_session_404(self, app):
+        other = _create(_as_user(app, 2))
+        assert (
+            _as_user(app, 1).get(f"/api/v2/sessions/{other}/messages").status_code == 404
+        )
+
+
+class TestNoAuthPassthrough:
+    """With auth off (user_id is None) ownership is not enforced — unchanged."""
+
+    def test_list_returns_all(self, app):
+        _create(_as_user(app, None), "/tmp/a")
+        _create(_as_user(app, None), "/tmp/b")
+        assert len(_as_user(app, None).get("/api/v2/sessions").json()) == 2
+
+    def test_get_any_session(self, app):
+        sid = _create(_as_user(app, None))
+        assert _as_user(app, None).get(f"/api/v2/sessions/{sid}").status_code == 200