#!/usr/bin/env python3 """Disposable production-path fault checks. No host manager, privilege or network. Uses the actual admission/store/worker implementations, substituting only OS principal checks for the unprivileged fixture and system-manager observations. """ import copy import importlib.util import json import os import pathlib import socket import stat import sys import tempfile import unittest from unittest.mock import patch SCRIPTS = pathlib.Path(__file__).resolve().parents[2] / 'scripts' sys.path.insert(0, str(SCRIPTS)) import d09_store as store import d09_wire as wire import d09_worker as worker owner = worker.owner def fixture_secure(path, directory=False): """Same leaf guarantees with the disposable user's UID, no root escalation.""" path = pathlib.Path(path) st = path.lstat() if st.st_uid != os.getuid() or st.st_mode & 0o022 or stat.S_ISLNK(st.st_mode): raise PermissionError('fixture object') if not directory and (not stat.S_ISREG(st.st_mode) or st.st_nlink != 1): raise PermissionError('fixture file') return os.open(path, os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0)) class Lifecycle(unittest.TestCase): def setUp(self): # Caller chooses TMPDIR outside this test to keep every byte exclusive. self.temp = tempfile.TemporaryDirectory(prefix='d09-functional-') self.root = pathlib.Path(self.temp.name) (self.root / 'observations').mkdir(mode=0o700) (self.root / 'payloads').mkdir(mode=0o700) self.patches = [patch.object(store, 'secure', fixture_secure), patch.object(owner, 'secure', fixture_secure), patch.object(worker, 'secure', fixture_secure), patch.object(owner, 'ROOT', self.root), patch.object(owner, 'RUN', self.root), patch.object(owner, 'PERMIT', self.root / 'permit'), patch.object(owner, 'command', side_effect=AssertionError('host command forbidden')), patch.object(owner, 'service', return_value={'InvocationID': '7' * 32, 'MainPID': '0', 'ActiveState': 'active', 'Job': '0 /'})] for p in self.patches: p.start() self.binding = dict(schema_version=1, target_binding_digest='a' * 64, composition_digest='b' * 64, transaction_id='1' * 32, admission_epoch_id='2' * 32, requester_uid=os.getuid(), requester_gid=os.getgid(), grants={route: 'c' * 64 for route in ('UPDATER', 'AUTOMATIC', 'MANUAL', 'GUARD', 'OBSERVER')}) self.binding['prerequisite'] = dict(grant_digest='0' * 64, backend_sha256='e' * 64, effective_config_digest='f' * 64) self.binding['release_identity'] = ['B', [['archipelago', 'B', 'e' * 64, 1]]] value = store.initial(self.binding) # These transaction tests begin after the separately tested grant-A # activation. No production grant/continuity bypass is installed. value['prerequisite'] = dict(state='STARTED', grant_digest='0' * 64, observation={'invocation_id': '5' * 32}) value['grants'].append('0' * 64) store.atomic(self.root / 'ledger.json', wire.canonical(value)) self.o = owner.Owner.__new__(owner.Owner) import threading self.o.mutex = threading.RLock() self.o.binding = self.binding self.o.boot = '3' * 32 self.o.store = store.Store(self.root, self.binding, admitted=True) self.o.authorize = lambda req, peer: None # Admission fixtures below focus on journal/wire semantics. Actual # object preflight is covered independently with real disposable files. self.o.preflight = lambda: None self.peer = (os.getpid(), os.getuid(), os.getgid()) def tearDown(self): os.close(self.o.store.fd) os.close(self.o.store.head_fd) for p in reversed(self.patches): p.stop() self.temp.cleanup() def request(self, operation='APPLY', request_id='4' * 32): return dict(schema_version=1, operation=operation, route='UPDATER' if operation == 'APPLY' else 'MANUAL', reason='APPLY_REQUEST' if operation == 'APPLY' else 'MANUAL_REQUEST', transaction_id='1' * 32, request_id=request_id, requester_invocation_id='5' * 32, target_binding_digest='a' * 64, composition_digest='b' * 64, grant_digest='c' * 64, expected_sequence=self.o.store.value['sequence'], observer_receipt_digest=None, selected_release_digest=wire.release_digest(self.binding['release_identity']) if operation == 'APPLY' else None, staging_generation='a' * 32 if operation == 'APPLY' else None) def submit(self, request): response = self.o.submit(request, self.peer) wire.bind_submit(request, response) return response def test_actual_a_retry_returns_immutable_acceptance_with_b_head(self): a = self.request() acceptance = self.submit(a)['acceptance'] b = self.request('RECOVER', '6' * 32) self.submit(b) before = (self.root / 'ledger.json').read_bytes() response = self.submit(a) self.assertEqual(response['acceptance'], acceptance) self.assertEqual(response['receipt']['request_digest'], wire.digest(wire.canonical(b))) self.assertEqual(response['receipt']['operation'], 'RECOVER') self.assertEqual(before, (self.root / 'ledger.json').read_bytes()) def test_duplicate_after_later_journal_head_does_not_consume(self): a = self.request() self.submit(a) r = dict(self.o.store.value['receipt'], phase='DISPATCH_INTENT') self.o.store.commit(self.o.store.value, r) before = copy.deepcopy(self.o.store.value) self.assertEqual(self.submit(a)['receipt']['sequence'], 2) self.assertEqual(before, self.o.store.value) def test_changed_body_and_new_stale_key_refuse(self): a = self.request() self.submit(a) changed = dict(a, expected_sequence=1) self.assertEqual(self.submit(changed)['reason'], 'IDEMPOTENCY_MISMATCH') self.assertEqual(self.submit(dict(a, request_id='6' * 32))['reason'], 'STALE_SEQUENCE') @patch('os.fsync') # Saturation checks admission, not power-loss durability. def test_inbox_saturation_keeps_accepted_retries(self, _fsync): a = self.request() self.submit(a) for n in range(1, 64): self.assertEqual(self.submit(self.request('RECOVER', '%032x' % n))['status'], 'ACCEPTED') before = (self.root / 'ledger.json').read_bytes() self.assertEqual(self.submit(self.request('RECOVER', 'f' * 32))['reason'], 'INBOX_FULL') self.assertEqual(self.submit(a)['status'], 'ACCEPTED') self.assertEqual(before, (self.root / 'ledger.json').read_bytes()) def test_owner_restart_never_seeds_anchor(self): self.submit(self.request()) fresh = store.Store(self.root, self.binding) with self.assertRaises(store.ContinuityError): fresh.check() os.close(fresh.fd) def test_zero_sequence_restart_also_requires_bootstrap(self): fresh = store.Store(self.root, self.binding) with self.assertRaises(store.ContinuityError): fresh.check() os.close(fresh.fd) def test_coherent_rollback_is_not_cached_success(self): a = self.request() self.submit(a) old = (self.root / 'ledger.json').read_bytes() self.submit(self.request('RECOVER', '6' * 32)) store.atomic(self.root / 'ledger.json', old) response = self.submit(a) self.assertEqual(response['reason'], 'CONTINUITY_REQUIRED') self.assertEqual(response['reservation'], 'UNKNOWN') self.assertIsNone(response['receipt']) def test_uncommitted_record_tail_is_preserved_and_not_reexecuted(self): a = self.request() original = store.atomic def crash(path, raw, mode=0o600): if pathlib.Path(path).name == 'ledger.json': raise OSError('fault after record fsync, before head publication') return original(path, raw, mode) with patch.object(store, 'atomic', crash): with self.assertRaises(OSError): self.submit(a) self.assertTrue((self.root / 'receipt-0000000000000001.json').exists()) self.assertEqual(self.submit(a)['reservation'], 'UNKNOWN') def test_missing_store_does_not_initialize(self): (self.root / 'ledger.json').unlink() with self.assertRaises(FileNotFoundError): store.Store(self.root, self.binding) self.assertFalse((self.root / 'ledger.json').exists()) def test_inspect_challenge_and_current_head(self): self.submit(self.request()) req = dict(schema_version=1, transaction_id=None, target_binding_digest='a' * 64, challenge='d' * 64) response = self.o.inspect(req, self.peer) wire.bind_inspect(req, response) with self.assertRaises(ValueError): wire.bind_inspect(dict(req, challenge='e' * 64), response) def test_strict_wire_lexical_and_nested_types(self): req = self.request() raw = wire.canonical(req) self.assertEqual(wire.decode('submit_request', raw), req) for bad in (raw + b'\n', b' ' + raw, raw.replace(b'"schema_version":1', b'"schema_version":true'), raw.replace(b'"schema_version":1', b'"schema_version":1.0'), raw.replace(b'"expected_sequence":0', b'"expected_sequence":-0'), raw.replace(b'{', b'{"operation":"APPLY",', 1)): with self.subTest(bad=bad), self.assertRaises(ValueError): wire.decode('submit_request', bad) def test_mutated_response_cannot_cross_bind(self): req = self.request() response = self.submit(req) for field in ('target_binding_digest', 'composition_digest', 'transaction_id'): bad = copy.deepcopy(response) bad['receipt'][field] = ('f' * len(bad['receipt'][field])) bad['receipt_digest'] = wire.digest(wire.canonical(bad['receipt'])) with self.assertRaises(ValueError): wire.bind_submit(req, bad) def worker_fixture(self): self.submit(self.request()) dest = self.root / 'payloads/web-ui' dest.mkdir(mode=0o755) (dest / 'index.html').write_text('old') (dest / 'index.html').chmod(0o644) staged = dest.with_name('web-ui.d09-' + '1' * 32 + '.new') staged.mkdir(mode=0o755) (staged / 'index.html').write_text('new') (staged / 'index.html').chmod(0o644) before, after = owner.object_digest(dest), owner.object_digest(staged) self.binding['objects'] = {'frontend': {'old': before, 'new': after, 'restore': before}} w = worker.Worker.__new__(worker.Worker) w.transaction = '1' * 32 w.binding = self.binding w.value = self.o.store.value def transition(phase, intent=None, consume=None, **fields): v = copy.deepcopy(self.o.store.value) v['intent'] = intent r = dict(v['receipt'], phase=phase, **fields) self.o.store.commit(v, r) w.value = self.o.store.value w.transition = transition w.refresh = lambda: None # live-owner adapter above publishes w.value return w, dest, staged def test_checked_frontend_move_and_reopen_after_first_rename(self): w, dest, staged = self.worker_fixture() original = os.rename def crash(src, dst, **kwargs): if src == staged.name: raise OSError('fault between payload renames') return original(src, dst, **kwargs) with patch.dict(owner.DESTINATIONS, frontend=dest), patch.object(os, 'rename', crash): with self.assertRaises(OSError): w.disposition('frontend', 'APPLY') self.assertFalse(dest.exists()) self.assertTrue(staged.exists()) with patch.dict(owner.DESTINATIONS, frontend=dest): w.disposition('frontend', 'APPLY') self.assertEqual((dest / 'index.html').read_text(), 'new') self.assertEqual((dest.with_name('web-ui.d09-' + '1' * 32 + '.old') / 'index.html').read_text(), 'old') def test_foreign_edit_is_preserved(self): w, dest, staged = self.worker_fixture() (dest / 'index.html').write_text('operator custom edit') with patch.dict(owner.DESTINATIONS, frontend=dest), self.assertRaises(ValueError): w.disposition('frontend', 'APPLY') self.assertEqual((dest / 'index.html').read_text(), 'operator custom edit') self.assertTrue(staged.exists()) def test_missing_backend_snapshot_is_not_success(self): w, dest, staged = self.worker_fixture() (staged / 'index.html').unlink() with patch.dict(owner.DESTINATIONS, frontend=dest), self.assertRaises(ValueError): w.disposition('frontend', 'APPLY') self.assertEqual((dest / 'index.html').read_text(), 'old') def test_complete_requires_independent_observation(self): self.submit(self.request()) r = copy.deepcopy(self.o.store.value['receipt']) r.update(phase='COMPLETE', outcome='COMPLETE') with self.assertRaises(ValueError): self.o.store.commit(self.o.store.value, r) def test_restored_notice_never_repeats_activation(self): w, _, _ = self.worker_fixture() r = dict(self.o.store.value['receipt'], phase='AWAITING_OBSERVATION', config_action='RESTORE', observed_invocation_id='7' * 32) self.o.store.commit(self.o.store.value, r) w.value = self.o.store.value w.execute('fixture-notice', dict(self.request('RECOVER'), reason='MANUAL_REQUEST')) self.assertEqual(w.value['receipt']['phase'], 'AWAITING_OBSERVATION') w.execute('fixture-notice', dict(self.request('RECOVER'), reason='AUTOMATIC_FAILURE')) self.assertEqual(w.value['receipt']['phase'], 'CONFLICT') self.assertEqual(w.value['receipt']['outcome'], 'RECOVERY_REQUIRED') def test_revoked_retry_keeps_acceptance_and_rejects_execution(self): req = self.request() acceptance = self.submit(req)['acceptance'] store.atomic(self.root / 'revocation.json', wire.canonical(dict(target_binding_digest='a' * 64, transaction_id='1' * 32, grant_digest='c' * 64))) response = self.submit(req) self.assertEqual(response['acceptance'], acceptance) self.assertEqual(response['outcome'], 'RECOVERY_REQUIRED') self.assertEqual(response['reason'], 'AUTHORITY_REVOKED') before = (self.root / 'ledger.json').read_bytes() self.submit(req) self.assertEqual(before, (self.root / 'ledger.json').read_bytes()) self.assertEqual(self.submit(self.request('RECOVER', '6' * 32))['reason'], 'AUTHORITY_REVOKED') def test_readmission_changes_epoch_without_rewriting_acceptance(self): req = self.request() acceptance = self.submit(req)['acceptance'] v = self.o.store.value index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')} history = owner.consumption_history(v) record = dict(index, schema_version=1, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'], observed_boot_id=owner.boot_id(), preserved_head_digest=wire.digest(wire.canonical(v['receipt'])), preserved_journal_digest=wire.digest(wire.canonical({'receipts': v['journal'], 'intents': v['intents']})), preserved_consumed_history_digest=wire.digest(wire.canonical(history)), preserved_target_index_digest=wire.digest(wire.canonical(index)), new_epoch_id='8' * 32, new_grant_digest='d' * 64, execution_grant_digest='e' * 64, external_history_digest='f' * 64, external_history_sequence=1, allowed_action='RESUME_SAME_TRANSACTION') store.atomic(self.root / 'readmission.json', wire.canonical(record)) self.o.store.anchor = None old_store = self.o.store self.assertTrue(self.o.readmit()['readmitted']) response = self.submit(req) self.assertEqual(response['acceptance'], acceptance) self.assertEqual(response['receipt']['admission_epoch_id'], '8' * 32) before = (self.root / 'ledger.json').read_bytes() self.assertTrue(self.o.readmit()['readmitted']) self.assertEqual(before, (self.root / 'ledger.json').read_bytes()) self.assertEqual(self.submit(self.request('RECOVER', '6' * 32))['reason'], 'GRANT_INVALID') self.assertEqual(self.submit(dict(self.request('RECOVER', '6' * 32), grant_digest='d' * 64))['status'], 'ACCEPTED') def test_authenticated_independent_observation_can_complete(self): self.submit(self.request()) r = dict(self.o.store.value['receipt'], phase='AWAITING_OBSERVATION', config_action='APPLY', config_pre_digest='d' * 64, config_post_digest='e' * 64, config_current_digest='e' * 64, effective_config_observation_digest='f' * 64, observed_invocation_id='7' * 32, executor=dict(pid=os.getpid(), start_ticks=1, boot_id='3' * 32, incarnation_id='9' * 32)) self.o.store.commit(self.o.store.value, r) directory = self.root / 'observations' directory.mkdir(mode=0o700, exist_ok=True) base = dict(transaction_id='1' * 32, observed_invocation_id='7' * 32) namespace = dict(base, default_terminal_created=True, default_terminal_workspace='/home/archipelago/Work', namespace_writable=['/home/archipelago/Work', '/home/archipelago/.local/state/archipelago/sessions']) preservation = dict(base, preserved={name: {'before': 'a' * 64, 'after': 'a' * 64} for name in ('work', 'history', 'custom_fragments')}) def publish(obj): raw = wire.canonical(obj) hashed = wire.digest(raw) store.atomic(directory / (hashed + '.json'), raw) return hashed observation = {k: r[k] for k in ('transaction_id', 'target_binding_digest', 'composition_digest', 'observed_invocation_id', 'effective_config_observation_digest', 'config_current_digest')} observation.update(namespace_observation_digest=publish(namespace), preservation_observation_digest=publish(preservation)) observed = publish(observation) req = dict(self.request('ACK_OBSERVATION', '6' * 32), route='OBSERVER', reason='OBSERVATION_READY', requester_invocation_id=None, observer_receipt_digest=observed) self.assertEqual(self.submit(req)['status'], 'ACCEPTED') self.assertEqual(self.o.observation(observed), observation) r = dict(self.o.store.value['receipt'], phase='COMPLETE', outcome='COMPLETE', namespace_observation_digest=observation['namespace_observation_digest'], preservation_observation_digest=observation['preservation_observation_digest']) self.o.store.commit(self.o.store.value, r) request = dict(schema_version=1, transaction_id=None, target_binding_digest='a' * 64, challenge='d' * 64) self.assertEqual(self.o.inspect(request, self.peer)['receipt']['outcome'], 'COMPLETE') with patch.object(owner, 'service', return_value={'InvocationID': '8' * 32}): self.assertEqual(self.o.inspect(request, self.peer)['status'], 'UNKNOWN') @patch('os.fsync') def test_worker_snapshot_remains_bounded_at_full_inbox(self, _fsync): self.submit(self.request()) for n in range(1, 64): self.submit(self.request('RECOVER', '%032x' % n)) result = self.o.worker_snapshot(self.o.store.value) self.assertLess(len(wire.canonical(result)), wire.MAX) self.assertEqual(len(result['value']['inbox']), 1) def test_no_public_restart_readmit_or_arbitrary_exec(self): for op in ('activate', 'readmit', 'execute', 'prepare'): with self.subTest(operation=op), self.assertRaises(ValueError): wire.decode('submit_request', wire.canonical(dict(self.request(), operation=op))) def test_guard_unknown_is_nonfatal_and_never_downgrades(self): import subprocess text = (SCRIPTS / 'ota-crash-guard.sh').read_text() client = self.root / 'client' client.write_text('#!/bin/sh\nexit 3\n') client.chmod(0o700) text = text.replace('D09_CLIENT=/usr/libexec/archipelago-d09ctl', 'D09_CLIENT=' + str(client)) # Disable logging and make every legacy mutator fail loudly. No real # command can escape the pre-marker unknown-state branch in this fixture. text = text.replace('set -u', 'set -u\nsystemd-cat() { :; }\ncp() { exit 98; }\nmv() { exit 98; }\nrm() { exit 98; }') result = subprocess.run(['bash'], input=text, capture_output=True, text=True, timeout=5) self.assertEqual(result.returncode, 0, result.stderr) def test_initial_foreign_preflight_refuses_without_stopping(self): import shutil w, dest, staged = self.worker_fixture() # Fresh empty target authority, with one real admitted fixture object. # worker_fixture already accepted one request; use a distinct Owner to # test preflight directly and assert the existing head is untouched. restore = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.restore') shutil.copytree(dest, restore) self.binding['helper_digest'] = wire.digest(b'helper') read = owner.read before = (self.root / 'ledger.json').read_bytes() with patch.dict(owner.DESTINATIONS, {'frontend': dest}, clear=True), patch.object(owner, 'read', side_effect=lambda p: b'helper' if str(p) == '/usr/libexec/archipelago-d09-prepare-workspace' else read(p)): owner.Owner.preflight(self.o) (dest / 'index.html').write_text('operator edit') with self.assertRaises(ValueError): owner.Owner.preflight(self.o) self.assertEqual(before, (self.root / 'ledger.json').read_bytes()) @patch('os.fsync') # Ordered lifecycle functional check; fsync faults above use real fsync. def test_full_apply_and_manual_recovery_own_both_payloads(self, _fsync): import shutil payloads = self.root / 'payloads' paths = {name: payloads / name for name in ('frontend', 'backend', 'config')} paths['frontend'].mkdir(mode=0o755) (paths['frontend'] / 'index.html').write_text('old frontend') (paths['frontend'] / 'index.html').chmod(0o644) paths['backend'].write_text('old backend') paths['backend'].chmod(0o755) objects = {} for name, dest in paths.items(): old = owner.object_digest(dest) new = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.new') restore = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.restore') if name == 'frontend': shutil.copytree(dest, restore) new.mkdir(mode=0o755) (new / 'index.html').write_text('new frontend') (new / 'index.html').chmod(0o644) elif name == 'backend': shutil.copy2(dest, restore) new.write_text('new backend') new.chmod(0o755) else: new.write_text('[Service]\nReadWritePaths=/home/archipelago/Work\n') new.chmod(0o644) objects[name] = dict(old=old, new=owner.object_digest(new), restore=owner.object_digest(restore)) self.binding.update(objects=objects, helper_digest=wire.digest(b'helper'), effective_config={action: wire.digest(b'effective') for action in ('APPLY', 'RESTORE')}) request = self.request() self.submit(request) w = worker.Worker.__new__(worker.Worker) w.transaction = '1' * 32 w.binding = self.binding w.value = self.o.store.value def transition(phase, intent=None, consume=None, **fields): v = copy.deepcopy(self.o.store.value) v['intent'] = intent if consume: v['inbox'].remove(consume) v['consumed'].append(consume) r = dict(v['receipt'], phase=phase, **fields) r['executor'] = dict(pid=os.getpid(), start_ticks=1, boot_id='3' * 32, incarnation_id='9' * 32) self.o.store.commit(v, r) w.value = self.o.store.value w.transition = transition w.observed_start = lambda action: '7' * 32 if action == 'APPLY' else '8' * 32 w.refresh = lambda: None state = dict(MainPID='42', InvocationID='5' * 32, ActiveState='active', Job='0 /') events = [] def command(*args): events.append(args) if args[1] == 'stop': state.update(MainPID='0', InvocationID='', ActiveState='inactive') elif args[1] == 'start': state.update(MainPID='43', InvocationID='7' * 32, ActiveState='active') elif args[1] == 'show': return 'no' if 'NeedDaemonReload' in args else 'effective' return '' with patch.dict(owner.DESTINATIONS, paths, clear=True), patch.object(owner, 'service', side_effect=lambda: dict(state)), \ patch.object(owner, 'command', side_effect=command), patch.object(worker, 'read', return_value=b'helper'): key = w.value['inbox'][0] w.execute(key, request) self.assertEqual(w.value['receipt']['phase'], 'AWAITING_OBSERVATION') self.assertEqual((paths['frontend'] / 'index.html').read_text(), 'new frontend') self.assertEqual(paths['backend'].read_text(), 'new backend') self.assertTrue(paths['config'].exists()) recovery = self.request('RECOVER', '6' * 32) self.submit(recovery) w.value = self.o.store.value w.execute(w.value['inbox'][0], recovery) self.assertEqual((paths['frontend'] / 'index.html').read_text(), 'old frontend') self.assertEqual(paths['backend'].read_text(), 'old backend') self.assertFalse(paths['config'].exists()) self.assertEqual(w.value['receipt']['backend_restoration'], 'SUCCEEDED') self.assertEqual(w.value['receipt']['frontend_restoration'], 'SUCCEEDED') self.assertEqual(w.value['receipt']['outcome'], 'PENDING') self.assertEqual(len([event for event in events if event[1] == 'start']), 2) self.assertEqual(len(w.value['consumed']), 2) self.assertTrue(paths['backend'].with_name('backend.d09-' + '1' * 32 + '.old').exists()) self.assertTrue(paths['backend'].with_name('backend.d09-' + '1' * 32 + '.failed').exists()) if __name__ == '__main__': unittest.main(verbosity=2)