559 lines
38 KiB
Python
559 lines
38 KiB
Python
"""Run only inside the disposable Computer, as root; mutations drop to UID 1000."""
|
|
import base64
|
|
import fcntl
|
|
import time
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
|
|
RUNNER = '/var/lib/lazyboy-runner/runner_jobs.py'
|
|
HOME = Path('/home/lazyboy')
|
|
DB = '/var/lib/lazyboy-runner/jobs.sqlite'
|
|
|
|
|
|
class RunnerFiles(unittest.TestCase):
|
|
def setUp(self):
|
|
if os.environ.get('LAZYBOY_RUNNER_TEST_CONTAINER') != '1' or not Path('/.dockerenv').exists():
|
|
self.fail('requires disposable test Computer')
|
|
self.identity = {'computer_id': 'fixture', 'bot_id': 'files-' + uuid.uuid4().hex, 'generation': 1}
|
|
self.path = 'journal-' + uuid.uuid4().hex
|
|
|
|
def call(self, action, body, identity=None, check=True):
|
|
request = {'action': action, 'input': json.dumps(body), 'identity': identity or self.identity}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, timeout=35)
|
|
if check:
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
return json.loads(result.stdout)
|
|
|
|
def write(self, data=b'original'):
|
|
return {'operation_id': uuid.uuid4().hex, 'action': 'write', 'path': self.path,
|
|
'arguments': {'contentBase64': base64.b64encode(data).decode(), 'expectedHash': ''}}
|
|
|
|
def test_runtime_health_is_read_only_and_does_not_attach_a_display(self):
|
|
self.call('file_events', {})
|
|
with sqlite3.connect(DB) as db:
|
|
before = [db.execute('SELECT count(*) FROM ' + table).fetchone()[0]
|
|
for table in ('jobs', 'outbox', 'file_operations', 'file_outbox')]
|
|
for slot in (None, 7):
|
|
result = self.call('runtime_health', {'slot': slot})
|
|
self.assertEqual(result['status'], 'observed')
|
|
self.assertEqual(json.loads(result['stdout']), {'version': 1, 'slot': slot, 'runner': True,
|
|
'desktop': None if slot is None else False, 'browser': None if slot is None else False,
|
|
'viewer': None if slot is None else False})
|
|
for body in ({}, {'slot': True}, {'slot': -1}, {'slot': 8}, {'slot': 0, 'start': True}):
|
|
self.assertEqual(self.call('runtime_health', body, check=False)['error'], 'INVALID_RUNTIME_HEALTH_REQUEST')
|
|
with sqlite3.connect(DB) as db:
|
|
after = [db.execute('SELECT count(*) FROM ' + table).fetchone()[0]
|
|
for table in ('jobs', 'outbox', 'file_operations', 'file_outbox')]
|
|
self.assertEqual(after, before)
|
|
|
|
def test_signed_retirement_recovers_only_verified_postconditions(self):
|
|
body = self.write(b'retired result')
|
|
self.call('file_status', {'operation_id': body['operation_id']}, check=False)
|
|
encoded = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
key = hashlib.sha256((encoded + ':' + body['operation_id']).encode()).hexdigest()
|
|
trigger = 'retirement_fault_' + uuid.uuid4().hex
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('CREATE TRIGGER ' + trigger + " BEFORE UPDATE OF result ON file_operations WHEN OLD.identity='" + encoded + "' BEGIN SELECT RAISE(ABORT, 'retirement receipt fault'); END")
|
|
try:
|
|
self.assertIn('retirement receipt fault', self.call('file_mutate', body, check=False)['error'])
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
with sqlite3.connect(DB) as db:
|
|
metadata = json.loads(db.execute('SELECT metadata FROM file_operations WHERE id=?', (key,)).fetchone()[0])
|
|
for field in ('controller', 'execution'):
|
|
metadata[field]['boot'] = '00000000-0000-0000-0000-000000000000:1'
|
|
db.execute('UPDATE file_operations SET metadata=? WHERE id=?', (json.dumps(metadata), key))
|
|
ordinary = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True, 'resolve': True})
|
|
self.assertEqual(ordinary['status'], 'unknown', 'changed boot alone cannot prove retirement')
|
|
current = {**self.identity, 'computer_id': 'replacement', 'generation': 2}
|
|
management_key = 'ab' * 32
|
|
now = int(time.time())
|
|
claims = {'version': 1, 'current': current, 'origin': self.identity, 'home_key': 'fixture-home',
|
|
'operation_id': body['operation_id'], 'proof_id': 'fixture-ack-' + uuid.uuid4().hex,
|
|
'proof_kind': 'provider_destroy_ack', 'origin_provider_ref': self.identity['computer_id'],
|
|
'issued_at': now, 'expires_at': now + 60}
|
|
payload = json.dumps(claims, separators=(',', ':'))
|
|
signature = hmac.new(bytes.fromhex(management_key), payload.encode(), hashlib.sha256).hexdigest()
|
|
def recover(mac=signature):
|
|
request = {'action': 'file_retirement_recover', 'identity': current,
|
|
'input': json.dumps({'payload': payload, 'signature': mac}),
|
|
'_file_retirement_key': management_key, '_file_retirement_home': 'fixture-home'}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, timeout=35)
|
|
return json.loads(result.stdout)
|
|
self.assertEqual(recover('0' * 64)['error'], 'INVALID_FILE_RETIREMENT_PROOF')
|
|
(HOME / body['path']).write_bytes(b'human conflict')
|
|
self.assertEqual(recover()['status'], 'unknown', 'valid retirement does not prove file postconditions')
|
|
self.assertEqual((HOME / body['path']).read_bytes(), b'human conflict')
|
|
(HOME / body['path']).write_bytes(b'retired result')
|
|
modified = (HOME / body['path']).stat().st_mtime_ns
|
|
result = recover()
|
|
self.assertEqual(result['status'], 'succeeded')
|
|
receipt = json.loads(result['stdout'])
|
|
self.assertEqual(receipt['retirementProofId'], claims['proof_id'])
|
|
self.assertEqual(receipt['quiescenceBasis'], 'provider_destroy_ack')
|
|
self.assertEqual(receipt['verificationBasis'], 'postcondition')
|
|
self.assertFalse(receipt['executionProven'])
|
|
self.assertEqual((HOME / body['path']).stat().st_mtime_ns, modified)
|
|
(HOME / body['path']).write_bytes(b'later human edit')
|
|
self.assertEqual(recover(), result)
|
|
self.assertEqual((HOME / body['path']).read_bytes(), b'later human edit')
|
|
with sqlite3.connect(DB) as db:
|
|
audit = db.execute('SELECT proof_id,claims FROM file_retirement_receipts WHERE operation_id=?', (key,)).fetchone()
|
|
self.assertEqual(audit[0], claims['proof_id'])
|
|
self.assertEqual(json.loads(audit[1]), claims)
|
|
self.assertNotIn(management_key, audit[1])
|
|
|
|
def test_history_reads_owned_receipts_without_resolving_unknown(self):
|
|
body = self.write()
|
|
receipt = self.call('file_mutate', body)
|
|
current = {**self.identity, 'computer_id': 'replacement', 'generation': 2}
|
|
history = {'operation_id': body['operation_id'], 'origin_computer_id': self.identity['computer_id'],
|
|
'origin_generation': self.identity['generation']}
|
|
(HOME / body['path']).write_bytes(b'human edit')
|
|
self.assertEqual(self.call('file_history', history, identity=current), receipt)
|
|
event_request = {'origin_computer_id': self.identity['computer_id'], 'origin_generation': 1, 'cursor': {}}
|
|
historical_events = self.call('file_history_events', event_request, identity=current)['events']
|
|
self.assertEqual(historical_events, self.call('file_events', {})['events'])
|
|
self.assertEqual(len(historical_events), 2)
|
|
ids = [e['id'] for e in historical_events]
|
|
self.call('ack_file_history_events', {**event_request, 'cursor': {'ids': ids}}, identity={**current, 'bot_id': 'foreign'})
|
|
self.assertEqual(self.call('file_history_events', event_request, identity=current)['events'], historical_events)
|
|
self.call('ack_file_history_events', {**event_request, 'cursor': {'ids': ids[:1]}}, identity=current)
|
|
self.assertEqual([e['id'] for e in self.call('file_history_events', event_request, identity=current)['events']], ids[1:])
|
|
foreign = {**current, 'bot_id': 'foreign-' + uuid.uuid4().hex}
|
|
self.assertEqual(self.call('file_history', history, identity=foreign, check=False)['error'], 'UNKNOWN_FILE_OPERATION')
|
|
for invalid in ({**history, 'origin_generation': 3}, {**history, 'resolve': True},
|
|
{**history, 'readback': True}, {**history, 'bot_id': self.identity['bot_id']}):
|
|
self.assertEqual(self.call('file_history', invalid, identity=current, check=False)['error'], 'INVALID_FILE_HISTORY_REQUEST')
|
|
encoded = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
key = hashlib.sha256((encoded + ':' + body['operation_id']).encode()).hexdigest()
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("UPDATE file_operations SET result=NULL,status='unknown' WHERE id=?", (key,))
|
|
events = db.execute('SELECT count(*) FROM file_outbox').fetchone()[0]
|
|
observations = db.execute('SELECT count(*) FROM file_observations').fetchone()[0]
|
|
result = self.call('file_history', history, identity=current)
|
|
self.assertEqual(result['status'], 'unknown')
|
|
self.assertFalse(json.loads(result['stdout'])['observation_performed'])
|
|
self.assertIsNone(json.loads(result['stdout'])['desired_state_matches'])
|
|
self.assertEqual((HOME / body['path']).read_bytes(), b'human edit')
|
|
with sqlite3.connect(DB) as db:
|
|
self.assertEqual(db.execute('SELECT result,status FROM file_operations WHERE id=?', (key,)).fetchone(), (None, 'unknown'))
|
|
self.assertEqual(db.execute('SELECT count(*) FROM file_outbox').fetchone()[0], events)
|
|
self.assertEqual(db.execute('SELECT count(*) FROM file_observations').fetchone()[0], observations)
|
|
|
|
def test_snapshot_quota_preserves_unknown_until_explicit_recovery(self):
|
|
directory = Path('/var/lib/lazyboy-runner/artifacts')
|
|
directory.mkdir(mode=0o700, exist_ok=True)
|
|
padding = directory / ('quota-padding-' + uuid.uuid4().hex)
|
|
fd = os.open(padding, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, 'wb') as stream:
|
|
stream.truncate(256 * 1024 * 1024)
|
|
body = self.write(('unique snapshot ' + uuid.uuid4().hex).encode())
|
|
try:
|
|
self.assertEqual(self.call('file_mutate', body)['status'], 'unknown')
|
|
self.assertTrue((HOME / self.path).exists(), 'effect happened but artifact was not durably captured')
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id']})['status'], 'unknown')
|
|
finally:
|
|
padding.unlink()
|
|
result = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True, 'resolve': True})
|
|
self.assertEqual(result['status'], 'succeeded')
|
|
self.assertIn('artifactRef', json.loads(result['stdout']))
|
|
|
|
def test_explicit_reconciliation_verifies_postconditions_without_replaying(self):
|
|
for action in ('write', 'patch', 'move'):
|
|
with self.subTest(action=action):
|
|
self.path += '-' + action
|
|
desired = b'recovered\x00\xff'
|
|
body = self.write(desired)
|
|
target = self.path
|
|
if action != 'write':
|
|
self.call('file_mutate', self.write(b'original'))
|
|
if action == 'patch':
|
|
body.update(action='patch', arguments={'offset': 0, 'deleteBytes': 8, 'contentBase64': base64.b64encode(desired).decode(), 'expectedHash': hashlib.sha256(b'original').hexdigest()})
|
|
else:
|
|
desired = b'original'
|
|
target = self.path + '-destination'
|
|
body.update(action='move', arguments={'destination': target, 'expectedHash': hashlib.sha256(desired).hexdigest(), 'expectedDestinationHash': ''})
|
|
encoded = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
key = hashlib.sha256((encoded + ':' + body['operation_id']).encode()).hexdigest()
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute(f"CREATE TRIGGER fail_resolve_receipt BEFORE UPDATE OF result ON file_operations WHEN NEW.id='{key}' BEGIN SELECT RAISE(ABORT,'receipt fault'); END")
|
|
try:
|
|
self.call('file_mutate', body, check=False)
|
|
self.assertEqual((HOME / target).read_bytes(), desired)
|
|
request = {'operation_id': body['operation_id'], 'readback': True, 'resolve': True}
|
|
(HOME / target).write_bytes(b'human change')
|
|
self.assertEqual(self.call('file_status', request)['status'], 'unknown')
|
|
(HOME / target).write_bytes(desired)
|
|
before = self.call('file_events', {})['events']
|
|
self.call('file_status', request, check=False)
|
|
self.assertEqual(self.call('file_events', {})['events'], before, 'failed receipt transaction must not emit success')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER fail_resolve_receipt')
|
|
resolved = self.call('file_status', request)
|
|
self.assertEqual(resolved['status'], 'succeeded')
|
|
proof = json.loads(resolved['stdout'])
|
|
self.assertEqual(proof['verificationBasis'], 'postcondition')
|
|
self.assertFalse(proof['executionProven'])
|
|
self.assertTrue(proof['helperQuiescent'])
|
|
self.assertTrue(proof['controllerQuiescent'])
|
|
events = self.call('file_events', {})['events']
|
|
self.assertEqual(self.call('file_status', request), resolved)
|
|
self.assertEqual(self.call('file_events', {})['events'], events)
|
|
(HOME / target).write_bytes(b'later human edit')
|
|
self.assertEqual(self.call('file_mutate', body), resolved)
|
|
self.assertEqual((HOME / target).read_bytes(), b'later human edit')
|
|
snapshot = self.call('file_artifact', {'operation_id': body['operation_id'], 'artifact_ref': proof['artifactRef'],
|
|
'origin_computer_id': self.identity['computer_id'], 'origin_generation': self.identity['generation']})
|
|
self.assertEqual(base64.b64decode(json.loads(snapshot['stdout'])['contentBase64']), desired)
|
|
with sqlite3.connect(DB) as db:
|
|
self.assertEqual(db.execute("SELECT count(*) FROM file_operations WHERE identity=? AND status IN ('accepted','unknown')", (encoded,)).fetchone()[0], 0)
|
|
|
|
def test_durable_receipt_replay_does_not_overwrite_later_changes(self):
|
|
body = self.write()
|
|
first = self.call('file_mutate', body)
|
|
self.assertEqual(first['status'], 'succeeded')
|
|
self.assertTrue(json.loads(first['stdout'])['verified'])
|
|
self.assertEqual((HOME / self.path).stat().st_uid, 1000)
|
|
(HOME / self.path).write_bytes(b'later human edit')
|
|
self.assertEqual(self.call('file_mutate', body), first)
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'later human edit')
|
|
receipt = json.loads(first['stdout'])
|
|
self.assertNotIn('snapshotContentBase64', first['stdout'])
|
|
self.assertNotIn(body['arguments']['contentBase64'], json.dumps(self.call('file_events', {})))
|
|
denied = subprocess.run(['runuser', '-u', 'lazyboy', '--', 'cat', '/var/lib/lazyboy-runner/artifacts/' + receipt['artifactRef']], capture_output=True)
|
|
self.assertNotEqual(denied.returncode, 0)
|
|
self.assertEqual(denied.stdout, b'')
|
|
artifact = {'operation_id': body['operation_id'], 'artifact_ref': receipt['artifactRef'],
|
|
'origin_computer_id': self.identity['computer_id'], 'origin_generation': self.identity['generation']}
|
|
data = self.call('file_artifact', artifact)
|
|
self.assertEqual(base64.b64decode(json.loads(data['stdout'])['contentBase64']), b'original')
|
|
self.assertEqual(self.call('file_artifact', artifact, {**self.identity, 'bot_id': 'other'}, check=False)['error'], 'ARTIFACT_UNAVAILABLE')
|
|
later = self.call('file_artifact', artifact, {**self.identity, 'generation': 2})
|
|
self.assertEqual(later, data, 'immutable historical receipt may be read after generation changes')
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id']}), first)
|
|
Path('/var/lib/lazyboy-runner/file-receipt-proof.json').write_text(json.dumps({'identity': self.identity, 'operation_id': body['operation_id'], 'result': first, 'events': self.call('file_events', {})['events']}))
|
|
changed = {**body, 'arguments': {**body['arguments'], 'contentBase64': 'Y2hhbmdlZA=='}}
|
|
self.assertEqual(self.call('file_mutate', changed, check=False)['error'], 'PAYLOAD_MISMATCH')
|
|
for identity in [{**self.identity, 'bot_id': 'other'}, {**self.identity, 'generation': 2}]:
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id']}, identity, check=False)['error'], 'UNKNOWN_FILE_OPERATION')
|
|
denied = subprocess.run(['runuser', '-u', 'lazyboy', '--', 'cat', DB], capture_output=True)
|
|
self.assertNotEqual(denied.returncode, 0)
|
|
with sqlite3.connect(DB) as db:
|
|
metadata = db.execute('SELECT metadata FROM file_operations WHERE identity=?', (json.dumps(self.identity, sort_keys=True, separators=(',', ':')),)).fetchone()[0]
|
|
self.assertNotIn(body['arguments']['contentBase64'], metadata)
|
|
|
|
def test_patch_move_and_conflict_receipts(self):
|
|
self.call('file_mutate', self.write())
|
|
patch = {'operation_id': uuid.uuid4().hex, 'action': 'patch', 'path': self.path,
|
|
'arguments': {'offset': 0, 'deleteBytes': 8, 'contentBase64': 'bmV3', 'expectedHash': hashlib.sha256(b'original').hexdigest()}}
|
|
self.assertEqual(self.call('file_mutate', patch)['status'], 'succeeded')
|
|
conflicting = {**patch, 'operation_id': uuid.uuid4().hex}
|
|
self.assertEqual(self.call('file_mutate', conflicting)['status'], 'failed')
|
|
move = {'operation_id': uuid.uuid4().hex, 'action': 'move', 'path': self.path,
|
|
'arguments': {'destination': self.path + '-moved', 'expectedHash': hashlib.sha256(b'new').hexdigest(), 'expectedDestinationHash': ''}}
|
|
self.assertEqual(self.call('file_mutate', move)['status'], 'succeeded')
|
|
self.assertFalse((HOME / self.path).exists())
|
|
self.assertEqual((HOME / (self.path + '-moved')).read_bytes(), b'new')
|
|
|
|
def test_patch_target_is_durable_before_effect_and_survives_lost_receipt(self):
|
|
self.call('file_mutate', self.write())
|
|
patch = {'operation_id': uuid.uuid4().hex, 'action': 'patch', 'path': self.path,
|
|
'arguments': {'offset': 0, 'deleteBytes': 8, 'contentBase64': 'bmV3', 'expectedHash': hashlib.sha256(b'original').hexdigest()}}
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
quoted = "'" + identity.replace("'", "''") + "'"
|
|
trigger = 'patch_plan_fault_' + uuid.uuid4().hex
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('CREATE TRIGGER ' + trigger + ' BEFORE UPDATE OF metadata ON file_operations WHEN OLD.identity=' + quoted + " BEGIN SELECT RAISE(ABORT, 'patch plan commit failure'); END")
|
|
try:
|
|
self.assertIn('patch plan commit failure', self.call('file_mutate', patch, check=False)['error'])
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'original')
|
|
self.assertEqual(self.call('file_mutate', patch)['status'], 'unknown')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
patch['operation_id'] = uuid.uuid4().hex
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('CREATE TRIGGER ' + trigger + ' BEFORE UPDATE OF result ON file_operations WHEN OLD.identity=' + quoted + " BEGIN SELECT RAISE(ABORT, 'patch receipt commit failure'); END")
|
|
try:
|
|
self.assertIn('patch receipt commit failure', self.call('file_mutate', patch, check=False)['error'])
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'new')
|
|
observed = self.call('file_status', {'operation_id': patch['operation_id'], 'readback': True})
|
|
self.assertEqual(observed['status'], 'unknown')
|
|
evidence = json.loads(observed['stdout'])
|
|
self.assertTrue(evidence['desired_state_matches'])
|
|
self.assertFalse(evidence['execution_proven'])
|
|
self.assertTrue(evidence['helper_quiescent'])
|
|
(HOME / self.path).write_bytes(b'later human edit')
|
|
observed = self.call('file_status', {'operation_id': patch['operation_id'], 'readback': True})
|
|
self.assertFalse(json.loads(observed['stdout'])['desired_state_matches'])
|
|
self.assertEqual(self.call('file_mutate', patch)['status'], 'unknown')
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'later human edit')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
|
|
def test_child_identity_commit_failure_prevents_handoff(self):
|
|
body = self.write()
|
|
body['path'] += '/nested/file'
|
|
self.call('file_status', {'operation_id': body['operation_id']}, check=False)
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
trigger = 'handoff_failure_' + uuid.uuid4().hex
|
|
quoted = "'" + identity.replace("'", "''") + "'"
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('CREATE TRIGGER ' + trigger + ' BEFORE UPDATE OF metadata ON file_operations WHEN OLD.identity=' + quoted + " AND json_extract(NEW.metadata,'$.execution.pid') IS NOT NULL BEGIN SELECT RAISE(ABORT, 'child identity commit failure'); END")
|
|
try:
|
|
self.assertIn('child identity commit failure', self.call('file_mutate', body, check=False)['error'])
|
|
self.assertFalse((HOME / self.path).exists())
|
|
result = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True})
|
|
self.assertEqual(result['status'], 'unknown')
|
|
self.assertTrue(json.loads(result['stdout'])['helper_quiescent'])
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
|
|
def test_controller_death_keeps_identity_of_blocked_child(self):
|
|
body = self.write()
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
lock = os.open(HOME, os.O_RDONLY | os.O_DIRECTORY)
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
worker = subprocess.Popen(['python3', '-I', RUNNER], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
worker.stdin.write(json.dumps({'action': 'file_mutate', 'identity': self.identity, 'input': json.dumps(body)}))
|
|
worker.stdin.close()
|
|
deadline = time.monotonic() + 5
|
|
metadata = None
|
|
while time.monotonic() < deadline:
|
|
with sqlite3.connect(DB) as db:
|
|
row = db.execute('SELECT metadata FROM file_operations WHERE identity=?', (identity,)).fetchone()
|
|
if row and 'execution' in json.loads(row[0]):
|
|
metadata = json.loads(row[0])
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertIsNotNone(metadata)
|
|
probe = "import runpy,json,sys; m=runpy.run_path(sys.argv[1]); print(json.dumps(m['file_quiescent'](json.loads(sys.argv[2]))))"
|
|
def quiescent():
|
|
result = subprocess.run(['python3', '-I', '-c', probe, RUNNER, json.dumps(metadata)], capture_output=True, text=True, check=True)
|
|
return json.loads(result.stdout)
|
|
self.assertFalse(quiescent())
|
|
worker.kill()
|
|
worker.wait(timeout=5)
|
|
self.assertFalse(quiescent(), 'controller exit alone cannot prove child exit')
|
|
fcntl.flock(lock, fcntl.LOCK_UN)
|
|
deadline = time.monotonic() + 5
|
|
while not quiescent() and time.monotonic() < deadline:
|
|
time.sleep(0.02)
|
|
self.assertTrue(quiescent())
|
|
observed = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True})
|
|
self.assertEqual(observed['status'], 'unknown')
|
|
self.assertTrue(json.loads(observed['stdout'])['helper_quiescent'])
|
|
self.assertTrue(json.loads(observed['stdout'])['desired_state_matches'])
|
|
finally:
|
|
fcntl.flock(lock, fcntl.LOCK_UN)
|
|
os.close(lock)
|
|
if worker.poll() is None:
|
|
worker.kill()
|
|
worker.wait(timeout=5)
|
|
worker.stdout.close()
|
|
worker.stderr.close()
|
|
|
|
def test_journal_quota_refuses_new_effects_but_keeps_replay(self):
|
|
body = self.write()
|
|
receipt = self.call('file_mutate', body)
|
|
prefix = 'quota-' + uuid.uuid4().hex
|
|
with sqlite3.connect(DB) as db:
|
|
db.executemany('INSERT INTO file_operations VALUES (?,?,?,?,?,?,?)',
|
|
[(prefix + str(index), '{}', '', 'accepted', None, '{}', 0) for index in range(1024)])
|
|
try:
|
|
blocked = self.write()
|
|
blocked['path'] += '-blocked'
|
|
self.assertEqual(self.call('file_mutate', blocked, check=False)['error'], 'FILE_JOURNAL_QUOTA')
|
|
self.assertFalse((HOME / blocked['path']).exists())
|
|
self.assertEqual(self.call('file_mutate', body), receipt)
|
|
# Known receipts do not permanently consume active-work slots.
|
|
# Keep all 1024 records; new mutations and old no-replay proof
|
|
# must both work after these operations have completed.
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("UPDATE file_operations SET status='succeeded',result='{}' WHERE id LIKE ?", (prefix + '%',))
|
|
self.assertEqual(self.call('file_mutate', blocked)['status'], 'succeeded')
|
|
(HOME / body['path']).write_bytes(b'later human edit')
|
|
self.assertEqual(self.call('file_mutate', body), receipt)
|
|
self.assertEqual((HOME / body['path']).read_bytes(), b'later human edit')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DELETE FROM file_operations WHERE id LIKE ?', (prefix + '%',))
|
|
|
|
def test_agent_quota_survives_generation_change_and_preserves_peer_admission(self):
|
|
original = self.write()
|
|
receipt = self.call('file_mutate', original)
|
|
prefix = 'agent-quota-' + uuid.uuid4().hex
|
|
historical = {**self.identity, 'computer_id': 'old-provider', 'generation': 1}
|
|
with sqlite3.connect(DB) as db:
|
|
db.executemany('INSERT INTO file_operations VALUES (?,?,?,?,?,?,?)',
|
|
[(prefix + str(i), json.dumps(historical if i % 2 else self.identity,
|
|
sort_keys=True, separators=(',', ':')), '',
|
|
'unknown' if i % 2 else 'accepted', None, '{}', 0) for i in range(256)])
|
|
try:
|
|
blocked = self.write()
|
|
blocked['path'] += '-quota'
|
|
for scope in (self.identity, {**self.identity, 'generation': 2, 'computer_id': 'new-provider'}):
|
|
self.assertEqual(self.call('file_mutate', blocked, identity=scope, check=False)['error'], 'FILE_AGENT_QUOTA')
|
|
self.assertFalse((HOME / blocked['path']).exists())
|
|
self.assertEqual(self.call('file_mutate', original), receipt)
|
|
peer = {**self.identity, 'bot_id': 'peer-' + uuid.uuid4().hex}
|
|
self.assertEqual(self.call('file_mutate', blocked, identity=peer)['status'], 'succeeded')
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("UPDATE file_operations SET status='failed',result='{}' WHERE id=?", (prefix + '0',))
|
|
resumed = self.write()
|
|
resumed['path'] += '-resumed'
|
|
self.assertEqual(self.call('file_mutate', resumed)['status'], 'succeeded')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DELETE FROM file_operations WHERE id LIKE ?', (prefix + '%',))
|
|
|
|
def test_receipt_commit_failure_retains_unknown_after_effect(self):
|
|
body = self.write()
|
|
# Ensure schema exists without admitting this mutation.
|
|
self.call('file_status', {'operation_id': body['operation_id']}, check=False)
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
trigger = 'reject_receipt_' + uuid.uuid4().hex
|
|
quoted = "'" + identity.replace("'", "''") + "'"
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('CREATE TRIGGER ' + trigger + ' BEFORE UPDATE OF result ON file_operations WHEN OLD.identity=' + quoted + " BEGIN SELECT RAISE(ABORT, 'injected receipt failure'); END")
|
|
try:
|
|
failure = self.call('file_mutate', body, check=False)
|
|
self.assertIn('injected receipt failure', failure['error'])
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'original')
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id']})['status'], 'unknown')
|
|
observed = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True})
|
|
self.assertEqual(observed['status'], 'unknown')
|
|
evidence = json.loads(observed['stdout'])
|
|
self.assertTrue(evidence['desired_state_matches'])
|
|
self.assertFalse(evidence['execution_proven'])
|
|
self.assertTrue(evidence['helper_quiescent'])
|
|
self.assertEqual(evidence['source']['sha256'], hashlib.sha256(b'original').hexdigest())
|
|
for foreign in [{**self.identity, 'bot_id': 'other'}, {**self.identity, 'generation': 2}]:
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id'], 'readback': True}, foreign, check=False)['error'], 'UNKNOWN_FILE_OPERATION')
|
|
with sqlite3.connect(DB) as db:
|
|
persisted = json.loads(db.execute('SELECT observation FROM file_observations WHERE id=(SELECT id FROM file_operations WHERE identity=?)', (identity,)).fetchone()[0])
|
|
self.assertEqual(persisted, evidence)
|
|
self.assertNotIn('original', json.dumps(persisted))
|
|
(HOME / self.path).write_bytes(b'later human edit')
|
|
changed = self.call('file_status', {'operation_id': body['operation_id'], 'readback': True})
|
|
self.assertEqual(changed['status'], 'unknown')
|
|
self.assertFalse(json.loads(changed['stdout'])['desired_state_matches'])
|
|
self.assertEqual(self.call('file_mutate', body)['status'], 'unknown')
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'later human edit')
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
|
|
def test_unfinished_intent_never_replays_and_blocks_pause_ack(self):
|
|
body = self.write()
|
|
self.call('file_mutate', body)
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("UPDATE file_operations SET status='accepted',result=NULL WHERE identity=?", (identity,))
|
|
(HOME / self.path).write_bytes(b'later')
|
|
self.assertEqual(self.call('file_mutate', body)['status'], 'unknown')
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'later')
|
|
paused = self.call('pause_agent', {}, check=False)
|
|
self.assertEqual(paused['error'], 'BARRIER_INCOMPLETE')
|
|
self.assertEqual(self.call('file_mutate', self.write(), check=False)['error'], 'SCOPE_PAUSED')
|
|
|
|
def test_outbox_redelivery_ack_is_scoped_and_receipt_remains(self):
|
|
body = self.write()
|
|
receipt = self.call('file_mutate', body)
|
|
events = self.call('file_events', {})['events']
|
|
self.assertEqual([e['payload']['status'] for e in events], ['accepted', 'succeeded'])
|
|
self.assertEqual(events[-1]['payload']['result'], receipt)
|
|
self.assertTrue(all(e['payload']['operation_id'] == body['operation_id'] for e in events))
|
|
self.assertNotIn(body['arguments']['contentBase64'], json.dumps(events))
|
|
self.assertEqual(self.call('file_events', {})['events'], events)
|
|
ids = [e['id'] for e in events]
|
|
for identity in [{**self.identity, 'bot_id': 'other'}, {**self.identity, 'generation': 2}]:
|
|
self.assertEqual(self.call('file_events', {}, identity)['events'], [])
|
|
self.call('ack_file_events', {'ids': ids}, identity)
|
|
self.assertEqual(self.call('file_events', {})['events'], events)
|
|
self.assertEqual(self.call('ack_file_events', {'ids': [True]}, check=False)['error'], 'INVALID_FILE_REQUEST')
|
|
self.call('ack_file_events', {'ids': ids})
|
|
self.call('ack_file_events', {'ids': ids})
|
|
self.assertEqual(self.call('file_events', {})['events'], [])
|
|
self.assertEqual(self.call('file_mutate', body), receipt)
|
|
self.assertEqual(self.call('file_events', {})['events'], [])
|
|
|
|
def test_outbox_failure_is_atomic_with_intent_and_receipt(self):
|
|
body = self.write()
|
|
self.call('file_status', {'operation_id': body['operation_id']}, check=False)
|
|
trigger = 'file_event_failure_' + uuid.uuid4().hex
|
|
identity = json.dumps(self.identity, sort_keys=True, separators=(',', ':'))
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("CREATE TRIGGER " + trigger + " BEFORE INSERT ON file_outbox BEGIN SELECT RAISE(ABORT,'fixture outbox unavailable'); END")
|
|
try:
|
|
self.assertIn('error', self.call('file_mutate', body, check=False))
|
|
self.assertFalse((HOME / self.path).exists())
|
|
with sqlite3.connect(DB) as db:
|
|
self.assertEqual(db.execute('SELECT count(*) FROM file_operations WHERE identity=?', (identity,)).fetchone()[0], 0)
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute("CREATE TRIGGER " + trigger + " BEFORE INSERT ON file_outbox WHEN json_extract(NEW.payload,'$.status') != 'accepted' BEGIN SELECT RAISE(ABORT,'fixture receipt outbox unavailable'); END")
|
|
try:
|
|
self.assertIn('error', self.call('file_mutate', body, check=False))
|
|
self.assertEqual((HOME / self.path).read_bytes(), b'original')
|
|
self.assertEqual(self.call('file_status', {'operation_id': body['operation_id']})['status'], 'unknown')
|
|
self.assertEqual([e['payload']['status'] for e in self.call('file_events', {})['events']], ['accepted'])
|
|
finally:
|
|
with sqlite3.connect(DB) as db:
|
|
db.execute('DROP TRIGGER ' + trigger)
|
|
|
|
def test_journal_failure_prevents_spawning(self):
|
|
body = self.write()
|
|
request = {'action': 'file_mutate', 'input': json.dumps(body), 'identity': self.identity}
|
|
code = "import runpy,json,sys; m=runpy.run_path(sys.argv[1]); db=m['connect'](); db.execute('PRAGMA query_only=ON'); m['file_journal_handle'](db,json.loads(sys.argv[2]),m['FILE_HELPER_SOURCE'],m['barrier_blocked'])"
|
|
result = subprocess.run(['python3', '-I', '-c', code, RUNNER, json.dumps(request)], capture_output=True, text=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('readonly', result.stderr)
|
|
self.assertFalse((HOME / self.path).exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if sys.argv[1:] == ['verify']:
|
|
if os.environ.get('LAZYBOY_RUNNER_TEST_CONTAINER') != '1' or not Path('/.dockerenv').exists():
|
|
raise RuntimeError('requires disposable test Computer')
|
|
proof = json.loads(Path('/var/lib/lazyboy-runner/file-receipt-proof.json').read_text())
|
|
request = {'action': 'file_status', 'identity': proof['identity'], 'input': json.dumps({'operation_id': proof['operation_id']})}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, check=True)
|
|
assert json.loads(result.stdout) == proof['result']
|
|
request = {'action': 'file_history', 'identity': {**proof['identity'], 'computer_id': 'replacement', 'generation': proof['identity']['generation'] + 1},
|
|
'input': json.dumps({'operation_id': proof['operation_id'], 'origin_computer_id': proof['identity']['computer_id'], 'origin_generation': proof['identity']['generation']})}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, check=True)
|
|
assert json.loads(result.stdout) == proof['result']
|
|
request = {'action': 'file_events', 'identity': proof['identity'], 'input': '{}'}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, check=True)
|
|
assert json.loads(result.stdout)['events'] == proof['events']
|
|
request = {'action': 'file_history_events', 'identity': {**proof['identity'], 'computer_id': 'replacement', 'generation': proof['identity']['generation'] + 1},
|
|
'input': json.dumps({'origin_computer_id': proof['identity']['computer_id'], 'origin_generation': proof['identity']['generation'], 'cursor': {}})}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, check=True)
|
|
assert json.loads(result.stdout)['events'] == proof['events']
|
|
receipt = json.loads(proof['result']['stdout'])
|
|
request = {'action': 'file_artifact', 'identity': proof['identity'], 'input': json.dumps({
|
|
'operation_id': proof['operation_id'], 'artifact_ref': receipt['artifactRef'],
|
|
'origin_computer_id': proof['identity']['computer_id'], 'origin_generation': proof['identity']['generation']})}
|
|
result = subprocess.run(['python3', '-I', RUNNER], input=json.dumps(request), capture_output=True, text=True, check=True)
|
|
assert base64.b64decode(json.loads(json.loads(result.stdout)['stdout'])['contentBase64']) == b'original'
|
|
print('historical root file receipt and unacknowledged outbox survived lifecycle change', flush=True)
|
|
else:
|
|
unittest.main()
|