267 lines
15 KiB
Python
267 lines
15 KiB
Python
"""Failure injection for bounded first-generation public OTA adoption retries."""
|
|
import hashlib
|
|
import errno
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import types
|
|
import urllib.error
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from test_public_update_adopt import LegacyAdoption
|
|
|
|
|
|
class ResumeAdoption(LegacyAdoption):
|
|
def simulate_root_owned_incomplete_file(self):
|
|
"""Fixture runs unprivileged; only the fixed discard file has root custody."""
|
|
discard=self.ns['adoption_discard']
|
|
def fixture_discard(path, reason='incomplete authenticated cache'):
|
|
real_fstat=os.fstat
|
|
def root_file_fstat(fd):
|
|
info=real_fstat(fd)
|
|
if Path(os.readlink(f'/proc/self/fd/{fd}')) in (Path(path), Path(path).parent / 'cache-discard.json'):
|
|
fields=list(info);fields[4]=0
|
|
return os.stat_result(fields)
|
|
return info
|
|
with mock.patch.object(os,'fstat',side_effect=root_file_fstat):
|
|
return discard(path, reason)
|
|
self.ns['adoption_discard']=fixture_discard
|
|
|
|
def alias(self):
|
|
return self.ns['PUBLIC_RELEASE_BASE'] + '/v2.0.4-alpha/legacy-manifest-1.9.1-alpha.json'
|
|
|
|
def transaction_dirs(self):
|
|
return sorted(path for path in self.ns['TRANSACTIONS'].iterdir() if path.is_dir())
|
|
|
|
def test_missing_historical_manifest_fails_before_component_download(self):
|
|
self.bytes.pop(self.alias())
|
|
original = self.ns['public_download']
|
|
def missing(url, path, size=None, sha256=None):
|
|
if url == self.alias():
|
|
raise urllib.error.HTTPError(url, 404, 'missing fixture alias', {}, None)
|
|
return original(url, path, size=size, sha256=sha256)
|
|
self.ns['public_download'] = missing
|
|
old_binary = self.ns['LEGACY_BINARY'].read_bytes()
|
|
old_frontend = self.ns['pair_identity'](self.ns['LEGACY_BINARY'], self.ns['LEGACY_FRONTEND'])
|
|
for _ in range(2):
|
|
with self.assertRaises(Exception): self.adopt()
|
|
self.assertEqual(self.ns['LEGACY_BINARY'].read_bytes(), old_binary)
|
|
self.assertEqual(self.ns['pair_identity'](self.ns['LEGACY_BINARY'], self.ns['LEGACY_FRONTEND']), old_frontend)
|
|
self.assertTrue(self.ns['LEGACY_MARKER'].exists())
|
|
bulk = [url for url in self.downloads if url.endswith('/archipelago') or url.endswith('.tar.gz')]
|
|
self.assertEqual(bulk, [], 'historical manifest must precede all large downloads')
|
|
self.assertLessEqual(len(self.transaction_dirs()), 1, 'identical failed restarts grew adoption directories')
|
|
|
|
def test_partial_component_retry_never_trusts_incomplete_bytes(self):
|
|
self.simulate_root_owned_incomplete_file()
|
|
original = self.ns['public_download']
|
|
target = self.frontend_url('2.0.4-alpha')
|
|
once = {'fired': False}
|
|
def partial(url, path, size=None, sha256=None):
|
|
if url == target and not once['fired']:
|
|
once['fired'] = True
|
|
path.write_bytes(self.bytes[url][:max(1, len(self.bytes[url]) // 2)])
|
|
path.chmod(0o664) # interrupted create under a permissive legacy umask
|
|
raise ValueError('injected partial download')
|
|
return original(url, path, size=size, sha256=sha256)
|
|
self.ns['public_download'] = partial
|
|
with self.assertRaisesRegex(ValueError, 'injected partial download'): self.adopt()
|
|
self.ns['public_download'] = original
|
|
state = self.adopt()
|
|
self.assertTrue(state['adoption'])
|
|
self.assertEqual(hashlib.sha256((Path(state['transaction_path']) / 'new-frontend/index.html').read_bytes()).hexdigest(),
|
|
hashlib.sha256(b'new UI').hexdigest())
|
|
self.assertLessEqual(len(self.transaction_dirs()), 1, 'partial retry left duplicate adoption directories')
|
|
|
|
def test_old_frontend_fallback_is_exact_signed_hash_and_size(self):
|
|
old_url = self.frontend_url('1.9.1-alpha')
|
|
fallback = self.ns['PUBLIC_RELEASE_BASE'] + '/v2.0.4-alpha/legacy-frontend-1.9.1-alpha.tar.gz'
|
|
old = self.bytes.pop(old_url)
|
|
self.bytes[fallback] = old
|
|
original = self.ns['public_download']
|
|
def missing_old(url, path, size=None, sha256=None):
|
|
if url == old_url:raise urllib.error.HTTPError(url, 404, 'missing old fixture', {}, None)
|
|
return original(url, path, size=size, sha256=sha256)
|
|
self.ns['public_download'] = missing_old
|
|
state = self.adopt()
|
|
self.assertTrue(state['adoption'])
|
|
self.assertIn(fallback, self.downloads)
|
|
self.assertEqual((Path(state['transaction_path']) / 'old-frontend/index.html').read_bytes(), b'old UI')
|
|
|
|
def test_old_frontend_fallback_rejects_corrupt_bytes(self):
|
|
old_url = self.frontend_url('1.9.1-alpha')
|
|
fallback = self.ns['PUBLIC_RELEASE_BASE'] + '/v2.0.4-alpha/legacy-frontend-1.9.1-alpha.tar.gz'
|
|
old = self.bytes.pop(old_url)
|
|
self.bytes[fallback] = old[:-1] + bytes([old[-1] ^ 1])
|
|
original = self.ns['public_download']
|
|
def missing_old(url, path, size=None, sha256=None):
|
|
if url == old_url:raise urllib.error.HTTPError(url, 404, 'missing old fixture', {}, None)
|
|
return original(url, path, size=size, sha256=sha256)
|
|
self.ns['public_download'] = missing_old
|
|
with self.assertRaises(Exception): self.adopt()
|
|
self.assertTrue(self.ns['LEGACY_MARKER'].exists())
|
|
self.assertEqual(self.ns['BINARY'].read_bytes(), b'new binary')
|
|
|
|
def test_old_frontend_digest_mismatch_never_uses_fallback(self):
|
|
old_url = self.frontend_url('1.9.1-alpha')
|
|
fallback = self.ns['PUBLIC_RELEASE_BASE'] + '/v2.0.4-alpha/legacy-frontend-1.9.1-alpha.tar.gz'
|
|
signed = self.bytes[old_url]
|
|
self.bytes[old_url] = signed[:-1] + bytes([signed[-1] ^ 1])
|
|
self.bytes[fallback] = signed
|
|
with self.assertRaises(Exception):self.adopt()
|
|
self.assertNotIn(fallback, self.downloads, 'digest mismatch is not an HTTP 404/410 fallback')
|
|
self.assertTrue(self.ns['LEGACY_MARKER'].exists())
|
|
|
|
def test_exact_cached_archives_reused_after_later_failure(self):
|
|
schedule = self.ns['schedule_transaction']
|
|
def fail_after_archives(tx, state, bundle):
|
|
raise ValueError('injected later scheduling failure')
|
|
self.ns['schedule_transaction'] = fail_after_archives
|
|
with self.assertRaisesRegex(ValueError, 'injected later scheduling failure'):self.adopt()
|
|
first_downloads = list(self.downloads)
|
|
self.assertGreaterEqual(len(first_downloads), 4)
|
|
self.ns['schedule_transaction'] = schedule
|
|
state = self.adopt()
|
|
self.assertTrue(state['adoption'])
|
|
self.assertEqual(self.downloads, first_downloads, 'exact authenticated archives should be reused')
|
|
self.assertEqual(len(self.transaction_dirs()), 1)
|
|
|
|
def test_orphaned_deterministic_journal_requires_reconciliation(self):
|
|
schedule = self.ns['schedule_transaction']
|
|
self.ns['schedule_transaction'] = lambda tx, state, bundle: (_ for _ in ()).throw(ValueError('injected stop'))
|
|
with self.assertRaisesRegex(ValueError, 'injected stop'):self.adopt()
|
|
self.ns['schedule_transaction'] = schedule
|
|
txs = self.transaction_dirs();self.assertEqual(len(txs), 1)
|
|
(txs[0] / 'state.json').write_text('{}')
|
|
prior_downloads = list(self.downloads)
|
|
with self.assertRaisesRegex(ValueError, 'orphaned legacy admission journal'):self.adopt()
|
|
self.assertEqual(self.transaction_dirs(), txs)
|
|
self.assertEqual(self.downloads, prior_downloads)
|
|
|
|
def test_root_space_refusal_precedes_bulk_and_preserves_inputs(self):
|
|
old_binary = self.ns['LEGACY_BINARY'].read_bytes()
|
|
old_frontend = self.ns['pair_identity'](self.ns['LEGACY_BINARY'], self.ns['LEGACY_FRONTEND'])
|
|
exhausted = types.SimpleNamespace(f_bavail=0, f_frsize=4096, f_bsize=4096)
|
|
with mock.patch.object(self.ns['os'], 'statvfs', return_value=exhausted):
|
|
with self.assertRaises(Exception):self.adopt()
|
|
bulk = [url for url in self.downloads if url.endswith('/archipelago') or url.endswith('.tar.gz')]
|
|
self.assertEqual(bulk, [], 'root free-space refusal must precede bulk downloads')
|
|
self.assertEqual(self.ns['LEGACY_BINARY'].read_bytes(), old_binary)
|
|
self.assertEqual(self.ns['pair_identity'](self.ns['LEGACY_BINARY'], self.ns['LEGACY_FRONTEND']), old_frontend)
|
|
self.assertTrue(self.ns['LEGACY_MARKER'].exists())
|
|
|
|
def test_authenticated_compressed_archive_needs_extracted_space_before_writes(self):
|
|
# The manifest fixture authenticates the new archive, whose expanded
|
|
# member is much larger than its compressed download.
|
|
self.install_release('2.0.4-alpha', b'new binary', {
|
|
'index.html': b'new UI',
|
|
'archipelago-runtime/RELEASE-PAYLOAD.json': b'{}',
|
|
'zero-filled.bin': b'\0' * (8 * 1024 * 1024),
|
|
})
|
|
ample = types.SimpleNamespace(f_bavail=10**9, f_frsize=4096, f_bsize=4096)
|
|
exhausted = types.SimpleNamespace(f_bavail=0, f_frsize=4096, f_bsize=4096)
|
|
extract = self.ns['extract']
|
|
calls = []
|
|
def forbidden_extract(*args, **kwargs):
|
|
calls.append(args)
|
|
return extract(*args, **kwargs)
|
|
self.ns['extract'] = forbidden_extract
|
|
with mock.patch.object(self.ns['os'], 'statvfs', side_effect=[ample, exhausted]):
|
|
with self.assertRaisesRegex(ValueError, 'authenticated frontend expansion'):
|
|
self.adopt()
|
|
self.assertEqual(calls, [], 'space check must precede old and new extraction')
|
|
self.assertTrue(self.ns['LEGACY_MARKER'].exists())
|
|
self.assertEqual(self.ns['BINARY'].read_bytes(), b'new binary')
|
|
|
|
def test_partial_discard_records_one_bounded_private_metadata_file(self):
|
|
self.simulate_root_owned_incomplete_file()
|
|
cache = self.ns['TRANSACTIONS'] / 'partial.tar.gz'
|
|
evidence = cache.parent / 'cache-discard.json'
|
|
for size in (127, 4096, 7):
|
|
cache.write_bytes(b'x' * size)
|
|
cache.chmod(0o664)
|
|
self.ns['adoption_discard'](cache)
|
|
self.assertFalse(cache.exists())
|
|
self.assertEqual(evidence.stat().st_mode & 0o777, 0o600)
|
|
self.assertLess(evidence.stat().st_size, 512)
|
|
record = json.loads(evidence.read_text())
|
|
self.assertEqual(record['path'], cache.name)
|
|
self.assertEqual(record['size_bytes'], size)
|
|
self.assertEqual(record['sha256'], hashlib.sha256(b'x' * size).hexdigest())
|
|
self.assertEqual([p.name for p in cache.parent.glob('cache-discard*')],
|
|
['cache-discard.json'], 'retry must replace one bounded record')
|
|
|
|
def test_cross_filesystem_marker_retention_is_durable_and_exact(self):
|
|
state=self.adopt();tx=Path(state['transaction_path'])
|
|
marker=self.ns['LEGACY_MARKER'];original=marker.read_bytes()
|
|
real_rename=Path.rename;real_fsync=os.fsync;syncs=[]
|
|
def exdev_only(source,destination):
|
|
if source==marker and Path(destination).parent==tx:
|
|
raise OSError(errno.EXDEV,'injected separate data/root filesystems')
|
|
return real_rename(source,destination)
|
|
def counted_fsync(fd):
|
|
syncs.append(fd);return real_fsync(fd)
|
|
with mock.patch.object(Path,'rename',exdev_only),mock.patch.object(os,'fsync',side_effect=counted_fsync):
|
|
self.ns['retain_legacy_marker'](tx,state)
|
|
retained=tx/'legacy-verification-marker.json'
|
|
self.assertEqual(retained.read_bytes(),original)
|
|
self.assertFalse(marker.exists())
|
|
self.assertGreaterEqual(len(syncs),1,'cross-filesystem retained marker must be fsynced')
|
|
marker.write_text('{}')
|
|
with self.assertRaisesRegex(ValueError,'marker changed'):
|
|
self.ns['retain_legacy_marker'](tx,state)
|
|
self.assertEqual(retained.read_bytes(),original)
|
|
self.assertEqual(marker.read_text(),'{}')
|
|
|
|
def test_partial_fixed_retained_marker_is_replaced_from_original(self):
|
|
self.simulate_root_owned_incomplete_file()
|
|
state=self.adopt();tx=Path(state['transaction_path'])
|
|
original=self.ns['LEGACY_MARKER'].read_bytes()
|
|
retained=tx/'legacy-verification-marker.json'
|
|
retained.write_bytes(original[:max(1,len(original)//2)])
|
|
retained.chmod(0o600)
|
|
self.ns['retain_legacy_marker'](tx,state)
|
|
self.assertEqual(retained.read_bytes(),original)
|
|
self.assertFalse(self.ns['LEGACY_MARKER'].exists())
|
|
self.assertEqual(len(self.transaction_dirs()),1)
|
|
|
|
def test_legacy_routing_capture_is_exact_path_nofollow_and_bounded_owner(self):
|
|
routing=self.root/'routing';routing.write_bytes(b'custom accepted nginx route\n');routing.chmod(0o644)
|
|
opened=[];trusted=self.ns['trusted_open']
|
|
def parent_only(path,directory=False):
|
|
opened.append((Path(path),directory))
|
|
return trusted(path,directory=directory)
|
|
self.ns['trusted_open']=parent_only
|
|
self.assertEqual(self.ns['legacy_routing_bytes'](routing),b'custom accepted nginx route\n')
|
|
self.assertIn((routing.parent,True),opened)
|
|
foreign=self.root/'other.conf';foreign.write_bytes(b'foreign\n')
|
|
with self.assertRaisesRegex(ValueError,'unexpected legacy routing path'):
|
|
self.ns['legacy_routing_bytes'](foreign)
|
|
routing.chmod(0o666)
|
|
with self.assertRaisesRegex(ValueError,'untrusted legacy routing file'):
|
|
self.ns['legacy_routing_bytes'](routing)
|
|
routing.chmod(0o644)
|
|
with mock.patch('pwd.getpwnam',return_value=types.SimpleNamespace(pw_uid=os.getuid()+1)):
|
|
with self.assertRaisesRegex(ValueError,'untrusted legacy routing file'):
|
|
self.ns['legacy_routing_bytes'](routing)
|
|
routing.unlink();routing.symlink_to(foreign)
|
|
with self.assertRaises(OSError):self.ns['legacy_routing_bytes'](routing)
|
|
|
|
def test_adoption_runtime_retains_custom_route_without_loosening_normal_custody(self):
|
|
routing=self.root/'routing';routing.write_bytes(b'custom accepted nginx route\n');routing.chmod(0o644)
|
|
state=self.adopt();tx=Path(state['transaction_path'])
|
|
template=tx/'new-frontend/archipelago-runtime/image-recipe/configs/nginx-archipelago.conf'
|
|
template.parent.mkdir(parents=True,exist_ok=True)
|
|
template.write_bytes(b'signed generic template\n')
|
|
self.ns['command']=lambda *args,**kwargs:'' # isolated nginx -t double only
|
|
final=self.ns['runtime_prepare'](tx,state)
|
|
self.assertEqual(template.read_bytes(),routing.read_bytes())
|
|
self.assertEqual(final['runtime']['routing']['old'],final['runtime']['routing']['new'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
import unittest
|
|
unittest.main()
|