Protect private API routes and scope agent data

SovereignAI-Starter/tests/test_auth.py

@@ -47,6 +47,63 @@ class AuthenticationTests(unittest.TestCase):
         self.assertEqual(response.status_code, 401, response.text)
         self.assertEqual(response.headers["www-authenticate"], "Bearer")
 
+    def test_private_api_routes_require_a_valid_bearer_session(self) -> None:
+        for path, method, payload in (
+            ("/v1/agent/tools", "get", None),
+            ("/v1/agent/skills", "get", None),
+            ("/v1/agent/run", "post", {"task": "hello"}),
+            ("/v1/chat/completions", "post", {"messages": [{"role": "user", "content": "hi"}]}),
+            ("/v1/agent/knowledge/search", "post", {"task": "find", "workspace_path": "."}),
+            ("/v1/web/search", "post", {"query": "test"}),
+        ):
+            response = getattr(self.client, method)(path, json=payload) if payload is not None else getattr(self.client, method)(path)
+            self.assertEqual(response.status_code, 401, f"{method.upper()} {path}: {response.text}")
+
+        audio = self.client.post(
+            "/v1/audio/transcriptions",
+            files={"file": ("recording.wav", b"", "audio/wav")},
+        )
+        self.assertEqual(audio.status_code, 401, audio.text)
+
+    def test_every_private_v1_openapi_operation_declares_bearer_auth(self) -> None:
+        public = {
+            ("/v1/models", "get"),
+            ("/v1/auth/local-session", "post"),
+            ("/v1/auth/register", "post"),
+            ("/v1/auth/login", "post"),
+        }
+        missing: list[str] = []
+        for path, operations in app.openapi()["paths"].items():
+            if not path.startswith("/v1/"):
+                continue
+            for method, operation in operations.items():
+                if (path, method) in public:
+                    continue
+                if not operation.get("security"):
+                    missing.append(f"{method.upper()} {path}")
+        self.assertEqual(missing, [])
+
+    def test_agent_audit_history_is_scoped_to_session_owner(self) -> None:
+        _, first_token = self._register()
+        _, second_token = self._register()
+        first_tool = self.client.get(
+            "/v1/agent/tools", headers={"Authorization": f"Bearer {first_token}"}
+        )
+        second_tool = self.client.get(
+            "/v1/agent/tools", headers={"Authorization": f"Bearer {second_token}"}
+        )
+        first_audit = self.client.get(
+            "/v1/agent/audit", headers={"Authorization": f"Bearer {first_token}"}
+        )
+        second_audit = self.client.get(
+            "/v1/agent/audit", headers={"Authorization": f"Bearer {second_token}"}
+        )
+
+        self.assertEqual(first_audit.status_code, 200, first_audit.text)
+        self.assertEqual(second_audit.status_code, 200, second_audit.text)
+        self.assertEqual([item["id"] for item in first_audit.json()["data"]], [first_tool.headers["x-agent-audit-id"]])
+        self.assertEqual([item["id"] for item in second_audit.json()["data"]], [second_tool.headers["x-agent-audit-id"]])
+
     def test_register_login_me_and_logout_use_revocable_sessions(self) -> None:
         user_id, token = self._register("User@Example.Test")
         me = self.client.get(