Add bounded read-only Athena terminal MCP

dev/test_athena_terminal_mcp.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Offline abuse and capability tests for the bounded Athena terminal MCP."""
+
+from __future__ import annotations
+
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SOURCE = Path(__file__).parents[1] / "platform" / "mcp" / "athena_terminal_mcp.py"
+
+
+def load_module():
+    spec = importlib.util.spec_from_file_location("athena_terminal_mcp_tested", SOURCE)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+class AthenaTerminalTests(unittest.TestCase):
+    def setUp(self):
+        self.module = load_module()
+        self.temporary = tempfile.TemporaryDirectory()
+        root = Path(self.temporary.name)
+        self.workspace = root / "workspace"
+        self.runtime = root / "runtime"
+        self.workspace.mkdir()
+        self.runtime.mkdir()
+        (self.workspace / "README.md").write_text("Athena evidence\n", encoding="utf-8")
+        (self.workspace / "valid.json").write_text('{"ok":true}\n', encoding="utf-8")
+        self.module.WORKSPACE_ROOT = self.workspace.resolve()
+        self.module.RUNTIME_ROOT = self.runtime.resolve()
+        self.module.ALLOWED_ROOTS = (
+            self.module.WORKSPACE_ROOT,
+            self.module.RUNTIME_ROOT,
+        )
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
+    def test_allowed_read_works(self):
+        result = self.module.run_command(
+            {"program": "cat", "arguments": ["README.md"], "working_directory": "workspace"}
+        )
+        self.assertEqual(result["exit_code"], 0)
+        self.assertEqual(result["output"], "Athena evidence\n")
+        self.assertTrue(result["read_only"])
+
+    def test_power_remote_shell_and_admin_programs_are_blocked(self):
+        for program in (
+            "shutdown", "reboot", "poweroff", "ssh", "scp", "bash",
+            "python3", "docker", "systemctl", "curl", "wget", "sudo",
+        ):
+            with self.subTest(program=program), self.assertRaises(PermissionError):
+                self.module.validate_arguments(program, [], self.workspace)
+
+    def test_shell_syntax_is_blocked(self):
+        for value in ("$(id)", "${HOME}", "`id`", "a|b", "x;y", ">file", "a&&b"):
+            with self.subTest(value=value), self.assertRaises(ValueError):
+                self.module.clean_scalar(value)
+
+    def test_path_escape_and_symlink_escape_are_blocked(self):
+        with self.assertRaises(PermissionError):
+            self.module.safe_path("/etc/passwd", self.workspace)
+        (self.workspace / "escape").symlink_to("/etc/passwd")
+        with self.assertRaises(PermissionError):
+            self.module.safe_path("escape", self.workspace)
+
+    def test_secret_like_path_is_blocked(self):
+        secret = self.workspace / "credentials"
+        secret.write_text("nope", encoding="utf-8")
+        with self.assertRaises(PermissionError):
+            self.module.safe_path("credentials", self.workspace)
+
+    def test_ripgrep_preprocessor_and_find_are_not_available(self):
+        with self.assertRaises(ValueError):
+            self.module.validate_arguments("rg", ["--pre", "sh", "x"], self.workspace)
+        with self.assertRaises(PermissionError):
+            self.module.validate_arguments("find", ["."], self.workspace)
+
+    def test_validation_parses_without_execution(self):
+        result = self.module.validate_source({"path": "valid.json", "kind": "auto"})
+        self.assertTrue(result["valid"])
+        self.assertFalse(result["executed"])
+        self.assertFalse(result["modified"])
+
+    def test_policy_states_missing_capabilities(self):
+        unavailable = " ".join(self.module.policy()["unavailable"])
+        for word in ("SSH", "shutdown", "reboot", "Docker", "network"):
+            self.assertIn(word, unavailable)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

dev/test_openwebui_filters.py

@@ -170,6 +170,23 @@ class AutoToolSelectorTests(unittest.IsolatedAsyncioTestCase):
         )
         self.assertEqual(result["tool_ids"], ["server:mcp:athena-platform"])
 
+    async def test_athena_terminal_is_selected_only_for_explicit_terminal_work(self):
+        result = await self._select(
+            "Zeige mir im Athena-Terminal mit ls die Dateien des KI-Hosts."
+        )
+        self.assertEqual(
+            result["tool_ids"], ["server:mcp:athena-terminal-local"]
+        )
+
+    async def test_mcp_build_from_github_gets_source_and_platform_context(self):
+        result = await self._select(
+            "Ich möchte hierfür einen MCP bauen: https://github.com/foo/bar"
+        )
+        self.assertEqual(
+            result["tool_ids"],
+            ["server:mcp:github-local", "server:mcp:athena-platform"],
+        )
+
     async def test_github_and_explicit_web_are_bounded_to_two(self):
         result = await self._select(
             "Prüfe dieses GitHub Repository und suche zusätzlich im Netz nach Nutzerstimmen."