fix(security): owner-scope workspace registry list/delete (#720) (#792)

* fix(security): owner-scope workspace registry list/delete (#720)

list_workspaces returned every entry's repo_path and deregister deleted by id
with no ownership check, so in hosted mode one tenant could enumerate and
delete another's registered workspaces. upsert also overwrote owner_user_id
with excluded (None on refresh), nulling a recorded owner.

- list_workspaces / deregister_workspace pass auth["user_id"] to
  list_all(owner_user_id=) / delete(owner_user_id=); auth off (None) keeps
  current behavior
- registry.delete() gains an owner_user_id filter (WHERE id=? AND owner=?)
- upsert: owner_user_id = COALESCE(excluded.owner_user_id, existing)

Closes #720

* docs: note owner-overwrite invariant in registry upsert SQL (#720 review)

* test: allow default JWT secret in owner-scope lifespan test (CI fix, #720)

The 'with TestClient(app)' triggers lifespan startup, which rejects the default
secret when auth is on; set CODEFRAME_ALLOW_INSECURE_SECRET=1 in the fixture so
CI (no AUTH_SECRET) passes. Passed locally only via ambient AUTH_SECRET.

Author: frankbria

codeframe/platform_store/repositories/workspace_registry_repository.py

@@ -61,7 +61,12 @@ def upsert(
                 -- COALESCE so a refresh that omits name/tech_stack (None) keeps the
                 -- previously-stored value instead of nulling it.
                 name = COALESCE(excluded.name, workspaces_registry.name),
-                owner_user_id = excluded.owner_user_id,
+                -- COALESCE so a refresh that omits the owner (None) keeps the
+                -- recorded owner instead of nulling it (#720). A refresh with a
+                -- *different* non-None owner still overwrites; that's safe only
+                -- because path-based tenant isolation (#655) stops user B from
+                -- re-registering a path they don't own.
+                owner_user_id = COALESCE(excluded.owner_user_id, workspaces_registry.owner_user_id),
                 tech_stack = COALESCE(excluded.tech_stack, workspaces_registry.tech_stack),
                 last_opened_at = excluded.last_opened_at
             """,
@@ -149,16 +154,30 @@ def update_last_opened(self, workspace_id: str) -> None:
         )
         self._commit()
 
-    def delete(self, workspace_id: str) -> bool:
+    def delete(self, workspace_id: str, owner_user_id: Optional[int] = None) -> bool:
         """Deregister a workspace (registry-only — never touches disk files).
 
+        Args:
+            workspace_id: Registry entry id to remove.
+            owner_user_id: When provided (auth enabled), only an entry owned by
+                this user is removed — a tenant cannot delete another's entry
+                (#720). When ``None`` (auth disabled/local), no owner filter is
+                applied and behavior is unchanged.
+
         Returns:
-            True if a row was removed, False if the id was not found.
+            True if a row was removed, False if the id was not found (or not
+            owned by ``owner_user_id`` when supplied).
         """
-        cursor = self._execute(
-            "DELETE FROM workspaces_registry WHERE id = ?",
-            (workspace_id,),
-        )
+        if owner_user_id is not None:
+            cursor = self._execute(
+                "DELETE FROM workspaces_registry WHERE id = ? AND owner_user_id = ?",
+                (workspace_id, owner_user_id),
+            )
+        else:
+            cursor = self._execute(
+                "DELETE FROM workspaces_registry WHERE id = ?",
+                (workspace_id,),
+            )
         self._commit()
         return cursor.rowcount > 0
 

codeframe/ui/routers/workspace_v2.py

@@ -188,7 +188,10 @@ def _register_workspace(
 
 @router.get("", response_model=WorkspaceListResponse)
 @rate_limit_standard()
-async def list_workspaces(request: Request) -> WorkspaceListResponse:
+async def list_workspaces(
+    request: Request,
+    auth: dict = Depends(require_auth),
+) -> WorkspaceListResponse:
     """List all registered workspaces (issue #601).
 
     Returns server-side registry entries ordered by recency, each annotated with
@@ -210,7 +213,9 @@ async def list_workspaces(request: Request) -> WorkspaceListResponse:
             ),
         )
 
-    entries = registry.list_all()
+    # Owner-scope the listing when auth is enabled (#720): a tenant sees only
+    # its own registered workspaces. auth off → user_id None → all entries.
+    entries = registry.list_all(owner_user_id=auth.get("user_id"))
     # path_exists is a per-entry blocking stat() inside an async handler. The
     # registry is recency-scoped and small in practice, so the event-loop cost is
     # negligible; revisit (e.g. run_in_executor) only if the list grows large.
@@ -511,6 +516,7 @@ async def check_workspace_exists(
 async def deregister_workspace(
     request: Request,
     workspace_id: str,
+    auth: dict = Depends(require_auth),
 ) -> Response:
     """Deregister a workspace from the server-side registry (issue #601).
 
@@ -536,7 +542,9 @@ async def deregister_workspace(
             ),
         )
 
-    deleted = registry.delete(workspace_id)
+    # Owner-scope the delete (#720): a tenant cannot deregister another's entry.
+    # auth off → user_id None → no owner filter (unchanged local behavior).
+    deleted = registry.delete(workspace_id, owner_user_id=auth.get("user_id"))
     if not deleted:
         raise HTTPException(
             status_code=404,

tests/platform_store/test_workspace_registry_repository.py

@@ -148,3 +148,50 @@ def test_delete_existing_returns_true(self, db):
 
     def test_delete_missing_returns_false(self, db):
         assert db.workspace_registry.delete("does-not-exist") is False
+
+
+@pytest.fixture
+def db_two_users(db):
+    """Registry DB with two real users (owner_user_id FKs to users.id)."""
+    db.conn.execute(
+        """
+        INSERT OR REPLACE INTO users (
+            id, email, name, hashed_password,
+            is_active, is_superuser, is_verified, email_verified
+        ) VALUES
+            (1, 'a@x.co', 'A', '!DISABLED!', 1, 0, 1, 1),
+            (2, 'b@x.co', 'B', '!DISABLED!', 1, 0, 1, 1)
+        """
+    )
+    db.conn.commit()
+    return db
+
+
+class TestOwnerScoping:
+    """Issue #720 / P0.9: registry must be owner-scoped so one tenant cannot
+    enumerate or delete another's workspaces, and a refresh must not null the
+    recorded owner."""
+
+    def test_upsert_preserves_owner_on_ownerless_refresh(self, db_two_users):
+        db = db_two_users
+        db.workspace_registry.upsert(repo_path="/p/alpha", name="alpha", owner_user_id=1)
+        # A later refresh that omits the owner (None) must keep owner=1.
+        db.workspace_registry.upsert(repo_path="/p/alpha", name="alpha", owner_user_id=None)
+        assert db.workspace_registry.get_by_path("/p/alpha")["owner_user_id"] == 1
+
+    def test_delete_is_owner_scoped(self, db_two_users):
+        db = db_two_users
+        entry = db.workspace_registry.upsert(repo_path="/p/a", name="a", owner_user_id=1)
+        wid = entry["id"]
+        # User 2 cannot delete user 1's entry.
+        assert db.workspace_registry.delete(wid, owner_user_id=2) is False
+        assert db.workspace_registry.get_by_id(wid) is not None
+        # Owner 1 can.
+        assert db.workspace_registry.delete(wid, owner_user_id=1) is True
+        assert db.workspace_registry.get_by_id(wid) is None
+
+    def test_delete_without_owner_filter_unchanged(self, db_two_users):
+        db = db_two_users
+        entry = db.workspace_registry.upsert(repo_path="/p/a", name="a", owner_user_id=1)
+        # auth-off path (owner_user_id=None): no filter, deletes by id.
+        assert db.workspace_registry.delete(entry["id"]) is True

tests/ui/test_workspace_registry_owner_scope.py

@@ -0,0 +1,85 @@
+"""Router-level owner scoping for the workspace registry (#720 / P0.9).
+
+With auth enabled, `GET /api/v2/workspaces` and `DELETE /api/v2/workspaces/{id}`
+must be owner-scoped: user B cannot see or deregister user A's registered
+workspace.
+"""
+
+import importlib
+
+import pytest
+from fastapi.testclient import TestClient
+
+from codeframe.auth.manager import reset_auth_engine
+from codeframe.platform_store.database import Database
+from tests.conftest import create_test_jwt_token
+
+pytestmark = pytest.mark.v2
+
+
+@pytest.fixture
+def app_with_user_a_workspace(tmp_path, monkeypatch):
+    """Auth-on server; users 1 & 2 seeded; one workspace owned by user 1."""
+    db_path = tmp_path / "state.db"
+    monkeypatch.setenv("DATABASE_PATH", str(db_path))
+    monkeypatch.setenv("CODEFRAME_AUTH_REQUIRED", "true")
+    # The `with TestClient(app)` below runs the lifespan startup, which rejects
+    # the default JWT secret unless this escape hatch is set (self-hosted only).
+    # create_test_jwt_token signs with the same default secret the server then
+    # validates with, so tokens stay valid.
+    monkeypatch.setenv("CODEFRAME_ALLOW_INSECURE_SECRET", "1")
+    monkeypatch.delenv("AUTH_SECRET", raising=False)
+    reset_auth_engine()
+
+    db = Database(db_path)
+    db.initialize()
+    db.conn.execute(
+        """
+        INSERT OR REPLACE INTO users (id, email, name, hashed_password,
+            is_active, is_superuser, is_verified, email_verified)
+        VALUES (1,'a@x.co','A','!DISABLED!',1,0,1,1),
+               (2,'b@x.co','B','!DISABLED!',1,0,1,1)
+        """
+    )
+    db.conn.commit()
+    entry = db.workspace_registry.upsert(
+        repo_path="/home/a/projects/alpha", name="alpha", owner_user_id=1
+    )
+    db.close()
+
+    from codeframe.ui import server
+
+    importlib.reload(server)
+    yield server.app, entry["id"]
+    reset_auth_engine()
+
+
+def _bearer(user_id):
+    return {"Authorization": f"Bearer {create_test_jwt_token(user_id)}"}
+
+
+def _list_paths(client, user_id):
+    r = client.get("/api/v2/workspaces", headers=_bearer(user_id))
+    assert r.status_code == 200, r.text
+    return {w["repo_path"] for w in r.json()["workspaces"]}
+
+
+def test_user_b_cannot_list_user_a_workspace(app_with_user_a_workspace):
+    app, _ = app_with_user_a_workspace
+    with TestClient(app) as client:  # context manager runs startup (wires app.state.db)
+        assert "/home/a/projects/alpha" not in _list_paths(client, 2)
+
+
+def test_user_a_sees_own_workspace(app_with_user_a_workspace):
+    app, _ = app_with_user_a_workspace
+    with TestClient(app) as client:
+        assert "/home/a/projects/alpha" in _list_paths(client, 1)
+
+
+def test_user_b_cannot_delete_user_a_workspace(app_with_user_a_workspace):
+    app, wid = app_with_user_a_workspace
+    with TestClient(app) as client:
+        r = client.delete(f"/api/v2/workspaces/{wid}", headers=_bearer(2))
+        assert r.status_code == 404  # not owned → not found for user B
+        # ...and user A can still delete it.
+        assert client.delete(f"/api/v2/workspaces/{wid}", headers=_bearer(1)).status_code == 204