"""Tests for approvals over HTTPS (no SSH). Covers the approvals expansion: box-relay.sh (agent client) -> exec-constrained.py named ops -> box-ctl.py backend verbs -> approvals.py (fleet CDP). Live execution is limited to validation-failure paths (BAD_NODE/BAD_ARGS, which fail before any CDP probe) plus quality-validate dry-runs. No live browser traffic and no live-socket round-trips here; instead we assert the exact argv each op builds. In particular the allow build must never carry --always/--force: remote allow is one-shot only. """ import importlib.util import json import subprocess import sys import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent BOX_CTL = REPO_ROOT / "bin" / "box-ctl.py" RELAY = REPO_ROOT / "bin" / "box-relay.sh" def _load(name, relpath): spec = importlib.util.spec_from_file_location(name, REPO_ROOT / relpath) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod exec_constrained = _load("exec_constrained_approvals", "bin/exec-constrained.py") box_ctl = _load("box_ctl_approvalstest", "bin/box-ctl.py") def _box_ctl(*args): return subprocess.run( [sys.executable, str(BOX_CTL), *args], capture_output=True, text=True, timeout=180) class _InProcResult: """Minimal CompletedProcess stand-in (returncode/stdout only).""" def __init__(self, returncode, stdout): self.returncode = returncode self.stdout = stdout def _box_ctl_inproc(*args): """In-process _box_ctl (same proven pattern as test_box_loop_https). Calls the real main(argv) -- identical argv parsing, dispatch, audit, and stdout JSON -- amortizing per-spawn interpreter cost over one import. Node order in fleet scans is nondeterministic either way (concurrent fan-out); per-node content is identical. """ import io from contextlib import redirect_stdout buf = io.StringIO() returncode = 0 with redirect_stdout(buf): try: box_ctl.main(["box-ctl.py", *args]) except SystemExit as e: returncode = e.code if isinstance(e.code, int) else 1 return _InProcResult(returncode, buf.getvalue()) class ExecApprovalOpsTests(unittest.TestCase): def test_ops_registered_and_side_effecting(self): spec = exec_constrained.OPS self.assertIn("approval.check", spec) self.assertFalse(spec["approval.check"]["side_effecting"]) for op in ("approval.deny", "approval.auto", "approval.allow"): self.assertIn(op, spec) self.assertTrue(spec[op]["side_effecting"], op) def test_known_identities_only(self): p = exec_constrained.permitted self.assertTrue(p("some-unknown-identity", "approval.check")) for op in ("approval.deny", "approval.auto", "approval.allow"): self.assertFalse(p("some-unknown-identity", op), op) self.assertTrue(p("operator-646", op), op) self.assertFalse(p("exec-canary", "approval.check")) def test_check_validate(self): v = exec_constrained.OPS["approval.check"]["validate"] self.assertEqual(v({}), {"node": None}) self.assertEqual(v({"node": None}), {"node": None}) self.assertEqual(v({"node": "646"}), {"node": "646"}) for bad in ({"node": "nope"}, {"node": "../x"}, {"node": "646", "bogus": 1}): with self.assertRaises(exec_constrained.OpError, msg=bad): v(bad) def test_deny_validate(self): v = exec_constrained.OPS["approval.deny"]["validate"] good = v({"node": "646", "message": "not trusted", "allow_main_chat": True}) self.assertEqual(good, {"node": "646", "message": "not trusted", "allow_main_chat": True}) self.assertFalse(v({"node": "646", "message": "m"})["allow_main_chat"]) # Multiline explanations are fine; other controls are not. v({"node": "646", "message": "line1\nline2"}) over = "x" * (exec_constrained.MAX_MESSAGE + 1) for bad in ({"node": "646"}, {"node": "646", "message": " "}, {"node": "646", "message": over}, {"node": "646", "message": "a\x07b"}, {"node": "nope", "message": "m"}, {"node": "646", "message": "m", "allow_main_chat": "yes"}, {"node": "646", "message": "m", "force": True}): with self.assertRaises(exec_constrained.OpError, msg=bad): v(bad) def test_auto_validate(self): v = exec_constrained.OPS["approval.auto"]["validate"] self.assertEqual(v({}), {"node": None}) self.assertEqual(v({"node": "opm"}), {"node": "opm"}) with self.assertRaises(exec_constrained.OpError): v({"node": "nope"}) with self.assertRaises(exec_constrained.OpError): v({"node": "646", "always": True}) def test_allow_validate(self): v = exec_constrained.OPS["approval.allow"]["validate"] good = v({"node": "646", "message": "trusted deploy script"}) self.assertEqual(good["message"], "trusted deploy script") self.assertFalse(good["allow_main_chat"]) # Attribution is mandatory: the flow notifies the waiting agent, # so a remote allow must carry its reason. with self.assertRaises(exec_constrained.OpError): v({"node": "646"}) with self.assertRaises(exec_constrained.OpError): v({"node": "646", "message": " "}) with self.assertRaises(exec_constrained.OpError): v({"node": "nope", "message": "m"}) # No persistence/force remotely, not even as rejected keys. with self.assertRaises(exec_constrained.OpError): v({"node": "646", "message": "m", "always": True}) with self.assertRaises(exec_constrained.OpError): v({"node": "646", "message": "m", "force": True}) def test_build_argv_shapes(self): ops = exec_constrained.OPS ck = ops["approval.check"] self.assertEqual(ck["build"]({"node": None})[-1], "approval-check") self.assertEqual(ck["build"]({"node": "646"})[-2:], ["approval-check", "646"]) de = ops["approval.deny"] argv = de["build"]({"node": "646", "message": "m", "allow_main_chat": False}) self.assertEqual(argv[-4:], ["approval-deny", "646", "--message", "m"]) argv = de["build"]({"node": "646", "message": "m", "allow_main_chat": True}) self.assertEqual(argv[-1], "--allow-main-chat") au = ops["approval.auto"] self.assertEqual(au["build"]({"node": None})[-1], "approval-auto") self.assertEqual(au["build"]({"node": "opm"})[-2:], ["approval-auto", "opm"]) al = ops["approval.allow"] argv = al["build"]({"node": "646", "message": "m", "allow_main_chat": False}) self.assertEqual(argv[-4:], ["approval-allow", "646", "--message", "m"]) self.assertNotIn("--always", argv) self.assertNotIn("--force", argv) argv = al["build"]({"node": "646", "message": "m", "allow_main_chat": True}) self.assertEqual(argv[-1], "--allow-main-chat") self.assertIsInstance(argv, list) class BoxCtlApprovalTests(unittest.TestCase): def test_rejects_unknown_node_before_cdp(self): for args in (["approval-check", "badnode"], ["approval-list", "badnode"], ["approval-allow", "badnode", "--message", "m"], ["approval-deny", "badnode", "--message", "m"], ["approval-auto", "badnode"]): r = _box_ctl_inproc(*args) self.assertNotEqual(r.returncode, 0, args) self.assertEqual(json.loads(r.stdout)["code"], "BAD_NODE", args) def test_rejects_missing_node(self): for args in (["approval-allow"], ["approval-deny"]): r = _box_ctl_inproc(*args) self.assertNotEqual(r.returncode, 0, args) self.assertEqual(json.loads(r.stdout)["code"], "BAD_ARGS", args) def test_quality_validate_approval_verbs(self): # Note: box-ctl leaves --message optional (SSH callers may rely on # the flow default); the exec layer is the narrower gate and # requires it. quality-validate mirrors box-ctl. cases = [ (["approval-check"], True), (["approval-check", "646"], True), (["approval-check", "nope"], False), (["approval-check", "646", "opm"], False), (["approval-list", "646"], True), (["approval-allow", "646", "--message", "hi"], True), (["approval-allow", "646"], True), (["approval-allow"], False), (["approval-allow", "nope", "--message", "x"], False), (["approval-allow", "646", "--always", "--force", "--message", "x"], True), (["approval-approve", "646", "--message", "x"], True), (["approval-deny", "646", "--message", "x"], True), (["approval-deny", "646", "--message", "x", "--allow-main-chat"], True), (["approval-deny"], False), (["approval-auto"], True), (["approval-auto", "646"], True), (["approval-auto", "nope"], False), (["approval-auto", "a", "b"], False), ] for args, valid in cases: r = _box_ctl_inproc("quality-validate", *args) self.assertEqual(json.loads(r.stdout)["valid"], valid, args) class BoxRelayApprovalTests(unittest.TestCase): def test_relay_help_lists_approval_commands(self): r = subprocess.run(["bash", str(RELAY), "help"], capture_output=True, text=True, timeout=30) self.assertEqual(r.returncode, 0, r.stderr) for line in ("box approvals check", "box approvals allow", "box approvals deny", "box approvals auto"): self.assertIn(line, r.stdout) def test_relay_maps_approval_commands_to_ops(self): text = RELAY.read_text() for op in ('"approval.check"', '"approval.deny"', '"approval.auto"', '"approval.allow"'): self.assertIn(op, text) if __name__ == "__main__": unittest.main()