"""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()