box/tests/test_box_md_https.py

"""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")
box_ctl = _load("box_ctl_mdtest", "bin/box-ctl.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 _InProcResult:
    """Minimal CompletedProcess stand-in (returncode/stdout only)."""
    def __init__(self, returncode, stdout):
        self.returncode = returncode
        self.stdout = stdout


def _box_ctl_inproc(*args, stdin=None):
    """In-process _box_ctl for dry-run and validation-failure verbs.

    Calls the real main(argv) -- identical argv parsing, dispatch, audit,
    stdin reads, and stdout JSON -- with stdio captured, amortizing the
    ~80ms per-spawn interpreter + module-exec cost over one import.
    stdin, when given, is fed exactly as subprocess input= would be.
    """
    import io
    from contextlib import redirect_stdout
    buf = io.StringIO()
    returncode = 0
    saved_stdin = sys.stdin
    if stdin is not None:
        sys.stdin = io.StringIO(stdin)
    try:
        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
    finally:
        sys.stdin = saved_stdin
    return _InProcResult(returncode, buf.getvalue())


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"],
        ]
        # In-process dispatch: same main(argv) path, fail() JSON, and
        # exit code as a subprocess call. Assertions below are unchanged.
        for args in cases:
            r = _box_ctl_inproc(*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"],
        ]
        # In-process dispatch: same main(argv) path, stdin reads, fail()
        # JSON, and exit code as a subprocess call. Assertions below are
        # unchanged. (Amend/append parse --stdin BEFORE validating the
        # path, so stdin is fed here exactly as the spawn did.)
        for args in cases:
            r = _box_ctl_inproc(*args)
            self.assertNotEqual(r.returncode, 0, args)
            self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME", args)
        r = _box_ctl_inproc("md-amend", "../x", "--stdin", stdin="hi")
        self.assertEqual(json.loads(r.stdout)["code"], "BAD_NAME")
        r = _box_ctl_inproc("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:
            # In-process dry-run: same main(argv) path and stdout JSON as a
            # subprocess call, without the per-case spawn cost. Assertions
            # below are unchanged.
            r = _box_ctl_inproc("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()