"""Offline teaching example: a governed tool call against an in-memory fake CRM. Python 3.9+, standard library only. No network calls or real customer records. Run: python3 tool_contract_lab.py --test Run: python3 tool_contract_lab.py --demo The lock and idempotency store cover one process only. They are not a durable transaction protocol for external effects. Production integrations need trusted identity and approval services, durable records, and downstream reconciliation. """ import argparse from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from hashlib import sha256 import json import re from threading import Lock import time import unittest class Denied(Exception): """A structured policy or contract failure with no new business write.""" @dataclass(frozen=True) class Principal: actor: str tenant: str scopes: frozenset @dataclass(frozen=True) class Approval: actor: str tenant: str action_hash: str expires_at: float def action_hash(principal, arguments): canonical = json.dumps({ 'tool': 'crm.set_case_status.v1', 'actor': principal.actor, 'tenant': principal.tenant, 'arguments': arguments, }, sort_keys=True, separators=(',', ':')) return sha256(canonical.encode('utf-8')).hexdigest() class Gateway: """Principal and approvals are injected by a trusted application boundary. The caller supplies only arguments, an idempotency key and an approval ID. The fake downstream mutation and replay record are atomic under one lock. """ def __init__(self, cases, approvals, clock=time.time): self.cases = dict(cases) self.approvals = dict(approvals) self.clock = clock self.records = {} self.used_approvals = set() self.audit = [] self.write_count = 0 self.lock = Lock() def call(self, principal, arguments, request_key, approval_id): with self.lock: digest = None try: if not isinstance(principal, Principal): raise Denied('untrusted_principal') if not isinstance(arguments, dict) or set(arguments) != {'case_id', 'new_status'}: raise Denied('invalid_argument_fields') if not isinstance(arguments['case_id'], str) or not re.fullmatch(r'case_[0-9]{1,12}', arguments['case_id']): raise Denied('invalid_case_id') if arguments['new_status'] not in ('resolved', 'escalated'): raise Denied('invalid_status') if not isinstance(request_key, str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}', request_key): raise Denied('invalid_idempotency_key') if principal.actor != 'support_agent' or 'cases:write' not in principal.scopes: raise Denied('insufficient_authority') resource = (principal.tenant, arguments['case_id']) if resource not in self.cases: raise Denied('resource_unavailable') digest = action_hash(principal, arguments) record_key = (principal.tenant, principal.actor, request_key) previous = self.records.get(record_key) if previous: if previous['action_hash'] != digest: raise Denied('idempotency_conflict') result = dict(previous['result'], replayed=True) self._audit(principal, request_key, digest, 'replayed', None) return result if not isinstance(approval_id, str): raise Denied('approval_missing') approval = self.approvals.get(approval_id) if not isinstance(approval, Approval): raise Denied('approval_missing') if approval.actor != principal.actor or approval.tenant != principal.tenant or approval.action_hash != digest: raise Denied('approval_binding_mismatch') if self.clock() >= approval.expires_at: raise Denied('approval_expired') if approval_id in self.used_approvals: raise Denied('approval_already_used') old_status = self.cases[resource] self.cases[resource] = arguments['new_status'] self.write_count += 1 result = {'case_id': arguments['case_id'], 'status': arguments['new_status'], 'replayed': False} self.records[record_key] = {'action_hash': digest, 'result': dict(result)} self.used_approvals.add(approval_id) self._audit(principal, request_key, digest, 'executed', None, before_status=old_status, after_status=arguments['new_status']) return result except Denied as error: self._audit(principal, request_key, digest, 'denied', str(error)) raise def _audit(self, principal, request_key, digest, decision, reason, **extra): self.audit.append({ 'timestamp': self.clock(), 'tool': 'crm.set_case_status.v1', 'actor': principal.actor if isinstance(principal, Principal) else None, 'tenant': principal.tenant if isinstance(principal, Principal) else None, 'request_key': request_key if isinstance(request_key, str) else None, 'action_hash': digest, 'decision': decision, 'reason': reason, **extra, }) def fixture(now=1000, expires_at=1100): principal = Principal('support_agent', 'tenant_demo', frozenset({'cases:write'})) arguments = {'case_id': 'case_1', 'new_status': 'resolved'} approval = Approval(principal.actor, principal.tenant, action_hash(principal, arguments), expires_at) gateway = Gateway({('tenant_demo', 'case_1'): 'open'}, {'approval_demo': approval}, clock=lambda: now) return gateway, principal, arguments class ContractTests(unittest.TestCase): def setUp(self): self.gateway, self.principal, self.arguments = fixture() def call(self, **overrides): values = dict(principal=self.principal, arguments=self.arguments, request_key='request_1', approval_id='approval_demo') values.update(overrides) return self.gateway.call(**values) def assert_denied(self, reason, **overrides): with self.assertRaisesRegex(Denied, '^' + reason + '$'): self.call(**overrides) self.assertEqual(self.gateway.write_count, 0) self.assertEqual(self.gateway.audit[-1]['decision'], 'denied') def test_valid_approved_action_executes(self): result = self.call() self.assertEqual(result['status'], 'resolved') self.assertFalse(result['replayed']) self.assertEqual(self.gateway.write_count, 1) def test_caller_cannot_supply_authority_fields(self): self.assert_denied('invalid_argument_fields', arguments=dict(self.arguments, scopes=['cases:write'])) def test_unknown_status_is_rejected(self): self.assert_denied('invalid_status', arguments=dict(self.arguments, new_status='deleted')) def test_missing_permission_is_rejected(self): principal = Principal('support_agent', 'tenant_demo', frozenset()) self.assert_denied('insufficient_authority', principal=principal) def test_another_tenant_cannot_access_the_case(self): principal = Principal('support_agent', 'another_tenant', frozenset({'cases:write'})) self.assert_denied('resource_unavailable', principal=principal) def test_expired_approval_cannot_start_a_write(self): self.gateway, self.principal, self.arguments = fixture(expires_at=1000) self.assert_denied('approval_expired') def test_approval_is_bound_to_the_exact_action(self): self.assert_denied('approval_binding_mismatch', arguments=dict(self.arguments, new_status='escalated')) def test_a_retry_returns_the_saved_result_without_another_write(self): self.call() self.assertTrue(self.call()['replayed']) self.assertEqual(self.gateway.write_count, 1) def test_a_key_cannot_be_reused_for_a_different_action(self): self.call() with self.assertRaisesRegex(Denied, '^idempotency_conflict$'): self.call(arguments=dict(self.arguments, new_status='escalated')) self.assertEqual(self.gateway.write_count, 1) def test_a_consumed_approval_cannot_authorize_a_new_request_key(self): self.call() with self.assertRaisesRegex(Denied, '^approval_already_used$'): self.call(request_key='another_request') self.assertEqual(self.gateway.write_count, 1) def test_concurrent_duplicates_share_one_write_in_this_process(self): with ThreadPoolExecutor(max_workers=8) as executor: results = list(executor.map(lambda _: self.call(), range(20))) self.assertEqual(self.gateway.write_count, 1) self.assertEqual(sum(not result['replayed'] for result in results), 1) def test_replay_still_checks_current_authority(self): self.call() principal = Principal('support_agent', 'tenant_demo', frozenset()) with self.assertRaisesRegex(Denied, '^insufficient_authority$'): self.call(principal=principal) self.assertEqual(self.gateway.write_count, 1) def test_audit_distinguishes_execution_replay_and_denial(self): self.call(); self.call() with self.assertRaises(Denied): self.call(arguments=dict(self.arguments, new_status='deleted')) self.assertEqual([row['decision'] for row in self.gateway.audit], ['executed', 'replayed', 'denied']) self.assertEqual(self.gateway.audit[0]['before_status'], 'open') self.assertNotIn('arguments', self.gateway.audit[0]) def demo(): gateway, principal, arguments = fixture() print(json.dumps(gateway.call(principal, arguments, 'demo_1', 'approval_demo'), indent=2)) print(json.dumps(gateway.call(principal, arguments, 'demo_1', 'approval_demo'), indent=2)) print(json.dumps({'writes': gateway.write_count, 'audit': gateway.audit}, indent=2)) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) action = parser.add_mutually_exclusive_group(required=True) action.add_argument('--test', action='store_true') action.add_argument('--demo', action='store_true') args = parser.parse_args() if args.test: unittest.main(argv=['tool_contract_lab.py'], verbosity=2) else: demo()