import itertools import json import os import re import subprocess import sys import unittest from datetime import datetime, timezone ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(ROOT, "verifier")) import check_mandate as cm # noqa: E402 import contract_diff as cd # noqa: E402 FX = os.path.join(ROOT, "fixtures") NOW = "2026-11-11T12:00:00Z" NOW_DT = datetime(2026, 11, 11, 12, 0, 0, tzinfo=timezone.utc) def run(script, *args): p = subprocess.run([sys.executable, os.path.join(ROOT, "verifier", script)] + list(args), capture_output=True, timeout=30) return p.returncode, p.stdout def diff_args(case): d = os.path.join(FX, case) return ["--contract", d + "/contract.json", "--policy", d + "/policy.json", "--grants", d + "/grants.json"] def diff(case): code, out = run("contract_diff.py", *diff_args(case)) return code, json.loads(out) def jload(name): with open(os.path.join(FX, name)) as fh: return json.load(fh) class ContractDiffTests(unittest.TestCase): def test_determinism(self): for case in ("contradictory_policy", "missing_rights", "ok_case"): a = run("contract_diff.py", *diff_args(case)) b = run("contract_diff.py", *diff_args(case)) self.assertEqual(a, b) def test_ok_case(self): code, d = diff("ok_case") self.assertEqual((code, d["verdict"]), (0, "OK")) self.assertIsNone(d["authorization_request"]) self.assertEqual(d["outcome_verified"], ["O1"]) self.assertEqual([g["id"] for g in d["gaps"]], ["O2"]) def test_missing_rights(self): code, d = diff("missing_rights") self.assertEqual((code, d["verdict"]), (3, "BLOCKED_MISSING_RIGHTS")) self.assertEqual(d["missing_rights"][0]["reason"], "not_granted") self.assertEqual(d["outcome_verified"], []) def test_contradiction(self): code, d = diff("contradictory_policy") self.assertEqual((code, d["verdict"]), (3, "BLOCKED_POLICY_CONTRADICTION")) self.assertEqual(d["contradictions"][0]["allow_rule_ids"], ["P-ALLOW-7"]) self.assertEqual(d["contradictions"][0]["deny_rule_ids"], ["P-DENY-9"]) self.assertEqual(d["blocking_policy"]["rule_id"], "P-DENY-9") self.assertEqual(d["blocking_policy"]["source"], "synthetic security policy s.2") def test_both(self): d = json.loads(json.dumps(jload("contradictory_policy/contract.json"))) out = cd.compute(d, jload("contradictory_policy/policy.json"), {"rights": []}) self.assertEqual(out["verdict"], "BLOCKED_BOTH") def test_higher_priority_resolves_conflict(self): pol = jload("contradictory_policy/policy.json") pol["rules"][1]["priority"] = 60 # deny now wins: plain deny, not contradiction out = cd.compute(jload("contradictory_policy/contract.json"), pol, jload("contradictory_policy/grants.json")) self.assertEqual(out["verdict"], "BLOCKED_MISSING_RIGHTS") self.assertEqual(out["missing_rights"][0]["reason"], "policy_deny") self.assertEqual(out["contradictions"], []) def test_request_is_narrow_and_human_addressed(self): for case in ("contradictory_policy", "missing_rights"): _, d = diff(case) req = d["authorization_request"] self.assertFalse(set(req["action"] + req["resource"]) & set("*?[]{}")) self.assertLessEqual(req["ttl_minutes"], 60) self.assertTrue(req["one_time"]) self.assertIs(req["granted"], False) self.assertTrue(cd.is_human_role(req["addressed_to_role"])) for k in ("signature", "signed_by", "signed_by_human", "signer_role"): self.assertNotIn(k, req) self.assertEqual(diff("missing_rights")[1]["authorization_request"]["ttl_minutes"], 60) # 120 capped def test_wildcard_input_rejected(self): c = jload("missing_rights/contract.json") c["required"][0]["required_rights"][0]["resource"] = "synthetic://archive/*" with self.assertRaises(cd.InputError): cd.compute(c, jload("missing_rights/policy.json"), jload("missing_rights/grants.json")) def test_wildcard_grant_not_honoured(self): out = cd.compute(jload("missing_rights/contract.json"), jload("missing_rights/policy.json"), {"rights": [{"action": "*", "resource": "*"}]}) self.assertEqual(out["verdict"], "BLOCKED_MISSING_RIGHTS") def test_agent_role_addressee_rejected(self): c = jload("missing_rights/contract.json") c["human_role"] = "llm_supervisor_agent" with self.assertRaises(cd.InputError): cd.compute(c, jload("missing_rights/policy.json"), jload("missing_rights/grants.json")) def test_usage_exit_code(self): self.assertEqual(run("contract_diff.py")[0], 2) self.assertEqual(run("contract_diff.py", "--contract", "/nonexistent", "--policy", "x", "--grants", "y")[0], 2) class MandateTests(unittest.TestCase): def setUp(self): _, d = diff("missing_rights") self.request = d["authorization_request"] self.reqfile = "/tmp/_dlb_request_%d.json" % os.getpid() with open(self.reqfile, "w") as fh: json.dump(d, fh) self.addCleanup(os.unlink, self.reqfile) def cli(self, mandate, *extra): code, out = run("check_mandate.py", "--request", self.reqfile, "--mandate", os.path.join(FX, mandate), "--used-nonces", os.path.join(FX, "used_nonces.json"), "--now", NOW, *extra) return code, json.loads(out) def test_good_accepted(self): code, out = self.cli("mandate_good.json") self.assertEqual((code, out["decision"], out["reasons"]), (0, "ACCEPT", [])) def test_bad_cases_rejected(self): expect = {"mandate_bad_wildcard.json": "wildcard", "mandate_expired.json": "expired", "mandate_wrong_resource.json": "resource does not exactly match"} for name, frag in expect.items(): code, out = self.cli(name) self.assertEqual((code, out["decision"]), (3, "REJECT"), name) self.assertTrue(any(frag in r for r in out["reasons"]), (name, out["reasons"])) def test_deterministic(self): self.assertEqual(run("check_mandate.py", "--request", self.reqfile, "--mandate", FX + "/mandate_good.json", "--now", NOW), run("check_mandate.py", "--request", self.reqfile, "--mandate", FX + "/mandate_good.json", "--now", NOW)) def test_used_nonce_rejected(self): m = jload("mandate_good.json") r = cm.check_mandate(self.request, m, {m["nonce"]}, NOW_DT) self.assertEqual(r["decision"], "REJECT") self.assertIn("nonce already used", r["reasons"]) def test_ttl_over_cap_and_over_request(self): m = jload("mandate_good.json") m["ttl_minutes"] = 61 self.assertEqual(cm.check_mandate(self.request, m, (), NOW_DT)["decision"], "REJECT") m["ttl_minutes"] = 30 m["expires_at"] = "2026-11-11T13:00:00Z" self.assertEqual(cm.check_mandate(self.request, m, (), NOW_DT)["decision"], "REJECT") def test_one_time_and_request_id(self): m = jload("mandate_good.json") m["one_time"] = False self.assertEqual(cm.check_mandate(self.request, m, (), NOW_DT)["decision"], "REJECT") m = jload("mandate_good.json") m["request_id"] = "0000000000000000" self.assertEqual(cm.check_mandate(self.request, m, (), NOW_DT)["decision"], "REJECT") def test_usage_exit_code(self): self.assertEqual(run("check_mandate.py")[0], 2) self.assertEqual(run("check_mandate.py", "--request", self.reqfile, "--mandate", FX + "/mandate_good.json", "--now", "garbage")[0], 2) def test_never_accept_without_human_signer(self): good = jload("mandate_good.json") signed = [False, None, "true", 1, 0, "yes", [], {}, "MISSING"] roles = ["agent", "llm", "assistant", "claude", "gpt-4", "ai", "bot", "auto_approver", "model_supervisor", "llm_agent", "", None, 5, "MISSING", "Security_Owner", "random_human"] for s, r in itertools.product(signed + [True], roles + ["security_owner"]): m = dict(good) for key, val in (("signed_by_human", s), ("signer_role", r)): if val == "MISSING": m.pop(key) else: m[key] = val res = cm.check_mandate(self.request, m, (), NOW_DT) human_ok = s is True and r == "security_owner" self.assertEqual(res["decision"] == "ACCEPT", human_ok, (s, r)) def test_signature_note_required(self): m = jload("mandate_good.json") m["signature_note"] = " " self.assertEqual(cm.check_mandate(self.request, m, (), NOW_DT)["decision"], "REJECT") def test_code_has_no_subprocess_or_network(self): for f in ("contract_diff.py", "check_mandate.py"): with open(os.path.join(ROOT, "verifier", f)) as fh: src = fh.read() self.assertIsNone(re.search(r"^\s*(import|from)\s+(subprocess|socket|urllib|http|requests)", src, re.M), f) class SkillTests(unittest.TestCase): def test_frontmatter_name_matches_dir(self): path = os.path.join(ROOT, "skills", "deadlock-breaker", "SKILL.md") with open(path) as fh: text = fh.read() m = re.match(r"^---\n(.*?)\n---\n", text, re.S) self.assertTrue(m) fm = dict(line.split(": ", 1) for line in m.group(1).splitlines()) self.assertEqual(set(fm), {"name", "description"}) self.assertEqual(fm["name"], os.path.basename(os.path.dirname(path))) self.assertIn("not an enforcement gate", text) def test_manifest_basics(self): with open(os.path.join(ROOT, "manifest.json")) as fh: m = json.load(fh) self.assertEqual((m["schema"], m["name"], m["version"], m["license"], m["tier"]), ("carehab-duo/1", "deadlock-breaker", "0.1.0", "MIT", "open")) self.assertEqual(m["tools_permissions"]["skill_allowed_tools"], ["Bash", "Read"]) self.assertEqual(m["tools_permissions"]["verifier"]["network"], "none") if __name__ == "__main__": unittest.main()