219 lines
9.5 KiB
Python
219 lines
9.5 KiB
Python
|
"""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")
|
||
|
|
|
||
|
|
|
||
|
|
def _box_ctl(*args):
|
||
|
|
return subprocess.run(
|
||
|
|
[sys.executable, str(BOX_CTL), *args],
|
||
|
|
capture_output=True, text=True, timeout=180)
|
||
|
|
|
||
|
|
|
||
|
|
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(*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(*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("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()
|