#!/usr/bin/env python3 """contract_diff: detect a permission deadlock and emit a ContractDiff plus a narrow authorization REQUEST addressed to a human role. stdlib only. This tool never grants anything. It reads three JSON files and prints JSON. Exit codes: 0 OK, 3 blocked, 2 usage/input error. """ import argparse import hashlib import json import re import sys HUMAN_ROLES = frozenset({ "clinic_manager", "compliance_officer", "data_protection_officer", "human_owner", "it_administrator", "medical_director", "security_owner", }) AGENT_ROLE_RE = re.compile(r"(agent|llm|bot|assistant|model|claude|gpt|\bai\b|auto)", re.I) WILDCARD_CHARS = set("*?[]{}") WILDCARD_WORDS = {"all", "any", "everything", "anything"} MAX_TTL_MINUTES = 60 DEFAULT_TTL_MINUTES = 15 class InputError(Exception): pass def has_wildcard(value): if not isinstance(value, str) or not value.strip(): return True # empty / non-string cannot be a single exact scope return bool(WILDCARD_CHARS & set(value)) or value.strip().lower() in WILDCARD_WORDS def is_human_role(role): return isinstance(role, str) and role in HUMAN_ROLES and not AGENT_ROLE_RE.search(role) def dumps(obj): return json.dumps(obj, sort_keys=True, indent=2, ensure_ascii=False) + "\n" def _load(path): try: with open(path, "r", encoding="utf-8") as fh: return json.load(fh) except (OSError, ValueError) as exc: raise InputError("cannot read %s: %s" % (path, exc)) def _right_key(r, where): if not isinstance(r, dict) or "action" not in r or "resource" not in r: raise InputError("%s: right needs action and resource" % where) if has_wildcard(r["action"]) or has_wildcard(r["resource"]): raise InputError("%s: wildcard or empty scope is not allowed: %r" % (where, r)) return (r["action"], r["resource"]) def _rule_matches(rule, action, resource): # policy rules may use a lone "*" to describe broad policy; requests never may return rule.get("action") in (action, "*") and rule.get("resource") in (resource, "*") def compute(contract, policy, grants): if not isinstance(contract, dict) or not isinstance(contract.get("required"), list): raise InputError("contract.required must be a list") role = contract.get("human_role") if not is_human_role(role): raise InputError("contract.human_role must be one of %s" % sorted(HUMAN_ROLES)) rules = policy.get("rules") if isinstance(policy, dict) else None if not isinstance(rules, list): raise InputError("policy.rules must be a list") for rule in rules: if rule.get("effect") not in ("allow", "deny") or not isinstance(rule.get("priority"), int) \ or not rule.get("id"): raise InputError("bad rule: %r" % (rule,)) granted = set() for r in (grants.get("rights") if isinstance(grants, dict) else None) or []: # a wildcard grant is never honoured as coverage for an exact right if isinstance(r, dict) and not has_wildcard(r.get("action")) and not has_wildcard(r.get("resource")): granted.add((r["action"], r["resource"])) items = sorted(contract["required"], key=lambda i: str(i.get("id"))) verified = [i["id"] for i in items if i.get("evidenced") is True] gap_items = [i for i in items if i.get("evidenced") is not True] gaps = [{"id": i["id"], "description": i.get("description", ""), "required_rights": sorted([{"action": a, "resource": b} for a, b in {_right_key(r, "item " + str(i["id"])) for r in i.get("required_rights", [])}], key=lambda x: (x["action"], x["resource"]))} for i in gap_items] needed = {} for g in gaps: for r in g["required_rights"]: needed.setdefault((r["action"], r["resource"]), []).append(g["id"]) missing, contradictions, blocking = [], [], None for (action, resource) in sorted(needed): matching = [r for r in rules if _rule_matches(r, action, resource)] top = max((r["priority"] for r in matching), default=None) at_top = sorted((r for r in matching if r["priority"] == top), key=lambda r: r["id"]) allows = [r for r in at_top if r["effect"] == "allow"] denies = [r for r in at_top if r["effect"] == "deny"] reason = None if allows and denies: contradictions.append({"action": action, "resource": resource, "priority": top, "allow_rule_ids": [r["id"] for r in allows], "deny_rule_ids": [r["id"] for r in denies]}) if blocking is None: d = denies[0] blocking = {"kind": "contradiction", "rule_id": d["id"], "text": d.get("text", ""), "source": d.get("source", ""), "conflicts_with": [r["id"] for r in allows]} elif denies: reason = "policy_deny" if blocking is None: d = denies[0] blocking = {"kind": "deny", "rule_id": d["id"], "text": d.get("text", ""), "source": d.get("source", ""), "conflicts_with": []} if reason is None and (action, resource) not in granted: reason = "not_granted" if reason: missing.append({"action": action, "resource": resource, "reason": reason, "needed_by": sorted(needed[(action, resource)])}) if missing and contradictions: verdict = "BLOCKED_BOTH" elif contradictions: verdict = "BLOCKED_POLICY_CONTRADICTION" elif missing: verdict = "BLOCKED_MISSING_RIGHTS" else: verdict = "OK" request = None if verdict != "OK": target = missing[0] if missing else contradictions[0] ttl = contract.get("requested_ttl_minutes", DEFAULT_TTL_MINUTES) if not isinstance(ttl, int) or isinstance(ttl, bool) or ttl < 1: ttl = DEFAULT_TTL_MINUTES ttl = min(ttl, MAX_TTL_MINUTES) core = {"action": target["action"], "resource": target["resource"], "ttl_minutes": ttl, "addressed_to_role": role, "contract_id": contract.get("contract_id", "")} rid = hashlib.sha256(json.dumps(core, sort_keys=True).encode()).hexdigest()[:16] why = ("Task blocked (%s). Needs exactly one right: %s on %s." % (verdict, target["action"], target["resource"])) if blocking: why += " Blocking policy %s (%s): %s" % (blocking["rule_id"], blocking["source"], blocking["text"]) request = dict(core, request_id=rid, one_time=True, granted=False, status="PENDING_HUMAN_DECISION", reason=why, remaining_blocked_rights=len({(m["action"], m["resource"]) for m in missing} | {(c["action"], c["resource"]) for c in contradictions}) - 1, note="Request only. The agent must stop and wait. Nothing is authorized by this document.") _assert_request_safe(request) return {"outcome_required": [i["id"] for i in items], "outcome_verified": verified, "gaps": gaps, "blocking_policy": blocking, "missing_rights": missing, "contradictions": contradictions, "verdict": verdict, "authorization_request": request} def _assert_request_safe(req): if has_wildcard(req["action"]) or has_wildcard(req["resource"]): raise InputError("refusing to emit wildcard request") if not (1 <= req["ttl_minutes"] <= MAX_TTL_MINUTES) or req["one_time"] is not True: raise InputError("refusing to emit unbounded request") if not is_human_role(req["addressed_to_role"]) or req["granted"] is not False: raise InputError("request must be addressed to a human role and ungranted") def main(argv=None): ap = argparse.ArgumentParser(description="Deadlock ContractDiff (never grants anything).") ap.add_argument("--contract", required=True) ap.add_argument("--policy", required=True) ap.add_argument("--grants", required=True) try: args = ap.parse_args(argv) except SystemExit: return 2 try: out = compute(_load(args.contract), _load(args.policy), _load(args.grants)) except (InputError, KeyError, TypeError, AttributeError) as exc: sys.stderr.write("error: %s\n" % exc) return 2 sys.stdout.write(dumps(out)) return 0 if out["verdict"] == "OK" else 3 if __name__ == "__main__": sys.exit(main())