operator
0065d11e97
- bin/muse_choice_watcher.py + systemd/muse-choices-reconcile.*: automatic choice answering and timer reconciliation - bin/digest.py: fleet log and health summarization - docs/BOX-*-HTTPS.md: comprehensive HTTPS execution contracts and API documentation - docs/MUSE-CHOICES-POLICY.md & docs/SUPERVISION-SPEC.md: autonomous execution specs - tests/test_*.py: unit test suites for HTTPS API, choice watcher, fleet heal, and swarm pruning
354 lines
14 KiB
Python
354 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
test_approvals.py — Unit and integration tests for:
|
|
1. approvals.py module (inspect, check_fleet, auto_approve, allow/deny structure)
|
|
2. box approvals CLI command (check, list, auto, --json)
|
|
3. box-ctl.py RPC actions (approval-check, approval-auto)
|
|
4. gravity.py loop break detection (approval_blocked taxonomy)
|
|
5. muse-chat-api.py account loading and connection
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
BIN_DIR = REPO_ROOT / "bin"
|
|
sys.path.insert(0, str(BIN_DIR))
|
|
|
|
import approvals
|
|
import gravity
|
|
|
|
|
|
class TestApprovalsModule(unittest.TestCase):
|
|
"""Test approvals.py core module functionality."""
|
|
|
|
def test_trusted_ips_configuration(self):
|
|
self.assertIn("34.139.37.135", approvals.TRUSTED_IPS)
|
|
self.assertIn("100.123.153.75", approvals.TRUSTED_IPS)
|
|
self.assertTrue(approvals.is_trusted_target("34.139.37.135"))
|
|
self.assertTrue(approvals.is_trusted_target("status.muse-dev.online"))
|
|
self.assertTrue(approvals.is_trusted_target("1.1.1.1"))
|
|
self.assertFalse(approvals.is_trusted_target("8.8.8.8"))
|
|
self.assertFalse(approvals.is_trusted_target("malicious-site.com"))
|
|
self.assertFalse(approvals.is_trusted_target("evilmuse-dev.online"))
|
|
self.assertFalse(approvals.is_trusted_target(
|
|
"evil.com", "operator-main wants to reach evil.com for status.muse-dev.online"))
|
|
self.assertFalse(approvals.is_trusted_target(None, "status.muse-dev.online"))
|
|
|
|
def test_redact_sensitive(self):
|
|
s1 = "Connecting with Bearer ya29.a0AfH6SMAKskd9238jdf"
|
|
self.assertEqual(approvals.redact_sensitive(s1), "Connecting with Bearer [REDACTED]")
|
|
s2 = "My api_key: secret12345678"
|
|
self.assertEqual(approvals.redact_sensitive(s2), "My api_key: [REDACTED]")
|
|
s3 = "JWT token eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0"
|
|
self.assertIn("[JWT-REDACTED]", approvals.redact_sensitive(s3))
|
|
|
|
def test_get_node_connection_info(self):
|
|
info = approvals.get_node_connection_info("pip")
|
|
self.assertEqual(info["node"], "pip")
|
|
self.assertEqual(info["cdp_port"], 9420)
|
|
self.assertTrue(info["peer_ip"].startswith("10.201."))
|
|
|
|
def test_check_fleet_approvals_structure(self):
|
|
res = approvals.check_fleet_approvals(nodes=["pip", "muse"])
|
|
self.assertIsInstance(res, list)
|
|
self.assertEqual(len(res), 2)
|
|
for it in res:
|
|
self.assertIn("node", it)
|
|
self.assertIn("status", it)
|
|
self.assertIn("has_pending", it)
|
|
self.assertIn("buttons", it)
|
|
self.assertIn("is_trusted", it)
|
|
self.assertIn("input_waits", it)
|
|
|
|
def test_auto_approve_fleet_structure(self):
|
|
res = approvals.auto_approve_fleet(nodes=["pip"])
|
|
self.assertTrue(res.get("ok"))
|
|
self.assertIn("auto_approved", res)
|
|
self.assertIn("untrusted_pending", res)
|
|
self.assertIn("clear_nodes", res)
|
|
|
|
|
|
class TestBoxApprovalsCli(unittest.TestCase):
|
|
"""Test 'box approvals' and 'box approval' CLI commands."""
|
|
|
|
def test_box_approvals_check_json(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "check", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertIn("approvals", data)
|
|
self.assertIsInstance(data["approvals"], list)
|
|
|
|
def test_box_approval_alias(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approval", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
|
|
def test_box_approvals_auto_json(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "auto", "--node", "pip", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
|
|
|
|
class TestBoxCtlApprovals(unittest.TestCase):
|
|
"""Test box-ctl.py allowlisted RPC actions."""
|
|
|
|
def test_box_ctl_approval_check(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "box-ctl.py"), "approval-check"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertIn("approvals", data)
|
|
|
|
def test_box_ctl_approval_auto(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "box-ctl.py"), "approval-auto", "pip"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
|
|
|
|
class TestGravityApprovalIntegration(unittest.TestCase):
|
|
"""Test loop break diagnostics for approvals."""
|
|
|
|
def test_diagnose_breaks_includes_approval_check(self):
|
|
breaks = gravity.diagnose_breaks()
|
|
self.assertIsInstance(breaks, list)
|
|
# Verify schema
|
|
for b in breaks:
|
|
self.assertIn("type", b)
|
|
self.assertIn("severity", b)
|
|
self.assertIn("detail", b)
|
|
|
|
|
|
class TestMuseChatApiConnection(unittest.TestCase):
|
|
"""Test muse-chat-api.py account mapping and connection."""
|
|
|
|
def test_muse_chat_api_approvals_command(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "muse-chat-api.py"), "--account", "pip", "approvals"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertIn(r.returncode, (0, 2))
|
|
if r.returncode == 0:
|
|
self.assertIn("No pending approvals", r.stdout)
|
|
else:
|
|
self.assertIn("APPROVAL_NEEDED", r.stdout)
|
|
|
|
|
|
|
|
class TestApprovalsReplySafety(unittest.TestCase):
|
|
"""Test reply policy enforcement on Main Chat."""
|
|
|
|
def test_reply_main_chat_refusal(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "reply", "pip", "hello test"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
# Should refuse with exit code 2 when target is Main Chat without --allow-main-chat
|
|
self.assertEqual(r.returncode, 2)
|
|
self.assertIn("Refusing reply by sidechat-first policy", r.stderr + r.stdout)
|
|
|
|
|
|
class TestKeyApprovalsAndPasskey(unittest.TestCase):
|
|
"""Test key approval workflow and passkey retrieval architecture."""
|
|
|
|
def test_request_and_resolve_key_approval(self):
|
|
req = approvals.request_key_approval("dev", reason="UnitTest passkey verification", caller="unit-test")
|
|
self.assertTrue(req.get("ok"))
|
|
self.assertEqual(req.get("node"), "dev")
|
|
|
|
# Verify active in check_node_key_request
|
|
pending = approvals.check_node_key_request("dev")
|
|
self.assertIsNotNone(pending)
|
|
self.assertEqual(pending.get("reason"), "UnitTest passkey verification")
|
|
|
|
# Inspect should report KEY_APPROVAL
|
|
info = approvals.inspect_node_approvals("dev")
|
|
self.assertEqual(info.get("status"), "KEY_APPROVAL")
|
|
self.assertTrue(info.get("has_pending"))
|
|
|
|
# Resolve via allow_node_approval
|
|
res = approvals.allow_node_approval("dev", caller="unit-test")
|
|
self.assertTrue(res.get("ok"))
|
|
self.assertEqual(res.get("type"), "key_approval")
|
|
self.assertEqual(res.get("decision"), "allow")
|
|
|
|
# After resolution, pending should be cleared
|
|
cleared = approvals.check_node_key_request("dev")
|
|
self.assertIsNone(cleared)
|
|
|
|
def test_box_passkey_info_json(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "passkey", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertEqual(data.get("operator_pin"), "3128")
|
|
self.assertIn("key_location", data)
|
|
self.assertIn("canonical_path", data["key_location"])
|
|
self.assertEqual(data["key_location"]["canonical_path"], "/srv/box/passkey.txt")
|
|
|
|
def test_box_passkey_fetch_json(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "passkey", "fetch", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertIn("operator_pin", data)
|
|
self.assertEqual(data.get("operator_pin"), "3128")
|
|
self.assertEqual(data.get("vm_host"), "34.139.37.135")
|
|
self.assertEqual(data.get("path"), "/srv/box/passkey.txt")
|
|
self.assertIn("operator_command", data)
|
|
|
|
def test_box_lookup_key(self):
|
|
cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "lookup", "key", "--json"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
self.assertEqual(r.returncode, 0)
|
|
data = json.loads(r.stdout)
|
|
self.assertTrue(data.get("ok"))
|
|
self.assertEqual(data.get("operator_pin"), "3128")
|
|
|
|
def test_cli_request_key_lifecycle(self):
|
|
# 1. Request key
|
|
r_req = subprocess.run([
|
|
sys.executable, str(BIN_DIR / "super-cli.py"),
|
|
"approvals", "request-key", "dev", "--reason", "CLI lifecycle test", "--json"
|
|
], capture_output=True, text=True)
|
|
self.assertEqual(r_req.returncode, 0)
|
|
req_data = json.loads(r_req.stdout)
|
|
self.assertTrue(req_data.get("ok"))
|
|
|
|
# 2. Check shows KEY_APPROVAL
|
|
r_check = subprocess.run([
|
|
sys.executable, str(BIN_DIR / "super-cli.py"),
|
|
"approvals", "check", "--node", "dev", "--json"
|
|
], capture_output=True, text=True)
|
|
self.assertEqual(r_check.returncode, 0)
|
|
check_data = json.loads(r_check.stdout)
|
|
dev_app = next(a for a in check_data["approvals"] if a["node"] == "dev")
|
|
self.assertEqual(dev_app["status"], "KEY_APPROVAL")
|
|
|
|
# 3. Deny key
|
|
r_deny = subprocess.run([
|
|
sys.executable, str(BIN_DIR / "super-cli.py"),
|
|
"approvals", "deny", "dev", "--json"
|
|
], capture_output=True, text=True)
|
|
self.assertEqual(r_deny.returncode, 0)
|
|
deny_data = json.loads(r_deny.stdout)
|
|
self.assertTrue(deny_data.get("ok"))
|
|
self.assertEqual(deny_data.get("decision"), "deny")
|
|
|
|
|
|
class _FakeWS:
|
|
"""Scripted stand-in for a CDP websocket (no network)."""
|
|
|
|
def __init__(self, recvs):
|
|
self._recvs = list(recvs)
|
|
self.sent_ids = []
|
|
|
|
def send(self, msg):
|
|
self.sent_ids.append(json.loads(msg)["id"])
|
|
|
|
def recv(self):
|
|
if not self._recvs:
|
|
raise Exception("recv queue exhausted")
|
|
item = self._recvs.pop(0)
|
|
if callable(item):
|
|
return item(self)
|
|
return item
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
def _echo_last_value(value):
|
|
def _recv(ws):
|
|
return json.dumps({"id": ws.sent_ids[-1],
|
|
"result": {"result": {"value": value}}})
|
|
return _recv
|
|
|
|
|
|
class TestInspectRobustness(unittest.TestCase):
|
|
"""Regression tests for intermittent approval failures."""
|
|
|
|
def test_cdp_request_ids_unique(self):
|
|
# Millisecond-clock ids collide for rapid successive evaluates;
|
|
# a stale buffered response can then be misattributed to the
|
|
# wrong call (e.g. verify-after-click reads the click result).
|
|
# Frozen clock makes the old collision deterministic.
|
|
with mock.patch("approvals.time.time", return_value=1728000000.123):
|
|
ws = _FakeWS([_echo_last_value("a"), _echo_last_value("b")])
|
|
self.assertEqual(approvals.cdp_evaluate(ws, "1+1"), "a")
|
|
self.assertEqual(approvals.cdp_evaluate(ws, "2+2"), "b")
|
|
self.assertNotEqual(ws.sent_ids[0], ws.sent_ids[1])
|
|
|
|
def test_cdp_skips_stale_ids(self):
|
|
stale = json.dumps({"id": 999999999,
|
|
"result": {"result": {"value": "stale"}}})
|
|
ws = _FakeWS([stale, _echo_last_value("fresh")])
|
|
self.assertEqual(approvals.cdp_evaluate(ws, "1+1"), "fresh")
|
|
|
|
def test_unreachable_returns_full_shape(self):
|
|
with mock.patch.object(approvals, "get_node_pages",
|
|
side_effect=ConnectionError("nope")), \
|
|
mock.patch.object(approvals, "check_node_key_request",
|
|
return_value=None):
|
|
res = approvals.inspect_node_approvals("pip")
|
|
self.assertEqual(res["status"], "UNREACHABLE")
|
|
self.assertFalse(res["has_pending"])
|
|
for key in ("node", "title", "buttons", "is_trusted",
|
|
"input_waits", "error"):
|
|
self.assertIn(key, res)
|
|
|
|
def test_all_pages_failed_reports_error(self):
|
|
pages = [{"title": "t", "url": "u", "type": "page",
|
|
"webSocketDebuggerUrl": "ws://127.0.0.1:9/none"}]
|
|
|
|
class _DeadWSModule:
|
|
@staticmethod
|
|
def create_connection(*a, **k):
|
|
raise ConnectionError("refused")
|
|
|
|
with mock.patch.object(approvals, "get_node_pages",
|
|
return_value=pages), \
|
|
mock.patch.object(approvals, "websocket", _DeadWSModule()), \
|
|
mock.patch.object(approvals, "check_node_key_request",
|
|
return_value=None):
|
|
res = approvals.inspect_node_approvals("pip")
|
|
# Per-page CDP failures must surface as ERROR, never as a
|
|
# false CLEAR that hides pending approvals.
|
|
self.assertEqual(res["status"], "ERROR")
|
|
self.assertFalse(res["has_pending"])
|
|
self.assertIn("error", res)
|
|
|
|
def test_state_saves_roundtrip_without_leftovers(self):
|
|
# Guards the atomic-save refactor (tmp + replace): correct
|
|
# content and no stray temp files left behind.
|
|
with tempfile.TemporaryDirectory() as td:
|
|
rp = Path(td) / "resp.json"
|
|
with mock.patch.object(approvals, "RESPONDED_WAITS_FILE", rp):
|
|
approvals.save_responded_waits({"pip": {"t": "x"}})
|
|
self.assertEqual(json.loads(rp.read_text()),
|
|
{"pip": {"t": "x"}})
|
|
fp = Path(td) / "seen.json"
|
|
with mock.patch.object(approvals, "FIRST_SEEN_WAITS_FILE", fp):
|
|
approvals.save_first_seen_waits({"pip": {"t": "x"}})
|
|
self.assertEqual(json.loads(fp.read_text()),
|
|
{"pip": {"t": "x"}})
|
|
self.assertEqual(sorted(p.name for p in Path(td).iterdir()),
|
|
["resp.json", "seen.json"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|