feat(box): passkey fetch, agent key-approval flow, unified lookups, tmux agent UX

- box passkey [show|fetch] (+ muse passkey): documents VM-only passkey
  (/srv/box/passkey.txt, fallback /etc/netvm/passkey.txt on 34.139.37.135),
  probes VM over SSH with graceful fallback; --json supported. No secrets on bl.
- approvals: request_key_approval / check_node_key_request; KEY_APPROVAL status
  surfaced in `box approvals check`; allow/deny resolve + audit to box-ctl.jsonl;
  never auto-approved. New `box approvals request-key <node> --reason`.
- box lookup (summary|fleet|threads|unread|approvals|key|docs) and docs-lookup
  engine with lookup_internal/ database (docs_internal symlink).
- muse-tmux: non-TTY attach falls back to scrollback capture; prune NameError fix.
- box/muse passthrough for tmux/muse/docs; thread list/view alias + prefix resolve.
- Docs: AGENTS.md, AGENT-TOOLING.md, BOX-WEB-SURFACE-GUIDE.md, README.
- Tests: key-approval + passkey tests; sync stale sidechat UUIDs and manifest name.
- .gitignore runtime trackers (subagent-sessions, conversation-nudge-tracker).
This commit is contained in:
operator
2026-10-05 20:18:47 +00:00
parent 59c9965791
commit adfcd2e602
38 changed files with 5948 additions and 168 deletions
+119
View File
@@ -29,6 +29,23 @@ class TestApprovalsModule(unittest.TestCase):
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")
@@ -46,6 +63,7 @@ class TestApprovalsModule(unittest.TestCase):
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"])
@@ -124,5 +142,106 @@ class TestMuseChatApiConnection(unittest.TestCase):
self.assertIn("No pending approvals", 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")
if __name__ == "__main__":
unittest.main()