fix: harden setup launch and Athenaeum ingestion

pantheon-core/tests/test_hestia.py

@@ -1,10 +1,10 @@
 """Tests for gods/hestia.py — HestiaChecker health checks."""
 from __future__ import annotations
 
+import sys
 from unittest.mock import MagicMock, patch
 
 import httpx
-import pytest
 
 from gods.hestia import HealthStatus, HestiaChecker
 
@@ -32,9 +32,8 @@ def _make_checker() -> HestiaChecker:
 class TestCheckOllama:
     def test_returns_ok_on_200(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(200)
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.return_value = _mock_response(200)
             result = checker.check_ollama()
         assert result.service == "ollama"
         assert result.ok is True
@@ -43,69 +42,72 @@ def test_returns_ok_on_200(self):
 
     def test_returns_failure_on_500(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(500)
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.return_value = _mock_response(500)
             result = checker.check_ollama()
         assert result.ok is False
         assert "500" in result.error
 
     def test_returns_failure_on_connection_error(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.side_effect = httpx.ConnectError("refused")
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.side_effect = httpx.ConnectError("refused")
             result = checker.check_ollama()
         assert result.ok is False
         assert result.error is not None
 
     def test_custom_host_and_port(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(200)
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.return_value = _mock_response(200)
             result = checker.check_ollama(host="myhost", port=9999)
         assert result.ok is True
-        call_url = ctx.get.call_args[0][0]
+        call_url = mock_get.call_args[0][0]
         assert "myhost:9999" in call_url
 
 
 class TestCheckChromadb:
-    def test_returns_ok_on_200(self):
+    def test_returns_ok_when_persistent_client_works(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(200)
+        chromadb = MagicMock()
+        collection = MagicMock()
+        collection.count.return_value = 3
+        chromadb.PersistentClient.return_value.list_collections.return_value = [collection]
+
+        with patch.dict(sys.modules, {"chromadb": chromadb}):
             result = checker.check_chromadb()
+
         assert result.service == "chromadb"
         assert result.ok is True
+        assert result.latency_ms is not None
+        assert "vectors" in (result.error or "")
 
-    def test_returns_failure_on_exception(self):
+    def test_returns_failure_when_persistent_client_errors(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.side_effect = TimeoutError("timed out")
+        chromadb = MagicMock()
+        chromadb.PersistentClient.side_effect = TimeoutError("timed out")
+
+        with patch.dict(sys.modules, {"chromadb": chromadb}):
             result = checker.check_chromadb()
+
         assert result.ok is False
         assert "timed out" in result.error
 
 
 class TestCheckPantheonApi:
     def test_returns_ok_on_200(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(200)
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.return_value = _mock_response(200)
             result = checker.check_pantheon_api()
         assert result.service == "pantheon-api"
         assert result.ok is True
 
     def test_404_is_still_ok(self):
         """4xx responses mean the server is alive — we accept < 500."""
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(404)
+        with patch("gods.hestia.httpx.get") as mock_get:
+            mock_get.return_value = _mock_response(404)
             result = checker.check_pantheon_api()
         assert result.ok is True
 
@@ -116,15 +118,18 @@ def test_404_is_still_ok(self):
 
 
 class TestCheckAll:
-    def test_returns_three_statuses(self):
+    def test_returns_all_statuses(self):
         checker = _make_checker()
-        with patch("gods.hestia.httpx.Client") as mock_client_cls:
-            ctx = mock_client_cls.return_value.__enter__.return_value
-            ctx.get.return_value = _mock_response(200)
+        with patch.object(checker, "check_ollama", return_value=HealthStatus("ollama", True)), \
+             patch.object(checker, "check_chromadb", return_value=HealthStatus("chromadb", True)), \
+             patch.object(checker, "check_pantheon_api", return_value=HealthStatus("pantheon-api", True)), \
+             patch.object(checker, "check_mcp_server", return_value=HealthStatus("mcp-server", True)), \
+             patch.object(checker, "check_disk_space", return_value=HealthStatus("disk-space", True)):
             results = checker.check_all()
-        assert len(results) == 3
+
+        assert len(results) == 5
         services = {r.service for r in results}
-        assert services == {"ollama", "chromadb", "pantheon-api"}
+        assert services == {"ollama", "chromadb", "pantheon-api", "mcp-server", "disk-space"}
 
     def test_all_healthy_true_when_all_ok(self):
         checker = _make_checker()

pantheon-core/tests/test_security_hardening.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import importlib
+import os
+import sys
+import types
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+def _load_pantheon_plugin(monkeypatch, tmp_path):
+    repo_root = Path(__file__).resolve().parents[2]
+    sys.path.insert(0, str(repo_root / "plugins"))
+    agent_module = types.ModuleType("agent")
+    memory_provider_module = types.ModuleType("agent.memory_provider")
+
+    class MemoryProvider:
+        pass
+
+    memory_provider_module.MemoryProvider = MemoryProvider
+    monkeypatch.setitem(sys.modules, "agent", agent_module)
+    monkeypatch.setitem(sys.modules, "agent.memory_provider", memory_provider_module)
+    monkeypatch.setenv("ATHENAEUM_ROOT", str(tmp_path / "athenaeum"))
+    monkeypatch.setenv("CHROMA_DIR", str(tmp_path / "chroma"))
+    module = importlib.import_module("pantheon")
+    return importlib.reload(module)
+
+
+def test_athenaeum_read_rejects_path_traversal(monkeypatch, tmp_path):
+    pantheon = _load_pantheon_plugin(monkeypatch, tmp_path)
+    root = tmp_path / "athenaeum"
+    root.mkdir()
+    outside = tmp_path / "secret.txt"
+    outside.write_text("do not read me", encoding="utf-8")
+
+    plugin = pantheon.PantheonMemoryProvider()
+    plugin._athenaeum_root = root
+    result = plugin._tool_read({"path": "../secret.txt"})
+
+    assert "error" in result
+    assert "Athenaeum" in result["error"]
+    assert "do not read me" not in str(result)
+
+
+def test_athenaeum_embed_rejects_absolute_paths(monkeypatch, tmp_path):
+    pantheon = _load_pantheon_plugin(monkeypatch, tmp_path)
+    outside = tmp_path / "secret.txt"
+    outside.write_text("do not embed me", encoding="utf-8")
+
+    plugin = pantheon.PantheonMemoryProvider()
+    plugin._athenaeum_root = tmp_path / "athenaeum"
+    plugin._athenaeum_root.mkdir()
+    result = plugin._tool_embed({"path": str(outside)})
+
+    assert "error" in result
+    assert "Athenaeum" in result["error"]
+
+
+def test_url_ingest_blocks_private_network_targets(monkeypatch, tmp_path):
+    repo_root = Path(__file__).resolve().parents[2]
+    sys.path.insert(0, str(repo_root / "plugins"))
+    _load_pantheon_plugin(monkeypatch, tmp_path)
+    monkeypatch.setenv("ATHENAEUM_ROOT", str(tmp_path / "athenaeum"))
+    ingest = importlib.import_module("pantheon.demeter.ingest")
+    ingest = importlib.reload(ingest)
+
+    with patch("httpx.get") as mock_get:
+        result = ingest.ingest_url("http://127.0.0.1:8010/mcp")
+
+    assert result.success is False
+    assert "private" in (result.error or "").lower() or "local" in (result.error or "").lower()
+    mock_get.assert_not_called()

pantheon-core/tests/test_setup_server_launch.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+
+def _load_setup_server():
+    module_path = Path(__file__).resolve().parents[2] / "scripts" / "setup-server.py"
+    spec = importlib.util.spec_from_file_location("pantheon_setup_server", module_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_launch_worker_starts_dashboard_on_port_8787(monkeypatch, tmp_path):
+    """Completing the wizard should launch a real dashboard, not just gateway."""
+    setup_server = _load_setup_server()
+    monkeypatch.setattr(setup_server, "PANTHEON_DIR", str(tmp_path))
+
+    popen_calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        return MagicMock(returncode=0, stdout="", stderr="")
+
+    def fake_popen(cmd, *args, **kwargs):
+        popen_calls.append(list(cmd))
+        return MagicMock(pid=1234)
+
+    def fake_get(url, *args, **kwargs):
+        response = MagicMock()
+        response.status_code = 200
+        return response
+
+    with patch.object(setup_server.subprocess, "run", side_effect=fake_run), \
+         patch.object(setup_server.subprocess, "Popen", side_effect=fake_popen), \
+         patch("httpx.get", side_effect=fake_get), \
+         patch.object(setup_server.time, "sleep", return_value=None):
+        setup_server._launch_worker()
+
+    assert ["hermes", "gateway"] in popen_calls
+    assert any(
+        cmd[:2] == ["hermes", "dashboard"]
+        and "--port" in cmd
+        and cmd[cmd.index("--port") + 1] == "8787"
+        and "--host" in cmd
+        and cmd[cmd.index("--host") + 1] == "127.0.0.1"
+        for cmd in popen_calls
+    )

plugins/pantheon/__init__.py

@@ -163,6 +163,24 @@ def _list_codexes(athenaeum_root: Path) -> List[str]:
     )
 
 
+def _resolve_under_root(root: Path, rel_path: str) -> Path:
+    """Resolve a user-supplied Athenaeum path and ensure it stays in root."""
+    if not rel_path:
+        raise ValueError("path is required")
+
+    root_resolved = root.expanduser().resolve()
+    candidate = Path(rel_path)
+    if candidate.is_absolute():
+        raise ValueError("Path must be relative to the Athenaeum")
+
+    full_path = (root_resolved / candidate).resolve()
+    try:
+        full_path.relative_to(root_resolved)
+    except ValueError as exc:
+        raise ValueError("Path escapes the Athenaeum") from exc
+    return full_path
+
+
 # ---------------------------------------------------------------------------
 # Embedding client
 # ---------------------------------------------------------------------------
@@ -968,7 +986,10 @@ def _tool_read(self, args: dict) -> dict:
         rel_path = args.get("path", "")
         if not rel_path:
             return {"error": "path is required"}
-        full_path = self._athenaeum_root / rel_path
+        try:
+            full_path = _resolve_under_root(self._athenaeum_root, rel_path)
+        except ValueError as exc:
+            return {"error": str(exc)}
         if not full_path.exists():
             return {"error": f"File not found: {rel_path}"}
         if not full_path.is_file():
@@ -985,7 +1006,10 @@ def _tool_read(self, args: dict) -> dict:
 
     def _tool_walk(self, args: dict) -> dict:
         rel_path = args.get("path", "INDEX.md")
-        full_path = self._athenaeum_root / rel_path
+        try:
+            full_path = _resolve_under_root(self._athenaeum_root, rel_path)
+        except ValueError as exc:
+            return {"error": str(exc)}
         if not full_path.exists():
             return {
                 "error": f"INDEX.md not found: {rel_path}",
@@ -1024,7 +1048,10 @@ def _tool_embed(self, args: dict) -> dict:
         rel_path = args.get("path", "")
         if not rel_path:
             return {"error": "path is required"}
-        full_path = self._athenaeum_root / rel_path
+        try:
+            full_path = _resolve_under_root(self._athenaeum_root, rel_path)
+        except ValueError as exc:
+            return {"error": str(exc)}
         if not full_path.exists():
             return {"error": f"File not found: {rel_path}"}