"""Synthetic custody/transport tests. No native data, Docker or cluster action.""" import copy import datetime import io import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch from contextlib import redirect_stdout import actuator from authority_journal import Journal, authority from native_program import PROGRAM from receipt_safety import counts CASE = Path('/home/worsch/informed-decision/docs/evidence/2026-10-11-purpose-image-release') NOW = datetime.datetime(2026, 10, 11, 2, 0, tzinfo=datetime.timezone.utc) REVISION = 'e' * 40 COUNT = {'schema_version': 5, 'integrity': 'ok', 'read_only_pre_open': True, 'store_constructed': False, 'lifecycle': {'records': 3, 'events': 8, 'pending': 1}, 'outbox': {'pending': 2, 'delivered': 10}, 'submissions': {'confirmed': 1, 'unresolved': 1}} def accepted(): approval = '11111111-1111-4111-8111-111111111111' review = 'review-22222222-2222-4222-8222-222222222222' disposition = 'disp-33333333-3333-4333-8333-333333333333' receipt = {'proposal_id': 'INFD-WP-0009-T04/purpose-overlay-v1', 'memo_id': 'infd-20261011-purpose-overlay-r1', 'memo_version': 1, 'memo_sha256': 'a' * 64, 'source_revision': REVISION, 'activation': 'APPROVED', 'authority_mode': 'ADMINISTER', 'target': 'realm:kubernetes/railiance01/informed-decision', 'verification_method': 'root-fixed-origin-native-readback', 'runtime_authenticated_grant': False, 'verified_at': '2026-10-11T02:00:00Z', 'accepted_at': '2026-10-11T01:59:00Z', 'review_memo_id': review, 'approval_id': approval, 'disposition_id': disposition} outcome = {'memo_id': receipt['memo_id'], 'review': {'memo_id': review, 'stale': False, 'intake_state': 'ready'}, 'dispositions': [{'approval_id': None, 'submission_state': 'confirmed', 'disposition': {'verb': 'accept', 'memo_version': 1, 'memo_id': review, 'approval_id': approval, 'id': disposition, 'actor': {'kind': 'person', 'sub': 'uid=platform-root,ou=people,dc=netkingdom,dc=local'}, 'at': receipt['accepted_at']}}]} return receipt, outcome class AuthorityAndCustody(unittest.TestCase): def test_exact_nested_founder_receipt_not_top_level_approval(self): receipt, outcome = accepted() result = authority(receipt, outcome, 'a' * 64, REVISION, 'drill-backup', NOW) self.assertEqual(result['approval_id'], receipt['approval_id']) def test_pending_stale_wrong_actor_version_and_binding_refuse(self): mutations = [lambda r, o: r.update(activation='PENDING'), lambda r, o: r.update(memo_sha256='changed'), lambda r, o: r.update(source_revision='changed'), lambda r, o: r.update(target='other'), lambda r, o: r.update(verified_at='2026-10-11T01:00:00Z'), lambda r, o: r.update(verification_method='caller-asserted'), lambda r, o: r.update(runtime_authenticated_grant=True), lambda r, o: o['review'].update(stale=True), lambda r, o: o['review'].update(intake_state='pending'), lambda r, o: o['dispositions'][0].update(submission_state='unresolved'), lambda r, o: o['dispositions'][0]['disposition'].update(memo_version=2), lambda r, o: o['dispositions'][0]['disposition'].update(actor={'kind': 'agent', 'sub': 'platform-root'}), lambda r, o: o['dispositions'].append(copy.deepcopy(o['dispositions'][0]))] for mutate in mutations: with self.subTest(mutate=mutate): r, o = accepted() mutate(r, o) with self.assertRaises(ValueError): authority(r, o, 'a' * 64, REVISION, 'release', NOW) def test_expired_window_and_unknown_phase_refuse(self): r, o = accepted() with self.assertRaises(ValueError): authority(r, o, 'a' * 64, REVISION, 'arbitrary', NOW) late = datetime.datetime(2026, 10, 13, tzinfo=datetime.timezone.utc) r['verified_at'] = late.isoformat() with self.assertRaises(ValueError): authority(r, o, 'a' * 64, REVISION, 'rollback', late) def test_durable_intent_precedes_dispatch_and_never_retries(self): with tempfile.TemporaryDirectory() as directory: parent = Path(directory) os.chmod(parent, 0o700) journal = Journal(parent, {'case': 'fixture'}) journal.start('drill-backup', {'approval_id': 'fixture'}) self.assertEqual(json.loads(journal.path.read_text())['phases']['drill-backup']['state'], 'RUNNING') journal.close() # simulated crash before result recovered = Journal(parent, {'case': 'fixture'}) with self.assertRaises(ValueError): recovered.start('drill-backup', {}) with self.assertRaises(ValueError): recovered.start('drill-copy', {}) recovered.close() def test_phase_order_and_uncertain_release_allows_only_distinct_recovery(self): with tempfile.TemporaryDirectory() as directory: os.chmod(directory, 0o700) j = Journal(directory, {'case': 'fixture'}) with self.assertRaises(ValueError): j.start('release', {}) for phase in actuator.PHASES[:6]: j.start(phase, {}) j.finish(phase, {'all_checks_pass': True}) j.start('release', {}) j.finish('release', {'all_checks_pass': False}) with self.assertRaises(ValueError): j.start('release', {}) with self.assertRaises(ValueError): j.start('verify', {}) j.start('rollback', {}) j.finish('rollback', {'all_checks_pass': True}) j.close() def test_secret_or_unknown_child_output_is_withheld(self): for body in ('not-json', '{"all_checks_pass":true,"token":"private-value"}', '{"all_checks_pass":true,"snapshot_readonly":{"schema_version":5}}'): with self.assertRaises(ValueError): actuator.json_result(subprocess.CompletedProcess([], 0, body, '')) wrong = copy.deepcopy(COUNT) wrong['outbox'] = {'private-row-value': 1} with self.assertRaises(ValueError): counts(wrong) counts(COUNT) # blocked/unresolved counts are not required to be zero. def test_binary_copy_is_bounded_private_and_never_echoed(self): blob = b'synthetic private fixture bytes' digest = __import__('hashlib').sha256(blob).hexdigest() metadata = {'all_checks_pass': True, 'snapshot_file': 'drill.sqlite', 'snapshot_bytes': len(blob), 'snapshot_sha256': digest, 'snapshot_readonly': COUNT, 'pod_name': 'fixture-pod', 'pod_uid': 'fixture-uid'} journal = type('FixtureJournal', (), {'data': {'phases': {'drill-backup': {'result': metadata}}}})() real_popen = subprocess.Popen for extra in (b'', b'oversized'): with self.subTest(extra=extra), tempfile.TemporaryDirectory() as folder: os.chmod(folder, 0o700) output = io.StringIO() def child(*args, **kwargs): return real_popen([sys.executable, '-c', 'import sys;sys.stdout.buffer.write(' + repr(blob + extra) + ')'], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) with patch.object(actuator, 'CUSTODY', Path(folder)), patch.object(actuator, 'remote', return_value=metadata), \ patch('subprocess.Popen', side_effect=child), redirect_stdout(output): if extra: with self.assertRaises(ValueError): actuator.copy_snapshot('drill-copy', REVISION, {}, journal) else: result = actuator.copy_snapshot('drill-copy', REVISION, {}, journal) self.assertTrue(result['off_cluster_private_copy_verified']) self.assertEqual((Path(folder) / 'drill.sqlite').read_bytes(), blob) self.assertEqual(output.getvalue(), '') self.assertEqual((Path(folder) / 'drill.sqlite').stat().st_mode & 0o777, 0o600) def test_recreated_offline_volume_cannot_be_deleted(self): identity = {'Name': actuator.VOLUME, 'Labels': {'nonce': 'old'}, 'CreatedAt': 'old'} journal = type('FixtureJournal', (), {'data': {'phases': { 'offline-verify': {'result': {'created_volume_identity': identity}}}}})() inspected = [{**identity, 'CreatedAt': 'recreated'}] with patch.object(actuator, 'docker', return_value=subprocess.CompletedProcess([], 0, json.dumps(inspected), '')) as mocked: with self.assertRaises(ValueError): actuator.cleanup_offline(journal) self.assertEqual(mocked.call_count, 1) def test_offline_post_admin_integrity_and_original_hash_are_rechecked(self): for defect in ('none', 'post-integrity', 'original-hash'): with self.subTest(defect=defect), tempfile.TemporaryDirectory() as directory: folder=Path(directory);os.chmod(folder,0o700) snapshot=folder/'drill.sqlite';snapshot.write_bytes(b'private synthetic snapshot');os.chmod(snapshot,0o600) digest=actuator.sha(snapshot) prior={'snapshot_sha256':digest,'snapshot_readonly':COUNT} journal=type('FixtureJournal',(),{'data':{'phases':{'drill-copy':{'result':prior},'offline-verify':{'attempt_id':'nonce'}}},'note':lambda *args:None})() reads=[] volume={'Name':actuator.VOLUME,'Labels':{'informed-decision.case':'infd-purpose-20261011-r1','informed-decision.use':'offline-restore-only','informed-decision.attempt':'nonce'},'Driver':'local','Scope':'local','CreatedAt':'fixture','Options':None} def docker(argv, body=None, **kwargs): if argv[0]=='info':body=json.dumps('bnt-lap001') elif argv[:2]==['volume','ls']:body='' elif argv[:2]==['volume','create']:body=actuator.VOLUME elif argv[:2]==['volume','inspect']:body=json.dumps([volume]) elif 'informed-decision-admin' in argv: body=json.dumps({key:COUNT[key] for key in ('schema_version','lifecycle','outbox','submissions')}) if defect=='original-hash':snapshot.write_bytes(b'changed synthetic original') elif '--entrypoint' in argv and 'python' in argv: reads.append(argv);proof=copy.deepcopy(COUNT) if len(reads)==2 and defect=='post-integrity':proof['integrity']='corrupt' body=json.dumps(proof) else:raise AssertionError('unexpected offline command') return subprocess.CompletedProcess(argv,0,body,'') with patch.object(actuator,'CUSTODY',folder),patch.object(actuator,'CASE',CASE),patch.object(actuator,'docker',side_effect=docker),patch('subprocess.run',return_value=subprocess.CompletedProcess([],0,'{"all_checks_pass":true,"private_copy_verified":true}','')): if defect=='none': result=actuator.offline_verify(journal) self.assertEqual(result['offline_post_admin_readonly'],COUNT) self.assertTrue(result['original_snapshot_hash_unchanged']) else: with self.assertRaises(ValueError):actuator.offline_verify(journal) self.assertEqual(len(reads),2) class FixedNativeProgram(unittest.TestCase): def fixture(self): sealed = json.loads((CASE / 'sealed-live-deployment.json').read_text()) live = copy.deepcopy(sealed) live['metadata'].update(generation=42) live['status'] = {'readyReplicas': 1, 'updatedReplicas': 1, 'observedGeneration': 42} dependencies = json.loads((CASE / 'sealed-native-dependencies.json').read_text()) cm = copy.deepcopy(dependencies['configmap']) cm.pop('canonical_data_sha256') service = json.loads((CASE / 'candidate-deployment.json').read_text()) source_service = json.loads((CASE.parents[2] / 'deploy/lifecycle/service.candidate.json').read_text()) cm['data'] = next(x for x in source_service['items'] if x['kind'] == 'ConfigMap')['data'] pvc = copy.deepcopy(dependencies['pvc']) pvc['status'] = {'phase': 'Bound'} app = {'metadata': {'uid': '499868f2-eabc-4df4-ae31-b89ce58d750d', 'resourceVersion': '17'}, 'spec': {'source': {'repoURL': 'https://forgejo.coulomb.social/coulomb/informed-decision.git', 'path': 'deploy/lifecycle', 'targetRevision': '7db8485ffecdebfabd77a9daec085553c74bf0bf'}, 'destination': {'namespace': 'informed-decision', 'server': 'https://kubernetes.default.svc'}}} objects = {'deployment': live, 'configmap': cm, 'pvc': pvc, 'application': app} payload = {'phase': 'release', 'source_revision': REVISION, 'approval': {}, 'sealed': sealed, 'dependencies': dependencies, 'candidate': service, 'rollback': json.loads((CASE / 'rollback-deployment.json').read_text()), 'readonly': (CASE / 'sqlite_readonly.py').read_text(),'purpose_checks':(CASE/'check_installed.py').read_text(), 'receipt_safety': (Path(__file__).parent / 'receipt_safety.py').read_text()} return objects, payload def execute_fake(self, objects, payload, ambiguous=False, schema_refused=False): writes = [] calls = [] def fake(argv, input=None, **kwargs): calls.append(argv) if argv[:2] == ['kubectl', 'get']: kind = argv[2] if kind in objects: body = objects[kind] elif kind == 'pods': body = {'items': [{'metadata': {'name': 'fixture-pod', 'uid': 'pod-uid', 'ownerReferences': [{'kind': 'ReplicaSet', 'name': 'fixture-rs', 'uid': 'rs-uid'}]}, 'spec': objects['deployment']['spec']['template']['spec'], 'status': {'phase': 'Running', 'containerStatuses': [{'name':'informed-decision','ready': True}]}}]} elif kind == 'replicaset': body = {'metadata': {'uid': 'rs-uid', 'ownerReferences': [ {'kind': 'Deployment', 'uid': payload['sealed']['metadata']['uid']}]}} elif kind == 'nodes': body = {'items': [{'metadata': {'name': 'fixture-node'}, 'status': { 'addresses': [{'address': '92.205.62.239'}], 'allocatable': {'cpu': '4'}, 'conditions': [{'type': key, 'status': 'True' if key == 'Ready' else 'False'} for key in ('Ready', 'MemoryPressure', 'DiskPressure', 'PIDPressure')]}}]} else: raise AssertionError('unexpected fixed metadata route') return subprocess.CompletedProcess(argv, 0, json.dumps(body), '') if argv[:2] == ['kubectl', 'exec']: if schema_refused: return subprocess.CompletedProcess(argv, 1, '', 'withheld') return subprocess.CompletedProcess(argv, 0, json.dumps({ 'all_checks_pass': True, 'live_readonly_schema_integrity_counts': COUNT}), '') if argv[:2] == ['kubectl', 'apply']: admitted = copy.deepcopy(payload['sealed']) admitted['spec']['template']['spec']['containers'][0]['image'] = (actuator.IMAGE if payload['phase']=='release' else payload['sealed']['spec']['template']['spec']['containers'][0]['image']) return subprocess.CompletedProcess(argv, 0, json.dumps(admitted), '') if argv[:2] == ['kubectl', 'describe']: return subprocess.CompletedProcess(argv, 0, 'Allocated resources:\n cpu 3460m (86%)\n', '') if argv[:2] == ['kubectl', 'diff']: return subprocess.CompletedProcess(argv, 1, '-image: base\n+image: candidate\n', '') if argv[:2] == ['kubectl', 'patch']: if '--dry-run=server' not in argv: writes.append(json.loads(input)) if ambiguous: return subprocess.CompletedProcess(argv, 1, '', 'withheld') return subprocess.CompletedProcess(argv, 0, 'application.argoproj.io/informed-decision\n', '') raise AssertionError('unexpected command') source = PROGRAM.replace('__PAYLOAD__', repr(json.dumps(payload))).replace( '__GUARD__', repr((CASE / 'change_guard.py').read_text())) output = io.StringIO() with patch('subprocess.run', side_effect=fake), redirect_stdout(output): try: exec(compile(source, 'fixed-native-fixture', 'exec'), {'__name__': '__fixture__'}) except (ValueError, SystemExit): pass receipt = json.loads(output.getvalue()) if output.getvalue() else {'all_checks_pass': False} return receipt, writes, calls def test_only_existing_application_operation_and_deployment_are_requested(self): objects, payload = self.fixture() receipt, writes, calls = self.execute_fake(objects, payload) self.assertTrue(receipt['all_checks_pass']) self.assertFalse(receipt['rollout_verified']) self.assertEqual(len(writes), 1) operation = writes[0][-1]['value'] self.assertFalse(operation['sync']['prune']) self.assertEqual(operation['sync']['resources'], [ {'group': 'apps', 'kind': 'Deployment', 'name': 'informed-decision', 'namespace': 'informed-decision'}]) self.assertTrue(any('--dry-run=server' in argv and argv[:2] == ['kubectl', 'patch'] for argv in calls)) self.assertEqual(writes[0][0]['path'], '/metadata/uid') def test_source_overrides_automation_and_custody_drift_stop_before_write(self): mutations = [lambda o: o['application']['metadata'].update(uid='recreated'), lambda o: o['application']['spec'].update(syncPolicy={'automated': {}}), lambda o: o['application']['spec']['source'].update(kustomize={'images': ['other']}), lambda o: o['application']['spec'].update(sources=[{}]), lambda o: o['configmap']['metadata'].update(uid='recreated'), lambda o: o['pvc']['spec'].update(volumeName='replaced'), lambda o: o['deployment']['status'].update(readyReplicas=0)] for mutate in mutations: with self.subTest(mutate=mutate): objects, payload = self.fixture() mutate(objects) receipt, writes, calls = self.execute_fake(objects, payload) self.assertFalse(receipt['all_checks_pass']) self.assertEqual(writes, []) def test_preopen_schema_refusal_prevents_image_operation(self): objects, payload = self.fixture() receipt, writes, calls = self.execute_fake(objects, payload, schema_refused=True) self.assertFalse(receipt['all_checks_pass']) self.assertEqual(writes, []) def test_rollback_unready_candidate_never_reads_pod_or_database(self): objects, payload = self.fixture() payload['phase'] = 'rollback' objects['deployment']['status'].update(readyReplicas=0, updatedReplicas=0) objects['deployment']['spec']['template']['spec']['containers'][0]['image'] = actuator.IMAGE objects['application']['spec']['source']['targetRevision'] = REVISION receipt, writes, calls = self.execute_fake(objects, payload, schema_refused=True) self.assertTrue(receipt['all_checks_pass']) self.assertFalse(receipt['rollout_verified']) self.assertEqual(len(writes), 1) self.assertFalse(any(argv[:2] == ['kubectl', 'exec'] or argv[:3] == ['kubectl', 'get', 'pods'] for argv in calls)) self.assertEqual(writes[0][-1]['value']['sync']['revision'], '7db8485ffecdebfabd77a9daec085553c74bf0bf') def test_rollback_still_refuses_custody_or_nonpair_image_drift(self): for defect in ('cm', 'pvc', 'image'): with self.subTest(defect=defect): objects, payload = self.fixture();payload['phase'] = 'rollback' objects['deployment']['status']['readyReplicas'] = 0 if defect == 'cm':objects['configmap']['metadata']['uid'] = 'recreated' elif defect == 'pvc':objects['pvc']['spec']['volumeName'] = 'replaced' else:objects['deployment']['spec']['template']['spec']['containers'][0]['image'] = 'foreign-image' receipt, writes, calls = self.execute_fake(objects, payload) self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes, []) def test_recovery_verification_is_separate_ready_old_image_schema_proof(self): objects, payload = self.fixture();payload['phase'] = 'rollback-verify' objects['application']['status'] = {'operationState': {'phase': 'Succeeded', 'operation': {'sync': { 'revision': '7db8485ffecdebfabd77a9daec085553c74bf0bf', 'prune': False, 'resources': [{'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}]}}}} receipt, writes, calls = self.execute_fake(objects, payload) self.assertTrue(receipt['all_checks_pass']);self.assertTrue(receipt['recovery_verification']) self.assertEqual(writes, []) self.assertTrue(any(argv[:2]==['kubectl','exec'] for argv in calls)) objects['deployment']['status']['readyReplicas'] = 0 receipt, writes, calls = self.execute_fake(objects,payload) self.assertFalse(receipt['all_checks_pass']) def test_ambiguous_patch_is_never_retried(self): objects, payload = self.fixture() receipt, writes, calls = self.execute_fake(objects, payload, ambiguous=True) self.assertFalse(receipt['all_checks_pass']) self.assertEqual(len(writes), 1) if __name__ == '__main__': unittest.main()