"""Version 1.0, 2026-09-09. Local policy example, not an auth service. Python 3.10+, standard library only. No network, secrets, or file writes. Trusted actor/approval state is supplied by this test harness, never a model. """ from copy import deepcopy from hashlib import sha256 import json import unittest class Denied(Exception): pass class Documents: def __init__(self): self.docs = { "DOC-001": {"tenant": "amber", "readers": {"maya"}, "editors": {"maya"}, "version": 1, "body": "Draft policy"}, "DOC-002": {"tenant": "birch", "readers": {"ben"}, "editors": {"ben"}, "version": 1, "body": "Birch policy"}, } self.active = {"maya", "ben", "lee"} self.approvers = {"amber": {"lee"}, "birch": set()} self.approvals = {} self.enabled = True def validate(self, actor, request): if (type(request) is not dict or type(request.get("action")) is not str or request["action"] not in {"read", "write"}): raise Denied("invalid_request") fields = {"action", "document_id"} if request["action"] == "write": fields |= {"body", "expected_version"} if set(request) != fields or type(request["document_id"]) is not str: raise Denied("invalid_request") doc = self.docs.get(request["document_id"]) if (not self.enabled or actor["user"] not in self.active or not doc or actor["tenant"] != doc["tenant"]): raise Denied("access_denied") acl = "readers" if request["action"] == "read" else "editors" if actor["user"] not in doc[acl]: raise Denied("access_denied") if request["action"] == "write": body = request["body"] if type(body) is not str or not body.strip() or len(body.encode("utf-8")) > 4096: raise Denied("invalid_body") if type(request["expected_version"]) is not int or request["expected_version"] != doc["version"]: raise Denied("stale_version") return doc @staticmethod def digest(actor, request): encoded = json.dumps({"actor": actor, "request": request}, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") return sha256(encoded).hexdigest() def approve(self, reviewer, actor, request, now): # Called by a trusted reviewer surface, not an exposed agent tool. self.validate(actor, request) if (request["action"] != "write" or reviewer["tenant"] != actor["tenant"] or reviewer["user"] not in self.active or reviewer["user"] not in self.approvers.get(actor["tenant"], set()) or reviewer["user"] == actor["user"]): raise Denied("reviewer_denied") approval_id = f"approval-{len(self.approvals) + 1}" self.approvals[approval_id] = {"digest": self.digest(actor, request), "reviewer": deepcopy(reviewer), "expires": now + 60, "used": False} return approval_id def execute(self, actor, request, now, approval_id=None): doc = self.validate(actor, request) # current ACL and version, even after approval if request["action"] == "read": return {"body": doc["body"], "version": doc["version"]} approval = self.approvals.get(approval_id) if (not approval or approval["used"] or now >= approval["expires"] or approval["digest"] != self.digest(actor, request)): raise Denied("approval_denied") reviewer = approval["reviewer"] if (reviewer["user"] not in self.active or reviewer["user"] not in self.approvers.get(actor["tenant"], set())): raise Denied("reviewer_denied") # A production implementation needs an atomic transaction for the # version check, approval consumption, and write. This is single-threaded. approval["used"] = True doc["body"] = request["body"] doc["version"] += 1 return {"version": doc["version"]} class PolicyTests(unittest.TestCase): def setUp(self): self.store = Documents() self.actor = {"user": "maya", "tenant": "amber"} self.reviewer = {"user": "lee", "tenant": "amber"} self.request = {"action": "write", "document_id": "DOC-001", "body": "Revised policy", "expected_version": 1} def approval(self): return self.store.approve(self.reviewer, self.actor, self.request, 100) def test_read(self): self.assertEqual(self.store.execute(self.actor, {"action": "read", "document_id": "DOC-001"}, 100)["version"], 1) def test_cross_tenant_read_and_write(self): for action in ("read", "write"): req = {"action": action, "document_id": "DOC-002"} if action == "write": req.update(body="change", expected_version=1) with self.assertRaises(Denied): self.store.execute(self.actor, req, 100) def test_approved_write(self): self.assertEqual(self.store.execute(self.actor, self.request, 101, self.approval()), {"version": 2}) def test_no_approval(self): with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 100) def test_altered_body(self): approval = self.approval() self.request["body"] = "Unreviewed change" with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 101, approval) def test_unknown_fields_and_types(self): for extra in ({"tenant": "birch"}, {"approved": True}, {"intent": "urgent"}, {"action": []}, {"action": {}}, {"document_id": 123}, {"body": 123}, {"body": "x" * 4097}): with self.assertRaises(Denied): self.store.execute(self.actor, self.request | extra, 100) def test_expired_approval(self): with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 160, self.approval()) def test_replay(self): approval = self.approval() self.store.execute(self.actor, self.request, 101, approval) self.request["expected_version"] = 2 with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 102, approval) def test_acl_revocation(self): approval = self.approval() self.store.docs["DOC-001"]["editors"].clear() with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 101, approval) def test_subject_revocation(self): approval = self.approval() self.store.active.remove("maya") with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 101, approval) def test_reviewer_revocation(self): approval = self.approval() self.store.approvers["amber"].clear() with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 101, approval) def test_self_approval_denied(self): with self.assertRaises(Denied): self.store.approve(self.actor, self.actor, self.request, 100) def test_stale_document(self): approval = self.approval() self.store.docs["DOC-001"]["version"] = 2 with self.assertRaises(Denied): self.store.execute(self.actor, self.request, 101, approval) def test_disable(self): self.store.enabled = False with self.assertRaises(Denied): self.store.execute(self.actor, {"action": "read", "document_id": "DOC-001"}, 100) if __name__ == "__main__": unittest.main(verbosity=2)