tegwick
312ab97e87
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
723 lines
57 KiB
Python
723 lines
57 KiB
Python
"""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, fresh_scan, recovery_authority
|
|
from native_program import PROGRAM
|
|
from receipt_safety import counts
|
|
|
|
CASE = Path(__file__).resolve().parent.parent
|
|
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-T05/zlib85091-r1',
|
|
'memo_id': 'infd-20261011-zlib-remediation-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, 'outcome_query': '/lifecycle/memos/infd-20261011-zlib-remediation-r1/outcomes?version=1',
|
|
'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-copy', 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=3),
|
|
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-backup', {})
|
|
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-zlib-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)
|
|
|
|
def test_fresh_full_scan_db_expiry_and_new_findings_refuse(self):
|
|
receipt={'image':actuator.IMAGE,'artifact_source_revision':'9b4bb25d325698ae543e7e8a8a3c3592db1f58fc','scanner':'aquasec/trivy@sha256:62b1e65e8869bc4b4c6aa4fa2b21595256c7c2f6018a9d9ad61caf87187c1969','no_suppression':True,'full_allseverity_scan':True,'exit_code':0,'completed_at':'2026-10-11T01:59:00Z','db_metadata':{'UpdatedAt':'2026-10-10T19:00:00Z','NextUpdate':'2026-10-11T19:00:00Z','Version':2},'db_sha256':'a'*64,'full_report_sha256':'b'*64}
|
|
contract=json.loads((CASE/'fresh-scan-contract.json').read_text())
|
|
report=json.loads((CASE/'fresh-scan-report-fixture.json').read_text())
|
|
self.assertEqual(fresh_scan(receipt,report,contract,NOW)['allseverity_findings'],0)
|
|
mutations=[lambda r,p:r.update(no_suppression=False),lambda r,p:r.update(image='foreign'),lambda r,p:r.update(scanner='other'),lambda r,p:r.update(completed_at='2026-10-10T21:00:00Z'),lambda r,p:r['db_metadata'].update(NextUpdate='2026-10-11T01:59:00Z'),lambda r,p:p['Results'][0].update(Vulnerabilities=[{'Severity':'LOW'}]),lambda r,p:p.update(Results=[]),lambda r,p:p.update(Results=[{}]),lambda r,p:p['Results'].pop(),lambda r,p:p['Results'][0]['Packages'].pop(),lambda r,p:p['Results'][0]['Packages'][0].update(Version='old'),lambda r,p:p['Metadata'].update(ImageID='other'),lambda r,p:p['Metadata']['DiffIDs'].pop(),lambda r,p:p['Metadata']['ImageConfig']['config']['Labels'].update(**{'org.opencontainers.image.revision':'other'}),lambda r,p:p.update(SchemaVersion=1),lambda r,p:p.update(ArtifactType='filesystem')]
|
|
for mutate in mutations:
|
|
r,p=copy.deepcopy(receipt),copy.deepcopy(report);mutate(r,p)
|
|
with self.assertRaises(ValueError):fresh_scan(r,p,contract,NOW)
|
|
|
|
|
|
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((Path('/home/worsch/informed-decision/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': 'c96b0c6d881e855d1d158003c6de90ef390262e2'},
|
|
'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(),
|
|
'installed_zlib': json.loads((CASE / 'installed-zlib-proof.json').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, ephemeral=False, expire_recovery_before_effect=False):
|
|
writes = []
|
|
calls = []
|
|
elapsed=[0];real_datetime=datetime.datetime
|
|
class Clock(real_datetime):
|
|
@classmethod
|
|
def now(cls,tz=None):return real_datetime.now(tz)+datetime.timedelta(seconds=elapsed[0])
|
|
|
|
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': 'c4cb1550-feda-4097-9de4-d891a657b662',
|
|
'ownerReferences': [{'kind': 'ReplicaSet', 'name': 'fixture-rs', 'uid': 'rs-uid'}]},
|
|
'spec': {**objects['deployment']['spec']['template']['spec'],**({'ephemeralContainers':[{'name':'debug','image':'unexpected'}]} if ephemeral else {})},
|
|
'status': {'phase': 'Running', 'containerStatuses': [{'name':'informed-decision','ready': True,'imageID':objects['deployment']['spec']['template']['spec']['containers'][0]['image']}] }}]}
|
|
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 expire_recovery_before_effect and '--dry-run=server' in argv:elapsed[0]=200
|
|
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), patch('datetime.datetime',Clock), 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'], 'c96b0c6d881e855d1d158003c6de90ef390262e2')
|
|
|
|
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'
|
|
resource={'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}
|
|
target='c96b0c6d881e855d1d158003c6de90ef390262e2'
|
|
payload['operation_intent']={'state':'COMPLETE','started_at':'2026-10-11T01:00:00Z','result':{'operation_requested':True,'source_revision':target,'prune':False,'resources':[resource]}}
|
|
objects['application']['status']={'operationState':{'phase':'Succeeded','startedAt':'2026-10-11T01:00:01Z','finishedAt':'2026-10-11T01:01:00Z','operation':{'initiatedBy':{'username':'custodian-codex'},'sync':{'revision':target,'resources':[resource]}},'syncResult':{'revision':target,'resources':[{**resource,'status':'Synced'}]}}}
|
|
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)
|
|
|
|
def test_false_omitempty_needs_bound_current_request_and_exact_completion(self):
|
|
objects,payload=self.fixture();payload['phase']='rollback-verify'
|
|
resource={'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}
|
|
target='c96b0c6d881e855d1d158003c6de90ef390262e2'
|
|
payload['operation_intent']={'state':'COMPLETE','started_at':'2026-10-11T01:00:00Z','result':{'operation_requested':True,'source_revision':target,'prune':False,'resources':[resource]}}
|
|
objects['application']['status']={'operationState':{'phase':'Succeeded','startedAt':'2026-10-11T01:00:01Z','finishedAt':'2026-10-11T01:01:00Z','operation':{'initiatedBy':{'username':'custodian-codex'},'sync':{'revision':target,'resources':[resource]}},'syncResult':{'revision':target,'resources':[{**resource,'status':'Synced'}]}}}
|
|
self.assertTrue(self.execute_fake(objects,payload)[0]['all_checks_pass'])
|
|
mutations=[lambda o,p:o['application']['status']['operationState']['operation']['sync'].update(prune=True),
|
|
lambda o,p:o['application']['status']['operationState']['operation']['sync'].update(prune=None),
|
|
lambda o,p:o['application']['status']['operationState'].update(startedAt='2026-10-11T00:59:00Z'),
|
|
lambda o,p:o['application']['status']['operationState']['operation'].update(initiatedBy={'username':'other'}),
|
|
lambda o,p:o['application']['status']['operationState']['syncResult'].update(revision='foreign'),
|
|
lambda o,p:o['application']['status']['operationState']['syncResult']['resources'][0].update(status='Pruned'),
|
|
lambda o,p:p['operation_intent']['result'].update(prune=True),
|
|
lambda o,p:p.update(operation_intent=None)]
|
|
for mutate in mutations:
|
|
o,p=copy.deepcopy(objects),copy.deepcopy(payload);mutate(o,p)
|
|
receipt,writes,calls=self.execute_fake(o,p)
|
|
self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes,[])
|
|
|
|
def test_optimized_executor_refuses_before_any_input(self):
|
|
result=subprocess.run([sys.executable,'-O',str(Path(__file__).parent/'actuator.py'),'--phase','release','--root-authority-receipt','/missing','--root-native-outcome','/missing','--source-revision',REVISION],capture_output=True,text=True)
|
|
self.assertNotEqual(result.returncode,0)
|
|
self.assertIn('optimized interpreter refused',result.stderr)
|
|
|
|
def test_actual_pod_ephemeral_container_refuses_before_native_effect(self):
|
|
objects,payload=self.fixture();receipt,writes,calls=self.execute_fake(objects,payload,ephemeral=True)
|
|
self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes,[])
|
|
self.assertFalse(any(argv[:2]==['kubectl','exec'] for argv in calls))
|
|
|
|
def test_outage_native_rollback_requires_failed_candidate_and_same_source(self):
|
|
objects,payload=self.fixture();payload.update(phase='rollback',outage_recovery=True,recovery_release_started_at=datetime.datetime.now(datetime.timezone.utc).isoformat())
|
|
objects['deployment']['spec']['template']['spec']['containers'][0]['image']=actuator.IMAGE
|
|
objects['deployment']['status'].update(readyReplicas=0,updatedReplicas=0)
|
|
objects['application']['spec']['source']['targetRevision']=REVISION
|
|
self.assertTrue(self.execute_fake(objects,payload)[0]['all_checks_pass'])
|
|
for defect in ('healthy','baseline','wrongsource'):
|
|
o,p=copy.deepcopy(objects),copy.deepcopy(payload)
|
|
if defect=='healthy':o['deployment']['status']['readyReplicas']=1
|
|
elif defect=='baseline':o['deployment']['spec']['template']['spec']['containers'][0]['image']=p['sealed']['spec']['template']['spec']['containers'][0]['image']
|
|
else:o['application']['spec']['source']['targetRevision']='other'
|
|
receipt,writes,calls=self.execute_fake(o,p);self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes,[])
|
|
|
|
def test_native_recovery_window_rechecked_immediately_before_effect(self):
|
|
objects,payload=self.fixture();payload.update(phase='rollback',outage_recovery=True,recovery_release_started_at=(datetime.datetime.now(datetime.timezone.utc)-datetime.timedelta(seconds=1700)).isoformat())
|
|
objects['deployment']['spec']['template']['spec']['containers'][0]['image']=actuator.IMAGE
|
|
objects['deployment']['status'].update(readyReplicas=0,updatedReplicas=0)
|
|
objects['application']['spec']['source']['targetRevision']=REVISION
|
|
receipt,writes,calls=self.execute_fake(objects,payload,expire_recovery_before_effect=True)
|
|
self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes,[])
|
|
self.assertTrue(any('--dry-run=server' in argv and argv[:2]==['kubectl','patch'] for argv in calls))
|
|
|
|
|
|
class FullCliDispatch(unittest.TestCase):
|
|
def invoke(self,phase,corrupt_source=False,outage=False,denial=False):
|
|
import hashlib,shutil
|
|
import authority_journal
|
|
realpath=Path;called=[];output=io.StringIO()
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
base=Path(folder);custody=base/'private';custody.mkdir(mode=0o700)
|
|
empty=custody/'empty-docker-config';empty.mkdir(mode=0o700)
|
|
owner=base/'owner';owner.mkdir();case=owner/'case';case.mkdir();execution=case/'execution';execution.mkdir()
|
|
for name in ('actuator.py','authority_journal.py','native_program.py','receipt_safety.py'):
|
|
shutil.copyfile(Path(__file__).parent/name,execution/name)
|
|
(case/'memo.json').write_text('{}')
|
|
for name in ('sealed-live-deployment.json','sealed-native-dependencies.json','candidate-deployment.json','rollback-deployment.json','check_installed.py','sqlite_readonly.py','change_guard.py','installed-zlib-proof.json','fresh-scan-contract.json'):
|
|
shutil.copyfile(CASE/name,case/name)
|
|
names=('deploy/lifecycle/admission.candidate.json','deploy/lifecycle/service.candidate.json','deploy/lifecycle/kustomization.yaml','deploy/lifecycle/registries.candidate.json')
|
|
for name in names:
|
|
target=owner/name;target.parent.mkdir(parents=True,exist_ok=True);target.write_text('sourcefixture')
|
|
plan={'state':'prepared-unapproved','published_artifact_verified':True,'proposal_id':'INFD-WP-0009-T05/zlib85091-r1','executor_files':{n:actuator.sha(execution/n) for n in ('actuator.py','authority_journal.py','native_program.py','receipt_safety.py')},'case_files':{p.name:actuator.sha(p) for p in case.iterdir() if p.is_file() and p.name!='memo.json'},'owner_source_files':{n:actuator.sha(owner/n) for n in names}}
|
|
(execution/'execution-plan.json').write_text(json.dumps(plan))
|
|
receipt,outcome=accepted();receipt.update(native_outcome_sha256=hashlib.sha256(json.dumps(outcome).encode()).hexdigest(),executor_plan_sha256=actuator.sha(execution/'execution-plan.json'))
|
|
mapping={}
|
|
for fixed,data in [('/tmp/security-infd-zlib-root-authority.json',receipt),('/tmp/security-infd-zlib-native-outcome-latest.json',outcome),('/tmp/security-infd-zlib-fresh-scan-receipt.json',{'fixture':True}),('/tmp/security-infd-zlib-fresh-scan-full.json',{'fixture':True})]:
|
|
target=base/Path(fixed).name;target.write_text(json.dumps(data));target.chmod(0o600);mapping[fixed]=target
|
|
scanreceipt=mapping['/tmp/security-infd-zlib-fresh-scan-receipt.json'];scanreceipt.write_text(json.dumps({'full_report_sha256':actuator.sha(mapping['/tmp/security-infd-zlib-fresh-scan-full.json'])}))
|
|
identity={'proposal':plan['proposal_id'],'source_revision':REVISION,'memo_sha256':actuator.sha(case/'memo.json'),'plan_sha256':actuator.sha(execution/'execution-plan.json')}
|
|
blob=b'private synthetic full-cli snapshot';metadata={'all_checks_pass':True,'snapshot_file':'drill.sqlite','snapshot_bytes':len(blob),'snapshot_sha256':hashlib.sha256(blob).hexdigest(),'snapshot_readonly':COUNT,'pod_name':'fixture-pod','pod_uid':'fixture-poduid'}
|
|
j=Journal(custody,identity)
|
|
predecessors=list(actuator.PHASES[:actuator.PHASES.index(phase)])
|
|
for predecessor in predecessors:
|
|
j.start(predecessor,{})
|
|
result=metadata if predecessor=='drill-backup' else {'all_checks_pass':True}
|
|
if predecessor in ('release','rollback'):result={'all_checks_pass':True,'operation_requested':True,'prune':False,'source_revision':REVISION,'resources':[{'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}]}
|
|
j.finish(predecessor,result)
|
|
j.close()
|
|
if outage:
|
|
receipt.update(memo_sha256=identity['memo_sha256'],bounded_outage_rollback_pre_authorized=True,verified_at='2026-10-11T01:59:00Z',accepted_at='2026-10-11T01:58:00Z')
|
|
outcome['dispositions'][0]['disposition']['at']=receipt['accepted_at']
|
|
cache={'root_authority':receipt,'native_outcome':copy.deepcopy(outcome),'memo_sha256':identity['memo_sha256'],'source_revision':REVISION,'plan_sha256':identity['plan_sha256']}
|
|
cp=custody/'release-authority-cache.json';cp.write_text(json.dumps(cache));cp.chmod(0o600)
|
|
data=json.loads(j.path.read_text());data['phases']['release'].update(started_at='2026-10-11T01:59:30Z',release_authority_cache_sha256=actuator.sha(cp));j.path.write_text(json.dumps(data))
|
|
sealed=json.loads((case/'sealed-live-deployment.json').read_text());deps=json.loads((case/'sealed-native-dependencies.json').read_text())
|
|
probe={'kind':'informed-decision.root-supervised-outage-recovery','proposal_id':plan['proposal_id'],'source_revision':REVISION,'memo_sha256':identity['memo_sha256'],'phase':phase,'release_attempt_id':data['phases']['release']['attempt_id'],'cache_sha256':actuator.sha(cp),'root_verified':True,'manual_supervised_recovery':True,'current_native_outcomes_claimed':False,'observed_at':NOW.isoformat(),'native_probe_status':503,'explicit_revoked':False,'current_conflict':False,'identity_or_scope_mismatch':False,'service_healthy':False,'deployment_uid':sealed['metadata']['uid'],'configmap_uid':deps['configmap']['metadata']['uid'],'pvc_uid':deps['pvc']['metadata']['uid'],'rollback_image':sealed['spec']['template']['spec']['containers'][0]['image'],'rollback_source':'c96b0c6d881e855d1d158003c6de90ef390262e2','failed_candidate':{'image':actuator.IMAGE,'application_source':REVISION,'ready_replicas':0}}
|
|
if phase=='rollback-verify':
|
|
witness=copy.deepcopy(probe);witness['phase']='rollback'
|
|
probe.update(original_outage_receipt_sha256='d'*64,recovered_image=probe['rollback_image'],recovered_application_source=probe['rollback_source'],ready_replicas=1)
|
|
probe.pop('failed_candidate')
|
|
data['phases']['rollback'].update(outage_recovery_receipt_sha256='d'*64,outage_recovery_anchor=witness)
|
|
data['phases']['rollback']['result']['source_revision']=probe['rollback_source']
|
|
j.path.write_text(json.dumps(data))
|
|
rp=base/'root-recovery.json';rp.write_text(json.dumps(probe));rp.chmod(0o600);mapping['/tmp/security-infd-zlib-root-recovery-receipt.json']=rp
|
|
if denial:outcome['http_status']=403
|
|
native=mapping['/tmp/security-infd-zlib-native-outcome-latest.json'];native.write_text(json.dumps(outcome));receipt['native_outcome_sha256']=actuator.sha(native);mapping['/tmp/security-infd-zlib-root-authority.json'].write_text(json.dumps(receipt))
|
|
if corrupt_source:(owner/names[0]).write_text('changed')
|
|
def mapped(value):return mapping.get(str(value),realpath(value))
|
|
def run(argv,body=None,**kwargs):
|
|
if argv[:4]==['git','-C',str(owner),'rev-parse']:return subprocess.CompletedProcess(argv,0,REVISION+'\n','')
|
|
if argv[:4]==['git','-C',str(base),'check-ignore']:return subprocess.CompletedProcess(argv,0,'ignored','')
|
|
if argv[:2]==['ssh','railiance01']:
|
|
import ast
|
|
parsed=ast.parse(body);node=next(x for x in parsed.body if isinstance(x,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='payload' for t in x.targets));payload=json.loads(ast.literal_eval(node.value.args[0]));called.append(payload)
|
|
self.assertEqual(json.loads(j.path.read_text())['phases'][phase]['state'],'RUNNING')
|
|
if phase=='drill-copy':self.assertEqual(payload['snapshot_identity'],metadata)
|
|
if phase in ('verify','rollback-verify'):self.assertEqual(payload['operation_intent']['state'],'COMPLETE');self.assertFalse(payload['operation_intent']['result']['prune'])
|
|
result=metadata if phase=='drill-copy' else {'all_checks_pass':True,'phase':phase}
|
|
return subprocess.CompletedProcess(argv,0,json.dumps(result),'')
|
|
raise AssertionError('unexpected CLI transport')
|
|
def git_show(argv,**kwargs):
|
|
self.assertEqual(argv[:4],['git','-C',str(owner),'show']);data=(owner/argv[4].split(':',1)[1]).read_bytes();return subprocess.CompletedProcess(argv,0,data,b'')
|
|
real_popen=subprocess.Popen
|
|
def binary(*args,**kwargs):return real_popen([sys.executable,'-c','import sys;sys.stdout.buffer.write('+repr(blob)+')'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL)
|
|
argv=['actuator.py','--phase',phase,'--root-authority-receipt','/tmp/security-infd-zlib-root-authority.json','--root-native-outcome','/tmp/security-infd-zlib-native-outcome-latest.json','--root-fresh-scan-receipt','/tmp/security-infd-zlib-fresh-scan-receipt.json','--source-revision',REVISION]
|
|
if outage:argv+=['--root-recovery-receipt','/tmp/security-infd-zlib-root-recovery-receipt.json']
|
|
real_recovery=authority_journal.recovery_authority;real_authority=authority_journal.authority
|
|
def authority_dispatch(*args,**kwargs):
|
|
if outage and len(args)<6:raise ValueError('service unavailable')
|
|
return real_authority(*args,**kwargs) if outage else {'fixture':True}
|
|
with patch.object(actuator,'ROOT',execution),patch.object(actuator,'CASE',case),patch.object(actuator,'OWNER',owner),patch.object(actuator,'CUSTODIAN',base),patch.object(actuator,'CUSTODY',custody),patch.object(actuator,'Path',side_effect=mapped),patch.object(actuator,'run',side_effect=run),patch('subprocess.run',side_effect=git_show),patch('subprocess.Popen',side_effect=binary),patch.object(authority_journal,'authority',side_effect=authority_dispatch),patch.object(authority_journal,'recovery_authority',side_effect=lambda *args: real_recovery(*args,now=NOW)),patch.object(authority_journal,'fresh_scan',return_value={'fixture':True}),patch('sys.argv',argv),redirect_stdout(output):
|
|
if corrupt_source or denial:
|
|
with self.assertRaises(ValueError):actuator.main()
|
|
else:actuator.main()
|
|
state=json.loads(j.path.read_text())['phases'].get(phase)
|
|
return called,state,output.getvalue()
|
|
|
|
def test_main_cli_copy_binds_snapshot_and_retains_private_bytes(self):
|
|
calls,state,output=self.invoke('drill-copy');self.assertEqual(len(calls),2);self.assertEqual(state['state'],'COMPLETE');self.assertNotIn('private synthetic',output)
|
|
|
|
def test_main_cli_verify_binds_completed_release_intent(self):
|
|
calls,state,output=self.invoke('verify');self.assertEqual(len(calls),1);self.assertEqual(state['state'],'COMPLETE')
|
|
|
|
def test_main_cli_release_dispatches_only_after_durable_phase(self):
|
|
calls,state,output=self.invoke('release');self.assertEqual(len(calls),1);self.assertEqual(state['state'],'COMPLETE')
|
|
|
|
def test_main_cli_source_hash_refuses_before_journal_or_transport(self):
|
|
calls,state,output=self.invoke('release',corrupt_source=True);self.assertEqual(calls,[]);self.assertIsNone(state);self.assertEqual(output,'')
|
|
|
|
def test_main_cli_bounded_outage_rollback_uses_retained_cache_and_witness(self):
|
|
calls,state,output=self.invoke('rollback',outage=True)
|
|
self.assertEqual(len(calls),1);self.assertTrue(calls[0]['outage_recovery'])
|
|
self.assertEqual(state['state'],'COMPLETE');self.assertIn('outage_recovery_receipt_sha256',state)
|
|
self.assertFalse(state['approval']['current_revocation_verified'])
|
|
|
|
def test_main_cli_current403_prevents_cached_fallback_before_phase(self):
|
|
calls,state,output=self.invoke('rollback',outage=True,denial=True)
|
|
self.assertEqual(calls,[]);self.assertIsNone(state);self.assertEqual(output,'')
|
|
|
|
def test_main_cli_outage_recovery_verify_binds_prior_completed_rollback(self):
|
|
calls,state,output=self.invoke('rollback-verify',outage=True)
|
|
self.assertEqual(len(calls),1);self.assertTrue(calls[0]['outage_recovery'])
|
|
self.assertEqual(calls[0]['operation_intent']['state'],'COMPLETE')
|
|
self.assertEqual(state['state'],'COMPLETE');self.assertFalse(state['approval']['current_revocation_verified'])
|
|
|
|
|
|
class PacketBindingAndBodyLimit(unittest.TestCase):
|
|
def test_lossless_gzip_bundle_retains_exact_source_hashes(self):
|
|
import gzip,base64,hashlib
|
|
bundle=json.loads((CASE/'execution-bundle.json').read_text())
|
|
for group in ('files','case_evidence'):
|
|
for relative,item in bundle[group].items():
|
|
data=gzip.decompress(base64.b64decode(item['content_base64'],validate=True))
|
|
self.assertEqual(len(data),item['uncompressed_bytes']);self.assertEqual(hashlib.sha256(data).hexdigest(),item['sha256'])
|
|
self.assertEqual(data,(CASE/relative).read_bytes())
|
|
item=next(iter(bundle['files'].values()));changed=bytearray(base64.b64decode(item['content_base64']));changed[-1]^=1
|
|
with self.assertRaises((OSError,EOFError)):gzip.decompress(changed)
|
|
|
|
def test_actual_source_http_256kib_limit_before_register_side_effect(self):
|
|
import os
|
|
script="""import io,json,sys
|
|
sys.dont_write_bytecode=True
|
|
sys.path.insert(0,'/home/worsch/informed-decision')
|
|
from informed_decision.lifecycle_http import LifecycleAPI,MAX_BODY
|
|
from informed_decision.lifecycle_service import Caller
|
|
from informed_decision.lifecycle_service import LifecycleError
|
|
from informed_decision.intake import validate_submission,SubmitterGrant
|
|
from informed_decision.lifecycle_commands import decode_memo,decode_documents
|
|
from pathlib import Path
|
|
root=Path(sys.argv[1]);commands=json.loads((root/'lifecycle-commands.json').read_text());envelope=json.loads((root/'intake.json').read_text())
|
|
validate_submission(envelope,submitter_id='state-hub',grant=SubmitterGrant('state-hub',frozenset({'decision-record'}),frozenset({'net-kingdom-admins'})))
|
|
class Service:
|
|
calls=0
|
|
def _caller(self,caller):pass
|
|
def register(self,*args,**kwargs):self.calls+=1;return 'fixture-event'
|
|
service=Service();api=LifecycleAPI(service,lambda token:object.__new__(Caller))
|
|
typedmemo=decode_memo(commands[0]['register']['memo']);decode_documents(commands[0]['register']['documents'],typedmemo)
|
|
try:decode_documents(commands[0]['register']['documents']+[commands[0]['register']['documents'][0]],typedmemo)
|
|
except ValueError:pass
|
|
else:raise AssertionError('duplicate digest accepted')
|
|
body=json.dumps(commands[0]['register']).encode();assert len(body)<240*1024
|
|
base={'REQUEST_METHOD':'POST','PATH_INFO':'/v1/decisions','QUERY_STRING':'','CONTENT_TYPE':'application/json','HTTP_AUTHORIZATION':'Bearer synthetic-fixture','HTTP_IDEMPOTENCY_KEY':'fixture'}
|
|
status,result=api.dispatch({**base,'CONTENT_LENGTH':str(len(body)),'wsgi.input':io.BytesIO(body)});assert status==200 and service.calls==1
|
|
oversize=b' '+body+b' '*(MAX_BODY+1-len(body)-1)
|
|
try:api.dispatch({**base,'CONTENT_LENGTH':str(len(oversize)),'wsgi.input':io.BytesIO(oversize)})
|
|
except LifecycleError as e:assert e.status==413
|
|
else:raise AssertionError('oversize accepted')
|
|
assert service.calls==1
|
|
print(json.dumps({'synthetic_body_limit_verified':True,'intake_validated':True,'register_bytes':len(body)}))
|
|
"""
|
|
result=subprocess.run([sys.executable,'-c',script,str(CASE)],capture_output=True,text=True,env={**os.environ,'PYTHONDONTWRITEBYTECODE':'1'})
|
|
self.assertEqual(result.returncode,0,result.stderr);self.assertTrue(json.loads(result.stdout)['synthetic_body_limit_verified'])
|
|
|
|
|
|
class OutcomeMetadataAndOutage(unittest.TestCase):
|
|
def load_probe(self):
|
|
import runpy
|
|
return runpy.run_path(str(CASE/'post-ready-native-outcomes-command.py'))
|
|
|
|
|
|
def outcome(self):
|
|
# Actual native shape: ONLYreview/dispositions; top identity absent.
|
|
return {'review':{'memo_id':'review-22222222-2222-4222-8222-222222222222','intake_state':'not_submitted','stale':False,'unknown':'PRIVATE_SENTINEL'},'dispositions':[{'disposition':{'verb':'comment','actor':{'kind':'system','sub':'system:fixture'},'note':'PRIVATE_SENTINEL','reasons':['PRIVATE_SENTINEL']},'submission_state':None,'approved_at':None}]+[{'disposition':{'verb':'accept','actor':{'kind':'person','sub':'uid=fixture'},'note':'PRIVATE_SENTINEL'},'submission_state':state,'approved_at':None} for state in ('prepared','in_flight','confirmed','unresolved')],'unknown':'PRIVATE_SENTINEL'}
|
|
|
|
def test_counts_only_omit_entire_private_native_body_and_never_infer_go(self):
|
|
probe=self.load_probe();body=self.outcome();self.assertNotIn('memo_id',body)
|
|
safe=probe['safe_outcome_metadata'](body);raw=json.dumps(safe)
|
|
self.assertNotIn('PRIVATE_SENTINEL',raw);self.assertNotIn('note',raw);self.assertNotIn('reasons',raw);self.assertNotIn('actor',raw);self.assertNotIn('submission_state',raw)
|
|
self.assertEqual(safe['disposition_count'],5);self.assertFalse(safe['approval_inferred']);self.assertFalse(safe['current_approval_claimed']);self.assertFalse(safe['dispositions_interpreted'])
|
|
self.assertEqual(safe['request_identity_source'],'fixed-authenticated-route')
|
|
self.assertFalse(probe['safe_outcome_metadata']({'review':None,'dispositions':[]})['has_review'])
|
|
|
|
def test_five_native_shape_counterexamples_refuse_without_echo(self):
|
|
probe=self.load_probe()
|
|
mutations=[lambda o:o.update(review=['PRIVATE_SENTINEL']),lambda o:o.update(dispositions='PRIVATE_SENTINEL'),lambda o:o.update(dispositions=[None]),lambda o:o.update(memo_id='PRIVATE_SENTINEL'),lambda o:o.update(version=True),lambda o:o.update(dispositions=[{}]*1025)]
|
|
for mutate in mutations:
|
|
o=self.outcome();mutate(o)
|
|
with self.assertRaises(ValueError) as error:probe['safe_outcome_metadata'](o)
|
|
self.assertNotIn('PRIVATE_SENTINEL',str(error.exception))
|
|
|
|
def test_actual_source_lifecycle_outcomes_native_shape_without_real_database(self):
|
|
import contextlib,types,os
|
|
script="""import contextlib,json,sys,types,runpy
|
|
sys.dont_write_bytecode=True
|
|
sys.path.insert(0,'/home/worsch/informed-decision')
|
|
from informed_decision.lifecycle_service import LifecycleService
|
|
from pathlib import Path
|
|
probe=runpy.run_path(str(Path(sys.argv[1])/'post-ready-native-outcomes-command.py'))
|
|
class Rows(list):
|
|
def fetchone(self):return self[0] if self else None
|
|
class DB:
|
|
def __init__(self,reviewed,ledger):self.reviewed,self.ledger=reviewed,ledger
|
|
def execute(self,sql,params):
|
|
if 'FROM lifecycle_events' in sql:return Rows([{'body':json.dumps({'review_memo_id':'review-22222222-2222-4222-8222-222222222222'})}] if self.reviewed else [])
|
|
if 'FROM intake_ledger' in sql:return Rows([{'state':self.ledger}] if self.ledger is not None else [])
|
|
if 'FROM dispositions' in sql:return Rows([{'body':json.dumps({'id':'fixture-comment','verb':'comment','actor':{'kind':'system','sub':'system:fixture'},'note':'PRIVATE_SENTINEL','reasons':['PRIVATE_SENTINEL']}),'state':None,'approved_at':None}]+[{'body':json.dumps({'id':'fixture-accept','verb':'accept','note':'PRIVATE_SENTINEL'}),'state':state,'approved_at':None} for state in ('prepared','in_flight','confirmed','unresolved')])
|
|
raise AssertionError('unexpected native outcomes query')
|
|
for reviewed,ledger in [(False,None),(True,None),(True,'intake_created'),(True,'ready')]:
|
|
db=DB(reviewed,ledger)
|
|
@contextlib.contextmanager
|
|
def connection():yield db
|
|
store=types.SimpleNamespace(_connection=connection,_memo=lambda *args:types.SimpleNamespace(version=1))
|
|
repository=types.SimpleNamespace(store=store,_record=lambda *args:None)
|
|
service=object.__new__(LifecycleService);service.repository=repository;calls=[]
|
|
service._authorize=lambda caller,memo,action,attrs:(lambda:calls.append('current'),None)
|
|
body=service.outcomes(types.SimpleNamespace(attribution=types.SimpleNamespace(tenant='synthetic')),probe['MEMO_ID'],1)
|
|
assert set(body)=={'review','dispositions'} and 'memo_id' not in body and calls==['current']
|
|
result=probe['safe_outcome_metadata'](body)
|
|
assert result['has_review']==reviewed and result['disposition_count']==(5 if reviewed else 0)
|
|
assert result['approval_inferred'] is False and result['current_approval_claimed'] is False and 'PRIVATE_SENTINEL' not in json.dumps(result)
|
|
print(json.dumps({'actual_source_outcomes_shape_verified':True,'real_database_opened':False,'provider_called':False}))
|
|
"""
|
|
result=subprocess.run([sys.executable,'-c',script,str(CASE)],capture_output=True,text=True,env={**os.environ,'PYTHONDONTWRITEBYTECODE':'1'})
|
|
self.assertEqual(result.returncode,0,result.stderr);self.assertTrue(json.loads(result.stdout)['actual_source_outcomes_shape_verified'])
|
|
|
|
def test_actual_command_mocked_opener_never_prints_private_fields(self):
|
|
import types,runpy,urllib.request
|
|
module=types.ModuleType('api.services.memo_dispatch');module.ORIGIN='http://synthetic.invalid';module.NoRedirect=urllib.request.HTTPRedirectHandler;module.token_file=lambda path:'synthetic-fixture'
|
|
class Response:
|
|
status=200
|
|
def __enter__(self):return self
|
|
def __exit__(self,*args):pass
|
|
def read(self,bound):return json.dumps(OutcomeMetadataAndOutage.outcome(self_test)).encode()
|
|
self_test=self
|
|
class Opener:
|
|
def open(self,request,timeout):return Response()
|
|
output=io.StringIO()
|
|
with patch.dict('sys.modules',{'api.services.memo_dispatch':module}),patch('urllib.request.build_opener',return_value=Opener()),redirect_stdout(output):
|
|
runpy.run_path(str(CASE/'post-ready-native-outcomes-command.py'),run_name='__main__')
|
|
self.assertNotIn('PRIVATE_SENTINEL',output.getvalue());self.assertNotIn('synthetic-fixture',output.getvalue())
|
|
self.assertTrue(json.loads(output.getvalue())['native_provider_path_returned'])
|
|
|
|
|
|
def test_actual_command_private_token_error_and_http403_never_echo_body(self):
|
|
import types,runpy,urllib.request,urllib.error
|
|
module=types.ModuleType('api.services.memo_dispatch');module.ORIGIN='http://synthetic.invalid';module.NoRedirect=urllib.request.HTTPRedirectHandler
|
|
def token_failure(path):raise ValueError('PRIVATE_SENTINEL')
|
|
for failure in ('token','http403'):
|
|
output=io.StringIO();module.token_file=token_failure if failure=='token' else lambda path:'synthetic-fixture'
|
|
class Opener:
|
|
def open(self,*args,**kwargs):raise urllib.error.HTTPError('http://synthetic.invalid',403,'PRIVATE_SENTINEL',{},io.BytesIO(b'PRIVATE_SENTINEL'))
|
|
with patch.dict('sys.modules',{'api.services.memo_dispatch':module}),patch('urllib.request.build_opener',return_value=Opener()),redirect_stdout(output):
|
|
with self.assertRaises(SystemExit):runpy.run_path(str(CASE/'post-ready-native-outcomes-command.py'),run_name='__main__')
|
|
raw=output.getvalue();self.assertNotIn('PRIVATE_SENTINEL',raw);self.assertNotIn('synthetic-fixture',raw)
|
|
receipt=json.loads(raw);self.assertEqual(receipt['status'],'unverified');self.assertNotIn('native_provider_path_returned',receipt)
|
|
if failure=='http403':self.assertEqual(receipt['http_status'],403)
|
|
|
|
def recovery_fixture(self):
|
|
receipt,outcome=accepted();receipt.update(bounded_outage_rollback_pre_authorized=True,verified_at='2026-10-11T01:59:00Z',accepted_at='2026-10-11T01:58:00Z')
|
|
outcome['dispositions'][0]['disposition']['at']=receipt['accepted_at']
|
|
sealed=json.loads((CASE/'sealed-live-deployment.json').read_text());deps=json.loads((CASE/'sealed-native-dependencies.json').read_text())
|
|
cache={'root_authority':receipt,'native_outcome':outcome,'memo_sha256':'a'*64,'source_revision':REVISION,'plan_sha256':'b'*64}
|
|
journal={'identity':{'plan_sha256':'b'*64},'phases':{'release':{'state':'COMPLETE','started_at':'2026-10-11T01:59:30Z','attempt_id':'release-fixture','release_authority_cache_sha256':'c'*64}}}
|
|
probe={'kind':'informed-decision.root-supervised-outage-recovery','proposal_id':'INFD-WP-0009-T05/zlib85091-r1','source_revision':REVISION,'memo_sha256':'a'*64,'phase':'rollback','release_attempt_id':'release-fixture','cache_sha256':'c'*64,'root_verified':True,'manual_supervised_recovery':True,'current_native_outcomes_claimed':False,'observed_at':NOW.isoformat(),'native_probe_status':503,'explicit_revoked':False,'current_conflict':False,'identity_or_scope_mismatch':False,'service_healthy':False,'deployment_uid':sealed['metadata']['uid'],'configmap_uid':deps['configmap']['metadata']['uid'],'pvc_uid':deps['pvc']['metadata']['uid'],'rollback_image':sealed['spec']['template']['spec']['containers'][0]['image'],'rollback_source':'c96b0c6d881e855d1d158003c6de90ef390262e2','failed_candidate':{'image':actuator.IMAGE,'application_source':REVISION,'ready_replicas':0}}
|
|
return probe,cache,journal,sealed,deps
|
|
|
|
def test_explicit_manual_cached_acceptance_only_outage_recovery(self):
|
|
probe,cache,journal,sealed,deps=self.recovery_fixture()
|
|
result=recovery_authority(probe,cache,journal,'a'*64,REVISION,sealed,deps,'rollback',NOW)
|
|
self.assertTrue(result['bounded_outage_recovery']);self.assertFalse(result['current_revocation_verified'])
|
|
mutations=[lambda p,c,j:p.update(native_probe_status=401),lambda p,c,j:p.update(native_probe_status=403),lambda p,c,j:p.update(explicit_revoked=True),lambda p,c,j:p.update(current_conflict=True),lambda p,c,j:p.update(identity_or_scope_mismatch=True),lambda p,c,j:p.update(service_healthy=True),lambda p,c,j:p.update(deployment_uid='recreated'),lambda p,c,j:p.update(source_revision='other'),lambda p,c,j:p.update(configmap_uid='recreated'),lambda p,c,j:p.update(pvc_uid='recreated'),lambda p,c,j:p['failed_candidate'].update(ready_replicas=1),lambda p,c,j:j['phases'].pop('release'),lambda p,c,j:c['root_authority'].update(bounded_outage_rollback_pre_authorized=False),lambda p,c,j:j['phases']['release'].update(started_at='2026-10-11T01:00:00Z')]
|
|
for mutate in mutations:
|
|
p,c,j=copy.deepcopy(probe),copy.deepcopy(cache),copy.deepcopy(journal);mutate(p,c,j)
|
|
with self.assertRaises(ValueError):recovery_authority(p,c,j,'a'*64,REVISION,sealed,deps,'rollback',NOW)
|
|
with self.assertRaises(ValueError):recovery_authority(probe,cache,journal,'a'*64,REVISION,sealed,deps,'release',NOW)
|
|
late=datetime.datetime(2026,10,13,tzinfo=datetime.timezone.utc)
|
|
with self.assertRaises(ValueError):recovery_authority(probe,cache,journal,'a'*64,REVISION,sealed,deps,'rollback',late)
|
|
|
|
def test_recovery_verify_requires_completed_prior_rollback_and_recovered_baseline(self):
|
|
probe,cache,journal,sealed,deps=self.recovery_fixture();probe.update(phase='rollback-verify',original_outage_receipt_sha256='d'*64,recovered_image=probe['rollback_image'],recovered_application_source=probe['rollback_source'],ready_replicas=1)
|
|
journal['phases']['rollback']={'state':'COMPLETE','outage_recovery_receipt_sha256':'d'*64,'outage_recovery_anchor':{'failed_candidate':probe.pop('failed_candidate')}}
|
|
self.assertTrue(recovery_authority(probe,cache,journal,'a'*64,REVISION,sealed,deps,'rollback-verify',NOW)['bounded_outage_recovery'])
|
|
for defect in ('state','link','image','source','ready'):
|
|
p,j=copy.deepcopy(probe),copy.deepcopy(journal)
|
|
if defect=='state':j['phases']['rollback']['state']='UNCERTAIN'
|
|
elif defect=='link':p['original_outage_receipt_sha256']='other'
|
|
elif defect=='image':p['recovered_image']=actuator.IMAGE
|
|
elif defect=='source':p['recovered_application_source']=REVISION
|
|
else:p['ready_replicas']=0
|
|
with self.assertRaises(ValueError):recovery_authority(p,cache,j,'a'*64,REVISION,sealed,deps,'rollback-verify',NOW)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|