Scope workspace roots per account

SovereignAI-Starter/tests/test_agent_skills.py

@@ -21,6 +21,11 @@ class AgentSkillTests(unittest.TestCase):
     def setUpClass(cls) -> None:
         cls.client = authenticated_client(app)
         cls.workspace = str(Path(__file__).resolve().parents[1])
+        cls.workspace_environment = patch.dict(
+            os.environ, {"SOVEREIGNAI_ALLOWED_WORKSPACES": cls.workspace}
+        )
+        cls.workspace_environment.start()
+        cls.addClassCleanup(cls.workspace_environment.stop)
 
     def test_skill_catalog_discloses_scope_and_permissions(self) -> None:
         response = self.client.get("/v1/agent/skills")

SovereignAI-Starter/tests/test_auth.py

@@ -1,6 +1,9 @@
+import json
 import os
+import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 from uuid import uuid4
 
 _PROJECT_ROOT = Path(__file__).resolve().parents[1]
@@ -199,6 +202,50 @@ class AuthenticationTests(unittest.TestCase):
         )
         auth.clear_login_failures(email, client_host)
 
+    def test_workspace_files_are_isolated_by_account_assignment(self) -> None:
+        first_email = f"{uuid4().hex}@example.test"
+        second_email = f"{uuid4().hex}@example.test"
+        _, first_token = self._register(first_email)
+        _, second_token = self._register(second_email)
+        with tempfile.TemporaryDirectory() as directory:
+            parent = Path(directory)
+            first_root = parent / "first"
+            second_root = parent / "second"
+            first_root.mkdir()
+            second_root.mkdir()
+            (first_root / "one.md").write_text("first", encoding="utf-8")
+            (second_root / "two.md").write_text("second", encoding="utf-8")
+            environment = {
+                "SOVEREIGNAI_ALLOWED_WORKSPACES": str(parent),
+                "SOVEREIGNAI_USER_WORKSPACES": json.dumps(
+                    {
+                        first_email: [str(first_root)],
+                        second_email: [str(second_root)],
+                    }
+                ),
+            }
+            with patch.dict(os.environ, environment, clear=False):
+                first_own = self.client.post(
+                    "/v1/agent/workspace/files",
+                    headers={"Authorization": f"Bearer {first_token}"},
+                    json={"workspace_path": str(first_root)},
+                )
+                first_other = self.client.post(
+                    "/v1/agent/workspace/files",
+                    headers={"Authorization": f"Bearer {first_token}"},
+                    json={"workspace_path": str(second_root)},
+                )
+                second_own = self.client.post(
+                    "/v1/agent/workspace/files",
+                    headers={"Authorization": f"Bearer {second_token}"},
+                    json={"workspace_path": str(second_root)},
+                )
+            self.assertEqual(first_own.status_code, 200, first_own.text)
+            self.assertEqual(first_own.json()["files"], ["one.md"])
+            self.assertEqual(first_other.status_code, 403, first_other.text)
+            self.assertEqual(second_own.status_code, 200, second_own.text)
+            self.assertEqual(second_own.json()["files"], ["two.md"])
+
     def test_local_bootstrap_is_restricted_to_loopback_clients(self) -> None:
         remote = self.client.post("/v1/auth/local-session", json={})
         self.assertEqual(remote.status_code, 403, remote.text)

SovereignAI-Starter/tests/test_knowledge.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import json
 import os
 import tempfile
 import unittest
@@ -25,6 +26,10 @@ class KnowledgeIndexTests(unittest.TestCase):
     def setUp(self) -> None:
         self.temp = tempfile.TemporaryDirectory()
         self.base = Path(self.temp.name)
+        self.workspace_environment = patch.dict(
+            os.environ, {"SOVEREIGNAI_ALLOWED_WORKSPACES": str(self.base)}
+        )
+        self.workspace_environment.start()
         self.database = self.base / "knowledge.sqlite3"
         self.workspace = self.base / "project"
         self.workspace.mkdir()
@@ -34,6 +39,7 @@ class KnowledgeIndexTests(unittest.TestCase):
             knowledge.initialize()
 
     def tearDown(self) -> None:
+        self.workspace_environment.stop()
         self.temp.cleanup()
 
     def _index(self, text: str, root: Path | None = None) -> None:
@@ -167,11 +173,14 @@ class KnowledgeIndexTests(unittest.TestCase):
     def test_api_knowledge_index_isolated_between_account_sessions(self) -> None:
         marker = "OwnerScopedKnowledgeMarker"
         (self.workspace / "private.md").write_text(marker, encoding="utf-8")
+        (self.other_workspace / "private.md").write_text(marker, encoding="utf-8")
+        first_email = f"{uuid4().hex}@example.test"
+        second_email = f"{uuid4().hex}@example.test"
         owner_id = auth.create_account(
-            f"{uuid4().hex}@example.test", "account one secure passphrase"
+            first_email, "account one secure passphrase"
         )
         other_id = auth.create_account(
-            f"{uuid4().hex}@example.test", "account two secure passphrase"
+            second_email, "account two secure passphrase"
         )
         self.addCleanup(self._delete_test_users, owner_id, other_id)
         owner_token, _ = auth.issue_session(owner_id)
@@ -182,20 +191,29 @@ class KnowledgeIndexTests(unittest.TestCase):
         other_client = TestClient(
             app, headers={"Authorization": f"Bearer {other_token}"}
         )
-        payload = {
-            "workspace_path": str(self.workspace),
-            "files": ["private.md"],
-        }
-
-        indexed = owner_client.post("/v1/agent/knowledge/index", json=payload)
-        owner_results = owner_client.post(
-            "/v1/agent/knowledge/search",
-            json={"workspace_path": str(self.workspace), "task": marker},
-        )
-        other_results = other_client.post(
-            "/v1/agent/knowledge/search",
-            json={"workspace_path": str(self.workspace), "task": marker},
-        )
+        with patch.dict(
+            os.environ,
+            {
+                "SOVEREIGNAI_USER_WORKSPACES": json.dumps(
+                    {
+                        first_email: [str(self.workspace)],
+                        second_email: [str(self.other_workspace)],
+                    }
+                ),
+            },
+        ):
+            indexed = owner_client.post(
+                "/v1/agent/knowledge/index",
+                json={"workspace_path": str(self.workspace), "files": ["private.md"]},
+            )
+            owner_results = owner_client.post(
+                "/v1/agent/knowledge/search",
+                json={"workspace_path": str(self.workspace), "task": marker},
+            )
+            other_results = other_client.post(
+                "/v1/agent/knowledge/search",
+                json={"workspace_path": str(self.other_workspace), "task": marker},
+            )
 
         self.assertEqual(indexed.status_code, 200, indexed.text)
         self.assertEqual(owner_results.status_code, 200, owner_results.text)
@@ -207,6 +225,7 @@ class KnowledgeIndexTests(unittest.TestCase):
             workspace_path=self.workspace,
             relative_path="private.md",
         )
+        (self.other_workspace / "private.md").unlink(missing_ok=True)
 
     @staticmethod
     def _delete_test_users(*user_ids: str) -> None:

SovereignAI-Starter/tests/test_mixed_pdf_knowledge_integration.py

@@ -38,7 +38,13 @@ class MixedPdfKnowledgeIntegrationTests(unittest.TestCase):
         try:
             with (
                 patch.object(database, "DATABASE_PATH", database_path),
-                patch.dict(os.environ, {"KNOWLEDGE_EMBEDDING_MODEL": "granite-embedding:278m"}),
+                patch.dict(
+                    os.environ,
+                    {
+                        "KNOWLEDGE_EMBEDDING_MODEL": "granite-embedding:278m",
+                        "SOVEREIGNAI_ALLOWED_WORKSPACES": str(data_dir),
+                    },
+                ),
                 patch("app.main.embeddings.embed_texts", new=AsyncMock(side_effect=fake_embeddings)),
             ):
                 database.initialize_database()

SovereignAI-Starter/tests/test_pdf_analysis.py

@@ -243,17 +243,20 @@ class PdfAnalysisTests(unittest.TestCase):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
             (root / "meetup.pdf").write_bytes(make_pdf("Community Meetup Amman"))
-            listed = self.client.post(
-                "/v1/agent/workspace/files", json={"workspace_path": str(root)}
-            )
-            indexed = self.client.post(
-                "/v1/agent/knowledge/index",
-                json={"workspace_path": str(root), "files": ["meetup.pdf"]},
-            )
-            searched = self.client.post(
-                "/v1/agent/knowledge/search",
-                json={"workspace_path": str(root), "task": "Community Meetup"},
-            )
+            with patch.dict(
+                os.environ, {"SOVEREIGNAI_ALLOWED_WORKSPACES": str(root)}
+            ):
+                listed = self.client.post(
+                    "/v1/agent/workspace/files", json={"workspace_path": str(root)}
+                )
+                indexed = self.client.post(
+                    "/v1/agent/knowledge/index",
+                    json={"workspace_path": str(root), "files": ["meetup.pdf"]},
+                )
+                searched = self.client.post(
+                    "/v1/agent/knowledge/search",
+                    json={"workspace_path": str(root), "task": "Community Meetup"},
+                )
 
         self.assertEqual(listed.status_code, 200, listed.text)
         self.assertIn("meetup.pdf", listed.json()["files"])

SovereignAI-Starter/tests/test_workspace_changes.py

@@ -5,10 +5,11 @@ from __future__ import annotations
 import tempfile
 import unittest
 import os
+import json
 from pathlib import Path
 from unittest.mock import patch
 
-from app import workspace
+from app import auth, workspace
 
 
 class WorkspaceChangeTests(unittest.TestCase):
@@ -99,10 +100,49 @@ class WorkspaceChangeTests(unittest.TestCase):
             {"SOVEREIGNAI_ALLOWED_WORKSPACES": str(allowed)},
             clear=False,
         ):
-            with self.assertRaisesRegex(ValueError, "خارج مجلدات المشاريع"):
+            with self.assertRaisesRegex(ValueError, "خارج مساحة العمل"):
                 workspace.selected_root(str(outside))
             self.assertEqual(workspace.selected_root(str(nested)), nested.resolve())
 
+    def test_account_workspaces_are_separate_and_admin_config_must_not_overlap(self) -> None:
+        allowed = self.root / "accounts"
+        first_root = allowed / "first"
+        second_root = allowed / "second"
+        first_root.mkdir(parents=True)
+        second_root.mkdir()
+        config = {
+            "first@example.test": [str(first_root)],
+            "second@example.test": [str(second_root)],
+        }
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "SOVEREIGNAI_ALLOWED_WORKSPACES": str(allowed),
+                    "SOVEREIGNAI_USER_WORKSPACES": json.dumps(config),
+                },
+                clear=False,
+            ),
+            patch.object(
+                auth,
+                "account_email",
+                side_effect=lambda user_id: {
+                    "user-first": "first@example.test",
+                    "user-second": "second@example.test",
+                }.get(user_id),
+            ),
+        ):
+            self.assertEqual(
+                workspace.selected_root(str(first_root), user_id="user-first"),
+                first_root.resolve(),
+            )
+            with self.assertRaisesRegex(ValueError, "خارج مساحة العمل"):
+                workspace.selected_root(str(second_root), user_id="user-first")
+            config["second@example.test"] = [str(allowed)]
+            os.environ["SOVEREIGNAI_USER_WORKSPACES"] = json.dumps(config)
+            with self.assertRaisesRegex(ValueError, "متداخلة"):
+                workspace.selected_root(str(first_root), user_id="user-first")
+
     def test_rejects_create_overwrite_and_update_of_missing_file(self) -> None:
         target = self.root / "src" / "existing.py"
         target.write_text("value = 1\n", encoding="utf-8")