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
413 lines
18 KiB
Python
413 lines
18 KiB
Python
"""Tests for md-file reads + governed writes over HTTPS (no SSH).
|
|
|
|
Covers the md expansion:
|
|
box-relay.sh (agent client) -> exec-constrained.py named ops
|
|
-> box-ctl.py backend verbs -> agent_md.py (Hatch gateway / shared templates).
|
|
|
|
Raw container writes (md-write) are deliberately NOT exposed over HTTPS;
|
|
the governed flows are amend/append (validated shared templates with git
|
|
commit) and pull/inject-drive/sync-all (push canonical templates).
|
|
|
|
Live execution is limited to validation-failure paths (which fail before
|
|
any gateway call or write), one stdin-plumbing path that the backend
|
|
safety gate rejects before writing, quality-validate dry-runs, and
|
|
agent_md validator unit tests. No live gateway traffic, no template
|
|
writes, and no live-socket round-trips here; instead we assert the exact
|
|
argv each op builds.
|
|
"""
|
|
import hashlib
|
|
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"
|
|
HEARTBEAT = REPO_ROOT / "shared" / "operators" / "HEARTBEAT.md"
|
|
|
|
MD_READ_OPS = ("md.audit", "md.list", "md.read", "md.diff")
|
|
MD_WRITE_OPS = ("md.pull", "md.inject_drive", "md.sync_all",
|
|
"md.amend", "md.append")
|
|
|
|
|
|
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_md", "bin/exec-constrained.py")
|
|
agent_md = _load("agent_md_mdtest", "bin/agent_md.py")
|
|
|
|
|
|
def _box_ctl(*args, stdin=None):
|
|
return subprocess.run(
|
|
[sys.executable, str(BOX_CTL), *args],
|
|
input=stdin, capture_output=True, text=True, timeout=180)
|
|
|
|
|
|
class ExecMdOpsTests(unittest.TestCase):
|
|
def test_ops_registered_and_side_effecting(self):
|
|
spec = exec_constrained.OPS
|
|
for op in MD_READ_OPS:
|
|
self.assertIn(op, spec)
|
|
self.assertFalse(spec[op]["side_effecting"], op)
|
|
for op in MD_WRITE_OPS:
|
|
self.assertIn(op, spec)
|
|
self.assertTrue(spec[op]["side_effecting"], op)
|
|
|
|
def test_raw_write_not_exposed(self):
|
|
self.assertNotIn("md.write", exec_constrained.OPS)
|
|
|
|
def test_known_identities_only(self):
|
|
p = exec_constrained.permitted
|
|
for op in MD_READ_OPS:
|
|
self.assertTrue(p("some-unknown-identity", op), op)
|
|
self.assertTrue(p("operator-646", op), op)
|
|
for op in MD_WRITE_OPS:
|
|
self.assertFalse(p("some-unknown-identity", op), op)
|
|
self.assertTrue(p("operator-646", op), op)
|
|
self.assertFalse(p("exec-canary", "md.read"))
|
|
self.assertFalse(p("exec-canary", "md.amend"))
|
|
|
|
def test_audit_validate(self):
|
|
v = exec_constrained.OPS["md.audit"]["validate"]
|
|
self.assertEqual(v({}), {"accounts": None})
|
|
# Explicit null means "all accounts", same as omitted (optional-arg
|
|
# convention shared with strat.set / loop.resolve).
|
|
self.assertEqual(v({"accounts": None}), {"accounts": None})
|
|
self.assertEqual(v({"accounts": ["646", "muse-main"]})["accounts"],
|
|
["646", "muse-main"])
|
|
for bad in ({"accounts": "646"}, {"accounts": []},
|
|
{"accounts": ["../x"]}, {"accounts": ["a b"]},
|
|
{"accounts": ["ok", ""]}, {"bogus": 1}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_list_validate(self):
|
|
v = exec_constrained.OPS["md.list"]["validate"]
|
|
self.assertEqual(v({"account": "646"}),
|
|
{"account": "646", "path": ""})
|
|
self.assertEqual(v({"account": "646", "path": "sub/dir"})["path"],
|
|
"sub/dir")
|
|
for bad in ({"account": "../x"}, {"account": "a b"},
|
|
{"account": "646", "path": ".."},
|
|
{"account": "646", "path": "/abs"},
|
|
{"account": "646", "path": "a/../../x"},
|
|
{"account": "646", "bogus": 1},
|
|
{"path": "sub"}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_read_validate(self):
|
|
v = exec_constrained.OPS["md.read"]["validate"]
|
|
good = v({"account": "646", "filename": "SOUL.md"})
|
|
self.assertEqual(good, {"account": "646", "filename": "SOUL.md"})
|
|
# Reads accept any container basename, not just shared templates.
|
|
self.assertEqual(
|
|
v({"account": "646", "filename": "NOTES.md"})["filename"],
|
|
"NOTES.md")
|
|
for bad in ({"account": "646", "filename": "../x"},
|
|
{"account": "646", "filename": "/abs"},
|
|
{"account": "646", "filename": ".."},
|
|
{"account": "646", "filename": "a/b"},
|
|
{"account": "646", "filename": ""},
|
|
{"account": "a b", "filename": "SOUL.md"},
|
|
{"account": "646", "filename": "SOUL.md", "x": 1}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_diff_validate(self):
|
|
v = exec_constrained.OPS["md.diff"]["validate"]
|
|
good = v({"account": "646", "filename": "SOUL.md"})
|
|
self.assertEqual(good, {"account": "646", "filename": "SOUL.md"})
|
|
# Template flows reject non-templates: the backend indexes
|
|
# shared/operators/ by filename.
|
|
for bad in ({"account": "646", "filename": "NOPE.md"},
|
|
{"account": "646", "filename": "../x"},
|
|
{"account": "646", "filename": "NOTES.md"}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_pull_validate(self):
|
|
v = exec_constrained.OPS["md.pull"]["validate"]
|
|
self.assertEqual(v({"account": "opm", "filename": "TOOLS.md"}),
|
|
{"account": "opm", "filename": "TOOLS.md"})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"account": "opm", "filename": "NOPE.md"})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"account": "../x", "filename": "TOOLS.md"})
|
|
|
|
def test_inject_drive_validate(self):
|
|
v = exec_constrained.OPS["md.inject_drive"]["validate"]
|
|
self.assertEqual(v({"account": "646"}), {"account": "646"})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"account": "../x"})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"account": "646", "force": True})
|
|
|
|
def test_sync_all_validate(self):
|
|
v = exec_constrained.OPS["md.sync_all"]["validate"]
|
|
self.assertEqual(v({}), {})
|
|
with self.assertRaises(exec_constrained.OpError):
|
|
v({"accounts": ["646"]})
|
|
|
|
def test_amend_validate(self):
|
|
v = exec_constrained.OPS["md.amend"]["validate"]
|
|
good = v({"filename": "SOUL.md", "content": "body",
|
|
"author": "646", "reason": "tune"})
|
|
self.assertEqual(good, {"filename": "SOUL.md", "content": "body",
|
|
"author": "646", "reason": "tune"})
|
|
defaults = v({"filename": "SOUL.md", "content": "body"})
|
|
self.assertEqual(defaults["author"], "operator")
|
|
self.assertEqual(defaults["reason"], "")
|
|
over = "x" * (exec_constrained.MD_MAX_AMEND + 1)
|
|
for bad in ({"filename": "NOPE.md", "content": "body"},
|
|
{"filename": "SOUL.md", "content": " "},
|
|
{"filename": "SOUL.md", "content": over},
|
|
{"filename": "SOUL.md", "content": "x", "author": ""},
|
|
{"filename": "SOUL.md", "content": "x",
|
|
"author": "a\nb"},
|
|
{"filename": "SOUL.md", "content": "x",
|
|
"author": "a" * 65},
|
|
{"filename": "SOUL.md", "content": "x",
|
|
"reason": "r" * 257},
|
|
{"filename": "SOUL.md", "content": "x",
|
|
"reason": "a\nb"},
|
|
{"filename": "SOUL.md", "content": "x", "bogus": 1},
|
|
{"filename": "SOUL.md"}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_append_validate(self):
|
|
v = exec_constrained.OPS["md.append"]["validate"]
|
|
good = v({"filename": "AGENTS.md", "text": "lesson",
|
|
"author": "opm", "section": "Wins"})
|
|
self.assertEqual(good["section"], "Wins")
|
|
self.assertIsNone(v({"filename": "AGENTS.md",
|
|
"text": "lesson"})["section"])
|
|
over = "x" * (exec_constrained.MD_MAX_APPEND + 1)
|
|
for bad in ({"filename": "NOPE.md", "text": "lesson"},
|
|
{"filename": "AGENTS.md", "text": " "},
|
|
{"filename": "AGENTS.md", "text": over},
|
|
{"filename": "AGENTS.md", "text": "t",
|
|
"section": " "},
|
|
{"filename": "AGENTS.md", "text": "t",
|
|
"section": "s" * 129}):
|
|
with self.assertRaises(exec_constrained.OpError, msg=bad):
|
|
v(bad)
|
|
|
|
def test_build_argv_shapes(self):
|
|
ops = exec_constrained.OPS
|
|
au = ops["md.audit"]
|
|
self.assertEqual(au["build"]({"accounts": None})[-1], "md-audit")
|
|
self.assertEqual(
|
|
au["build"]({"accounts": ["646", "opm"]})[-3:],
|
|
["md-audit", "646", "opm"])
|
|
li = ops["md.list"]
|
|
self.assertEqual(li["build"]({"account": "646", "path": ""})[-2:],
|
|
["md-list", "646"])
|
|
self.assertEqual(
|
|
li["build"]({"account": "646", "path": "sub"})[-3:],
|
|
["md-list", "646", "sub"])
|
|
rd = ops["md.read"]
|
|
self.assertEqual(
|
|
rd["build"]({"account": "646", "filename": "SOUL.md"})[-3:],
|
|
["md-read", "646", "SOUL.md"])
|
|
df = ops["md.diff"]
|
|
self.assertEqual(
|
|
df["build"]({"account": "646", "filename": "SOUL.md"})[-3:],
|
|
["md-diff", "646", "SOUL.md"])
|
|
pu = ops["md.pull"]
|
|
self.assertEqual(
|
|
pu["build"]({"account": "646", "filename": "SOUL.md"})[-3:],
|
|
["md-pull", "646", "SOUL.md"])
|
|
inj = ops["md.inject_drive"]
|
|
self.assertEqual(inj["build"]({"account": "646"})[-2:],
|
|
["md-inject-drive", "646"])
|
|
self.assertEqual(ops["md.sync_all"]["build"]({})[-1], "md-sync-all")
|
|
am = ops["md.amend"]
|
|
argv = am["build"]({"filename": "SOUL.md", "content": "x",
|
|
"author": "646", "reason": "why"})
|
|
self.assertEqual(argv[-7:],
|
|
["md-amend", "SOUL.md", "--stdin",
|
|
"--author", "646", "--reason", "why"])
|
|
argv = am["build"]({"filename": "SOUL.md", "content": "x",
|
|
"author": "646", "reason": ""})
|
|
self.assertEqual(argv[-5:],
|
|
["md-amend", "SOUL.md", "--stdin",
|
|
"--author", "646"])
|
|
ap = ops["md.append"]
|
|
argv = ap["build"]({"filename": "AGENTS.md", "text": "t",
|
|
"author": "opm", "section": "Wins"})
|
|
self.assertEqual(argv[-7:],
|
|
["md-append", "AGENTS.md", "--stdin",
|
|
"--author", "opm", "--section", "Wins"])
|
|
argv = ap["build"]({"filename": "AGENTS.md", "text": "t",
|
|
"author": "opm", "section": None})
|
|
self.assertEqual(argv[-5:],
|
|
["md-append", "AGENTS.md", "--stdin",
|
|
"--author", "opm"])
|
|
self.assertIsInstance(argv, list)
|
|
|
|
def test_stdin_body(self):
|
|
sb = exec_constrained._stdin_body
|
|
self.assertEqual(sb("md.amend", {"content": "C"}), "C")
|
|
self.assertEqual(sb("md.append", {"text": "T"}), "T")
|
|
self.assertIsNone(sb("md.read", {"account": "646"}))
|
|
|
|
|
|
class AgentMdValidationTests(unittest.TestCase):
|
|
def test_account(self):
|
|
for good in ("646", "muse", "muse-main", "opm", "dev", "def"):
|
|
self.assertEqual(agent_md.validate_account(good), good)
|
|
for bad in ("../x", "a b", "", "a/b", "x" * 33, "-lead"):
|
|
with self.assertRaises(agent_md.MDValidationError, msg=bad):
|
|
agent_md.validate_account(bad)
|
|
|
|
def test_filename(self):
|
|
for good in ("SOUL.md", "NOTES.md", "a"):
|
|
self.assertEqual(agent_md.validate_filename(good), good)
|
|
self.assertEqual(
|
|
agent_md.validate_filename("SOUL.md", template_only=True),
|
|
"SOUL.md")
|
|
for bad in ("../x", "/abs", "..", ".", "a/b", ""):
|
|
with self.assertRaises(agent_md.MDValidationError, msg=bad):
|
|
agent_md.validate_filename(bad)
|
|
for bad in ("NOPE.md", "../SOUL.md", "NOTES.md"):
|
|
with self.assertRaises(agent_md.MDValidationError, msg=bad):
|
|
agent_md.validate_filename(bad, template_only=True)
|
|
|
|
def test_subpath(self):
|
|
self.assertEqual(agent_md.validate_subpath(""), "")
|
|
self.assertEqual(agent_md.validate_subpath("a/b"), "a/b")
|
|
for bad in ("..", "/abs", "a/../../x", "a b"):
|
|
with self.assertRaises(agent_md.MDValidationError, msg=bad):
|
|
agent_md.validate_subpath(bad)
|
|
|
|
def test_rejects_before_gateway(self):
|
|
# Validation failures raise MDValidationError; a call that
|
|
# reached the gateway would raise RuntimeError (no gateway
|
|
# module here) or FileNotFoundError (no cookies) instead.
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.read_md("646", "../x")
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.list_files("../x", "")
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.audit_agents(["ok", "../x"])
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.diff_md("646", "NOPE.md")
|
|
|
|
def test_amend_rejects_before_write(self):
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.amend_md("../x", "body")
|
|
with self.assertRaises(agent_md.MDValidationError):
|
|
agent_md.append_md("NOPE.md", "note")
|
|
|
|
|
|
class BoxCtlMdTests(unittest.TestCase):
|
|
def test_rejects_traversal_before_gateway(self):
|
|
cases = [
|
|
["md-read", "646", "../x"],
|
|
["md-read", "646", ".."],
|
|
["md-list", "bad!", "x"],
|
|
["md-list", "646", "../.."],
|
|
["md-write", "646", "/abs", "hi"],
|
|
["md-audit", "../x"],
|
|
["md", "read", "646", "../x"],
|
|
]
|
|
for args in cases:
|
|
r = _box_ctl(*args)
|
|
self.assertNotEqual(r.returncode, 0, args)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME", args)
|
|
|
|
def test_rejects_non_template_before_write(self):
|
|
cases = [
|
|
["md-diff", "646", "NOPE.md"],
|
|
["md-pull", "646", "NOPE.md"],
|
|
["md", "diff", "646", "NOPE.md"],
|
|
["md", "amend", "NOPE.md", "content here"],
|
|
["md", "append", "NOPE.md", "note here"],
|
|
]
|
|
for args in cases:
|
|
r = _box_ctl(*args)
|
|
self.assertNotEqual(r.returncode, 0, args)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME", args)
|
|
r = _box_ctl("md-amend", "../x", "--stdin", stdin="hi")
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
r = _box_ctl("md-append", "../x", "--stdin", stdin="hi")
|
|
self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
|
|
|
|
def test_amend_stdin_safety_rejection_writes_nothing(self):
|
|
before = hashlib.sha256(HEARTBEAT.read_bytes()).hexdigest()
|
|
# Gutted HEARTBEAT content via --stdin: proves stdin plumbing
|
|
# reaches the backend, and the safety gate rejects it before
|
|
# any write or git commit.
|
|
r = _box_ctl("md-amend", "HEARTBEAT.md", "--stdin", stdin="gutted")
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertEqual(json.loads(r.stdout)["code"], "AMEND_FAILED")
|
|
after = hashlib.sha256(HEARTBEAT.read_bytes()).hexdigest()
|
|
self.assertEqual(before, after)
|
|
|
|
def test_quality_validate_md_verbs(self):
|
|
cases = [
|
|
(["md-audit"], True),
|
|
(["md-audit", "646", "opm"], True),
|
|
(["md-audit", "../x"], False),
|
|
(["md-list", "646"], True),
|
|
(["md-list", "646", "sub/dir"], True),
|
|
(["md-list", "646", ".."], False),
|
|
(["md-list"], False),
|
|
(["md-read", "646", "SOUL.md"], True),
|
|
(["md-read", "646", "../x"], False),
|
|
(["md-read", "646"], False),
|
|
(["md-diff", "646", "SOUL.md"], True),
|
|
(["md-diff", "646", "NOPE.md"], False),
|
|
(["md-pull", "646", "SOUL.md"], True),
|
|
(["md-pull", "646"], False),
|
|
(["md-inject-drive", "646"], True),
|
|
(["md-inject-drive", "646", "--force"], True),
|
|
(["md-inject-drive"], False),
|
|
(["md-sync-all"], True),
|
|
(["md-sync-all", "--force"], True),
|
|
(["md-sync-all", "646"], False),
|
|
(["md-amend", "SOUL.md", "--stdin"], True),
|
|
(["md-amend", "SOUL.md", "--stdin", "--author", "646"], True),
|
|
(["md-amend", "NOPE.md", "--stdin"], False),
|
|
(["md-amend"], False),
|
|
(["md-append", "SOUL.md", "--stdin"], True),
|
|
(["md-append", "SOUL.md", "note", "--section", "s"], True),
|
|
(["md-append", "x", "y", "z"], False),
|
|
(["md-write", "646", "SOUL.md", "x"], True),
|
|
(["md-write", "646", "SOUL.md"], False),
|
|
]
|
|
for args, valid in cases:
|
|
r = _box_ctl("quality-validate", *args)
|
|
self.assertEqual(json.loads(r.stdout)["valid"], valid, args)
|
|
|
|
|
|
class BoxRelayMdTests(unittest.TestCase):
|
|
def test_relay_help_lists_md_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 md audit", "box md list", "box md read",
|
|
"box md diff", "box md pull", "box md inject-drive",
|
|
"box md sync-all", "box md amend", "box md append"):
|
|
self.assertIn(line, r.stdout)
|
|
|
|
def test_relay_maps_md_commands_to_ops(self):
|
|
text = RELAY.read_text()
|
|
for op in ('"md.audit"', '"md.list"', '"md.read"', '"md.diff"',
|
|
'"md.pull"', '"md.inject_drive"', '"md.sync_all"',
|
|
'"md.amend"', '"md.append"'):
|
|
self.assertIn(op, text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|