Add controlled Athena platform operator
This commit is contained in:
Mikei386
@@ -0,0 +1,135 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Offline safety and workflow tests for the Athena Operator executor."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SOURCE = Path(__file__).parents[1] / "platform" / "operator" / "athena_operatord.py"
|
||||
|
||||
|
||||
def load_module():
|
||||
spec = importlib.util.spec_from_file_location("athena_operatord_tested", SOURCE)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class OperatorTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.module = load_module()
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
root = Path(self.temporary.name)
|
||||
self.repo, self.stack = root / "repository", root / "stack"
|
||||
self.models, self.state = root / "models", root / "state"
|
||||
for path in (self.repo, self.stack, self.models, self.state):
|
||||
path.mkdir()
|
||||
(self.repo / ".git").mkdir()
|
||||
(self.repo / "docs").mkdir()
|
||||
(self.stack / "docs").mkdir()
|
||||
for base in (self.repo, self.stack):
|
||||
(base / "docs" / "test.md").write_text("before\n", encoding="utf-8")
|
||||
self.module.REPOSITORY = self.repo.resolve()
|
||||
self.module.STACK = self.stack.resolve()
|
||||
self.module.MODELS = self.models.resolve()
|
||||
self.module.STATE = self.state.resolve()
|
||||
self.module.audit = lambda *args, **kwargs: None
|
||||
self.module.run = lambda argv, **kwargs: {"argv": argv, "exit_code": 0, "output": "ok"}
|
||||
|
||||
def tearDown(self):
|
||||
self.temporary.cleanup()
|
||||
|
||||
def prepare_file(self):
|
||||
before = self.module.sha((self.repo / "docs" / "test.md").read_bytes())
|
||||
return self.module.prepare({
|
||||
"operation": "file_update",
|
||||
"payload": {"files": [{"path": "docs/test.md", "content": "after\n", "expected_sha256": before}]},
|
||||
})
|
||||
|
||||
def test_preview_changes_nothing(self):
|
||||
proposal = self.prepare_file()
|
||||
self.assertEqual((self.repo / "docs" / "test.md").read_text(), "before\n")
|
||||
self.assertIn("-before", proposal["preview"])
|
||||
self.assertIn("+after", proposal["preview"])
|
||||
|
||||
def test_exact_later_confirmation_updates_repo_and_deploy_tree(self):
|
||||
proposal = self.prepare_file()
|
||||
result = self.module.execute({"ticket": proposal["ticket"], "confirmation": proposal["required_confirmation"]})
|
||||
self.assertEqual((self.repo / "docs" / "test.md").read_text(), "after\n")
|
||||
self.assertEqual((self.stack / "docs" / "test.md").read_text(), "after\n")
|
||||
self.assertEqual(result["operation"], "file_update")
|
||||
with self.assertRaises(ValueError):
|
||||
self.module.execute({"ticket": proposal["ticket"], "confirmation": proposal["required_confirmation"]})
|
||||
|
||||
def test_wrong_confirmation_and_expired_ticket_are_rejected(self):
|
||||
proposal = self.prepare_file()
|
||||
with self.assertRaises(PermissionError):
|
||||
self.module.execute({"ticket": proposal["ticket"], "confirmation": "yes"})
|
||||
path = self.state / "pending" / f"{proposal['ticket']}.json"
|
||||
record = json.loads(path.read_text())
|
||||
record["expires"] = int(time.time()) - 1
|
||||
self.module.json_write(path, record)
|
||||
with self.assertRaises(PermissionError):
|
||||
self.module.execute({"ticket": proposal["ticket"], "confirmation": proposal["required_confirmation"]})
|
||||
|
||||
def test_source_drift_blocks_apply(self):
|
||||
proposal = self.prepare_file()
|
||||
(self.repo / "docs" / "test.md").write_text("drift\n")
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.module.execute({"ticket": proposal["ticket"], "confirmation": proposal["required_confirmation"]})
|
||||
|
||||
def test_remote_access_and_power_operations_do_not_exist(self):
|
||||
self.assertNotIn("shell", self.module.ALLOWED_OPERATIONS)
|
||||
for operation in ("shutdown", "reboot", "ssh", "network", "command"):
|
||||
with self.assertRaises(ValueError):
|
||||
self.module.normalise_operation(operation, {})
|
||||
|
||||
def test_wireguard_gateway_cannot_be_stopped(self):
|
||||
with self.assertRaises(PermissionError):
|
||||
self.module.normalise_operation("container_action", {"action": "stop", "containers": ["mike-ai-wireguard-gateway"]})
|
||||
|
||||
def test_only_huggingface_model_downloads_are_accepted(self):
|
||||
with self.assertRaises(PermissionError):
|
||||
self.module.normalise_operation("model_download", {"url": "https://evil.invalid/model.gguf", "destination": "model.gguf"})
|
||||
payload, _ = self.module.normalise_operation("model_download", {"url": "https://huggingface.co/org/repo/resolve/main/model.gguf", "destination": "qwen/model.gguf"})
|
||||
self.assertEqual(payload["destination"], "qwen/model.gguf")
|
||||
|
||||
def test_openwebui_sync_is_structured_and_requires_a_ticket(self):
|
||||
payload, preview = self.module.normalise_operation("openwebui_sync", {})
|
||||
self.assertEqual(payload, {})
|
||||
self.assertIn("Synchronise", preview)
|
||||
|
||||
def test_git_publish_preview_is_bound_to_reviewed_worktree(self):
|
||||
calls = []
|
||||
def fake_run(argv, **kwargs):
|
||||
calls.append(argv)
|
||||
if argv[:3] == ["git", "status", "--short"]:
|
||||
return {"argv": argv, "exit_code": 0, "output": " M compose.yaml"}
|
||||
if argv[:3] == ["git", "diff", "--stat"]:
|
||||
return {"argv": argv, "exit_code": 0, "output": " compose.yaml | 2 +-"}
|
||||
return {"argv": argv, "exit_code": 0, "output": "ok"}
|
||||
self.module.run = fake_run
|
||||
payload, preview = self.module.normalise_operation("git_publish", {"message": "Update Athena platform"})
|
||||
self.assertEqual(payload["reviewed_status"], "M compose.yaml")
|
||||
self.assertIn("compose.yaml | 2 +-", preview)
|
||||
|
||||
def test_git_publish_rejects_empty_repository(self):
|
||||
self.module.run = lambda argv, **kwargs: {"argv": argv, "exit_code": 0, "output": ""}
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.module.normalise_operation("git_publish", {"message": "Update Athena platform"})
|
||||
|
||||
def test_path_traversal_and_protected_paths_are_rejected(self):
|
||||
for path in ("../etc/passwd", ".git/config", "/etc/passwd", "secrets/key"):
|
||||
with self.subTest(path=path), self.assertRaises((ValueError, PermissionError)):
|
||||
self.module.safe_relative(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -1,99 +0,0 @@
|
||||
#!/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)
|
||||
with self.assertRaises(ValueError):
|
||||
self.module.validate_arguments("grep", ["-R", "Athena", "."], 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)
|
||||
@@ -170,12 +170,12 @@ 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):
|
||||
async def test_athena_operator_is_selected_for_platform_work(self):
|
||||
result = await self._select(
|
||||
"Zeige mir im Athena-Terminal mit ls die Dateien des KI-Hosts."
|
||||
"Baue und deploye auf Athena einen neuen MCP-Container."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:athena-terminal-local"]
|
||||
result["tool_ids"], ["server:mcp:athena-operator-local"]
|
||||
)
|
||||
|
||||
async def test_mcp_build_from_github_gets_source_and_platform_context(self):
|
||||
@@ -184,7 +184,7 @@ class AutoToolSelectorTests(unittest.IsolatedAsyncioTestCase):
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:github-local", "server:mcp:athena-platform"],
|
||||
["server:mcp:github-local", "server:mcp:athena-operator-local"],
|
||||
)
|
||||
|
||||
async def test_github_and_explicit_web_are_bounded_to_two(self):
|
||||
|
||||
Reference in New Issue
Block a user