302 lines
16 KiB
Python
302 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""B1/B2 regressions through production client/owner/store, disposable OS adapters."""
|
|
import copy
|
|
import importlib.util
|
|
import io
|
|
import os
|
|
import pathlib
|
|
import shutil
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
|
|
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
fixture = load('d09_fixture', HERE / 'd09-lifecycle.py')
|
|
owner, store, wire = fixture.owner, fixture.store, fixture.wire
|
|
client = load('d09_client', fixture.SCRIPTS / 'd09-lifecycle-client.py')
|
|
|
|
|
|
class Corrections(unittest.TestCase):
|
|
def setUp(self):
|
|
self.case = fixture.Lifecycle()
|
|
self.case.setUp()
|
|
self.o, self.root, self.binding = self.case.o, self.case.root, self.case.binding
|
|
self.sent = []
|
|
self.exchange_hook = None
|
|
self.patches = [patch.object(client, 'CACHE', self.root / 'payloads/client-cache'),
|
|
patch.object(client, 'exchange', side_effect=self.exchange),
|
|
patch.dict(os.environ, {'INVOCATION_ID': '5' * 32})]
|
|
for p in self.patches:
|
|
p.start()
|
|
|
|
def tearDown(self):
|
|
for p in reversed(self.patches):
|
|
p.stop()
|
|
self.case.tearDown()
|
|
|
|
def exchange(self, request, record):
|
|
if record == 'inspect_response':
|
|
return wire.bind_inspect(request, self.o.inspect(request, self.case.peer))
|
|
self.sent.append(copy.deepcopy(request))
|
|
if self.exchange_hook:
|
|
result = self.exchange_hook(request)
|
|
if result is not None:
|
|
return result
|
|
return wire.bind_submit(request, self.o.submit(request, self.case.peer))
|
|
|
|
def active_bytes(self):
|
|
token = wire.digest(wire.canonical([self.binding['transaction_id'], 'MANUAL', '5' * 32]))
|
|
return (client.CACHE / (token + '.json')).read_bytes()
|
|
|
|
def advance(self):
|
|
self.o.store.commit(self.o.store.value, dict(self.o.store.value['receipt'], phase='DISPATCH_INTENT'))
|
|
|
|
def test_review_stale_interleaving_recovers_then_replays_acceptance(self):
|
|
self.case.submit(self.case.request())
|
|
def race(request):
|
|
if len(self.sent) == 1:
|
|
self.advance()
|
|
self.exchange_hook = race
|
|
results = [client.convenience(self.binding, 'recover', 'manual') for _ in range(3)]
|
|
self.assertEqual([r['expected_sequence'] for r in self.sent], [1, 2, 2, 2])
|
|
self.assertNotEqual(self.sent[0]['request_id'], self.sent[1]['request_id'])
|
|
self.assertEqual(len({r['request_id'] for r in self.sent[1:]}), 1)
|
|
self.assertEqual(self.o.store.value['sequence'], 3)
|
|
self.assertEqual(len(self.o.store.value['requests']), 2)
|
|
self.assertEqual(results[0]['acceptance'], results[2]['acceptance'])
|
|
evidence = list(client.CACHE.glob('*.refusal-*.json'))
|
|
self.assertEqual(len(evidence), 1)
|
|
kept = wire.parse(evidence[0].read_bytes())
|
|
self.assertEqual(kept['request'], self.sent[0])
|
|
self.assertEqual(kept['response']['reason'], 'STALE_SEQUENCE')
|
|
self.assertEqual(kept['response']['request_digest'], wire.digest(wire.canonical(self.sent[0])))
|
|
self.assertTrue(any(p.read_bytes() == wire.canonical(self.sent[0]) for p in client.CACHE.glob('*.request.json')))
|
|
|
|
def test_lost_accepted_reply_retries_identical_bytes(self):
|
|
self.case.submit(self.case.request())
|
|
def lost(request):
|
|
if len(self.sent) == 1:
|
|
self.assertEqual(self.o.submit(request, self.case.peer)['status'], 'ACCEPTED')
|
|
raise TimeoutError('reply lost after durable acceptance')
|
|
self.exchange_hook = lost
|
|
with self.assertRaises(TimeoutError):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
before = self.active_bytes()
|
|
result = client.convenience(self.binding, 'recover', 'manual')
|
|
self.assertEqual(result['status'], 'ACCEPTED')
|
|
self.assertEqual(self.sent[0], self.sent[1])
|
|
self.assertEqual(self.active_bytes(), before)
|
|
self.assertEqual(len(self.o.store.value['requests']), 2)
|
|
|
|
def test_unknown_and_wrong_digest_refusals_never_rotate(self):
|
|
self.case.submit(self.case.request())
|
|
for reason, reservation, hashed in [('CONTINUITY_REQUIRED', 'UNKNOWN', None),
|
|
('STALE_SEQUENCE', 'UNKNOWN', None),
|
|
('STALE_SEQUENCE', 'RETAINED', 'f' * 64)]:
|
|
self.exchange_hook = lambda req: owner.refusal(reason, hashed or wire.digest(wire.canonical(req)), reservation)
|
|
with self.assertRaises(PermissionError):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
self.assertEqual(len(self.sent), 3)
|
|
self.assertTrue(all(r == self.sent[0] for r in self.sent))
|
|
self.exchange_hook = None
|
|
self.assertEqual(client.convenience(self.binding, 'recover', 'manual')['status'], 'ACCEPTED')
|
|
self.assertEqual(self.sent[-1], self.sent[0])
|
|
|
|
def test_known_acceptance_cannot_be_replaced_by_refusal(self):
|
|
self.case.submit(self.case.request())
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
before = self.active_bytes()
|
|
self.exchange_hook = lambda req: owner.refusal('STALE_SEQUENCE', wire.digest(wire.canonical(req)), 'RETAINED')
|
|
with self.assertRaises(PermissionError):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
self.assertEqual(len(self.sent), 2)
|
|
self.assertEqual(self.active_bytes(), before)
|
|
|
|
def test_disposition_must_be_durable_before_replacement(self):
|
|
self.case.submit(self.case.request())
|
|
self.exchange_hook = lambda req: self.advance() if len(self.sent) == 1 else None
|
|
original = client.cache_write
|
|
def fail_evidence(path, raw, replace=False):
|
|
if '.refusal-' in path.name:
|
|
raise OSError('fault before refusal publication')
|
|
return original(path, raw, replace)
|
|
with patch.object(client, 'cache_write', side_effect=fail_evidence), self.assertRaises(OSError):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
original_request = self.sent[0]
|
|
self.exchange_hook = None
|
|
self.assertEqual(client.convenience(self.binding, 'recover', 'manual')['status'], 'ACCEPTED')
|
|
self.assertEqual(self.sent[1], original_request)
|
|
self.assertNotEqual(self.sent[2]['request_id'], original_request['request_id'])
|
|
|
|
def test_bounded_repeated_race_does_not_poison_next_call(self):
|
|
self.case.submit(self.case.request())
|
|
self.exchange_hook = lambda req: self.advance()
|
|
with self.assertRaisesRegex(PermissionError, 'STALE_SEQUENCE'):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
self.assertEqual(len(self.sent), 2)
|
|
self.exchange_hook = None
|
|
self.assertEqual(client.convenience(self.binding, 'recover', 'manual')['status'], 'ACCEPTED')
|
|
self.assertEqual(self.sent[1], self.sent[2])
|
|
self.assertNotEqual(self.sent[2]['request_id'], self.sent[3]['request_id'])
|
|
|
|
def test_readmission_grant_refusal_allows_fresh_identity(self):
|
|
self.case.submit(self.case.request())
|
|
# Use the actual readmission path, not a synthetic grant override.
|
|
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))
|
|
old = self.o.store
|
|
self.o.readmit()
|
|
with self.assertRaisesRegex(PermissionError, 'GRANT_INVALID'):
|
|
client.convenience(self.binding, 'recover', 'manual')
|
|
refreshed = dict(self.binding, grants={r: 'd' * 64 for r in self.binding['grants']})
|
|
self.assertEqual(client.convenience(refreshed, 'recover', 'manual')['status'], 'ACCEPTED')
|
|
self.assertEqual(self.sent[-1]['grant_digest'], 'd' * 64)
|
|
self.assertNotEqual(self.sent[-1]['request_id'], self.sent[-2]['request_id'])
|
|
self.assertEqual(len(self.o.store.value['requests']), 2)
|
|
|
|
def prerequisite_fixture(self):
|
|
# Reset only this disposable fixture to the actual provisioned empty
|
|
# state. Production has no reset operation and all real commands refuse.
|
|
os.close(self.o.store.fd)
|
|
os.close(self.o.store.head_fd)
|
|
store.atomic(self.root / 'ledger.json', wire.canonical(store.initial(self.binding)))
|
|
self.o.store = store.Store(self.root, self.binding, admitted=True)
|
|
self.o.incarnation = '6' * 32
|
|
paths = {name: self.root / 'payloads' / name for name in ('backend', 'frontend', 'config')}
|
|
paths['backend'].write_bytes(b'prerequisite backend')
|
|
paths['backend'].chmod(0o755)
|
|
paths['frontend'].mkdir(mode=0o755)
|
|
(paths['frontend'] / 'index.html').write_bytes(b'prerequisite frontend')
|
|
(paths['frontend'] / 'index.html').chmod(0o644)
|
|
objects = {}
|
|
for name, dest in paths.items():
|
|
obj = {'old': owner.object_digest(dest)}
|
|
for generation, suffix in [('new', '.new'), ('restore', '.restore')]:
|
|
stage = dest.with_name(dest.name + '.d09-' + self.binding['transaction_id'] + suffix)
|
|
if generation == 'restore' and not dest.exists():
|
|
pass
|
|
elif dest.is_dir():
|
|
shutil.copytree(dest, stage)
|
|
else:
|
|
stage.write_bytes(dest.read_bytes() if dest.exists() else b'[Service]\n')
|
|
stage.chmod(0o755 if name == 'backend' else 0o644)
|
|
obj[generation] = owner.object_digest(stage)
|
|
objects[name] = obj
|
|
self.binding.update(objects=objects, helper_digest=wire.digest(b'helper'), prerequisite=dict(
|
|
grant_digest='0' * 64, backend_sha256=wire.digest(b'prerequisite backend'),
|
|
effective_config_digest=wire.digest(b'prereq-effective')))
|
|
self.o.preflight = types.MethodType(owner.Owner.preflight, self.o)
|
|
self.state = dict(ActiveState='inactive', MainPID='0', InvocationID='', Job='0 /')
|
|
self.commands = []
|
|
original_read, original_open = owner.read, open
|
|
def command(*args):
|
|
self.commands.append(args)
|
|
if args[1] == 'show':
|
|
return 'prereq-effective'
|
|
if args[1] == 'start':
|
|
self.assertTrue(owner.PERMIT.exists())
|
|
self.assertEqual(self.o.store.value['prerequisite']['state'], 'START_INTENT')
|
|
self.state.update(ActiveState='active', MainPID='4242', InvocationID='5' * 32)
|
|
return ''
|
|
if args[1:] == ('--no-block', 'stop', owner.SERVICE):
|
|
self.state.update(ActiveState='inactive', MainPID='0', InvocationID='')
|
|
return ''
|
|
raise AssertionError('unexpected manager command: ' + str(args))
|
|
def process(pid):
|
|
if self.state['MainPID'] != str(pid):
|
|
raise FileNotFoundError('fixture process exited')
|
|
return dict(pid=pid, start_ticks=123, boot_id=self.o.boot)
|
|
new = [patch.dict(owner.DESTINATIONS, paths, clear=True),
|
|
patch.object(owner, 'service', side_effect=lambda: dict(self.state)),
|
|
patch.object(owner, 'command', side_effect=command),
|
|
patch.object(owner, 'process', side_effect=process),
|
|
patch.object(owner, 'read', side_effect=lambda p: b'helper' if str(p) == '/usr/libexec/archipelago-d09-prepare-workspace' else original_read(p)),
|
|
patch('builtins.open', side_effect=lambda p, *a, **kw: io.BytesIO(b'prerequisite backend') if str(p) == '/proc/4242/exe' else original_open(p, *a, **kw))]
|
|
for p in new:
|
|
p.start()
|
|
self.patches.append(p)
|
|
return paths
|
|
|
|
def test_grant_a_starts_once_then_real_client_can_submit_apply(self):
|
|
self.prerequisite_fixture()
|
|
with self.assertRaisesRegex(PermissionError, 'DISPATCH_PENDING'):
|
|
client.convenience(self.binding, 'prepare', None, self.binding['release_identity'], 'a' * 32)
|
|
self.assertEqual(self.o.store.value['sequence'], 0)
|
|
first = self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))
|
|
self.assertTrue(first['prerequisite_started'])
|
|
self.assertFalse(owner.PERMIT.exists())
|
|
self.assertEqual(self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0)), first)
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
# Exercise the real caller authorization against substituted OS cgroup
|
|
# observation; the client cannot claim a JSON UID or pick a unit.
|
|
self.o.authorize = types.MethodType(owner.Owner.authorize, self.o)
|
|
original = pathlib.Path.read_text
|
|
def cgroup(path, *args, **kwargs):
|
|
if str(path) == '/proc/%d/cgroup' % self.case.peer[0]:
|
|
return '0::/system.slice/archipelago.service\n'
|
|
return original(path, *args, **kwargs)
|
|
with patch.object(pathlib.Path, 'read_text', cgroup):
|
|
self.assertEqual(client.convenience(self.binding, 'prepare', None, self.binding['release_identity'], 'a' * 32)['status'], 'ACCEPTED')
|
|
with self.assertRaises(PermissionError):
|
|
self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
|
|
def test_grant_a_unknown_continuity_and_nonroot_cannot_activate(self):
|
|
self.prerequisite_fixture()
|
|
with self.assertRaises(PermissionError):
|
|
self.o.worker_call({'method': 'activate_prerequisite'}, self.case.peer)
|
|
self.o.store.anchor = None
|
|
with self.assertRaises(store.ContinuityError):
|
|
self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))
|
|
self.assertFalse(owner.PERMIT.exists())
|
|
self.assertFalse(self.commands)
|
|
|
|
def test_grant_a_wrong_artifact_refuses_before_consumption(self):
|
|
paths = self.prerequisite_fixture()
|
|
paths['backend'].write_bytes(b'foreign backend')
|
|
with self.assertRaises(ValueError):
|
|
self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))
|
|
self.assertIsNone(self.o.store.value['prerequisite'])
|
|
self.assertFalse(owner.PERMIT.exists())
|
|
self.assertFalse(self.commands)
|
|
|
|
def test_grant_a_lost_submission_reply_is_observed_never_restarted(self):
|
|
self.prerequisite_fixture()
|
|
original = owner.command.side_effect
|
|
def lost(*args):
|
|
value = original(*args)
|
|
if args[1] == 'start':
|
|
raise TimeoutError('manager start reply lost')
|
|
return value
|
|
with patch.object(owner, 'command', side_effect=lost), self.assertRaises(TimeoutError):
|
|
self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))
|
|
self.assertEqual(self.o.store.value['prerequisite']['state'], 'START_INTENT')
|
|
self.assertTrue(owner.PERMIT.exists())
|
|
self.assertTrue(self.o.worker_call({'method': 'activate_prerequisite'}, (1, 0, 0))['prerequisite_started'])
|
|
self.assertFalse(owner.PERMIT.exists())
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
self.assertEqual(self.state['ActiveState'], 'active')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|