134 lines
6.0 KiB
Python
134 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Root filesystem qualification in disposable mounts; NOT signature/OTA acceptance.
|
|
|
|
The signature verifier and service query are explicit test doubles. All snapshot
|
|
creation, no-follow checks, file copying, replay checks and failure retention use
|
|
the production implementation against real files in isolated tmpfs mounts.
|
|
"""
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import tarfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
def require_isolation():
|
|
if os.geteuid() != 0 or os.environ.get('ARCHY_PUBLIC_PREPARE_FIXTURE') != '1':
|
|
raise RuntimeError('root disposable namespace fixture required')
|
|
mounts = {}
|
|
for line in Path('/proc/self/mountinfo').read_text().splitlines():
|
|
before, after = line.split(' - ', 1)
|
|
mounts[before.split()[4]] = after.split()[0]
|
|
for path in ['/etc', '/var/lib/archipelago', '/var/lib/archipelago-public-updates']:
|
|
if mounts.get(path) != 'tmpfs':
|
|
raise RuntimeError(f'refusing filesystem test without isolated tmpfs: {path}')
|
|
|
|
|
|
require_isolation()
|
|
spec = importlib.util.spec_from_file_location(
|
|
'public_update_prepare', Path(__file__).resolve().parents[1] / 'public_update_prepare.py')
|
|
prepare = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(prepare)
|
|
|
|
|
|
class RootPreparationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
for parent in [prepare.SNAPSHOTS, prepare.STAGING.parent]:
|
|
for entry in parent.iterdir():
|
|
if entry.is_dir() and not entry.is_symlink():
|
|
shutil.rmtree(entry)
|
|
else:
|
|
entry.unlink()
|
|
prepare.STAGING.mkdir()
|
|
(prepare.STAGING / 'archipelago').write_bytes(b'fixture backend: never executed')
|
|
self.archive = 'archipelago-frontend-2.0.4-alpha.tar.gz'
|
|
with tarfile.open(prepare.STAGING / self.archive, 'w:gz') as archive:
|
|
for name in ['index.html', 'archipelago-runtime/RELEASE-PAYLOAD.json']:
|
|
member = tarfile.TarInfo(name)
|
|
member.size = 2
|
|
member.mode = 0o644
|
|
archive.addfile(member, io.BytesIO(b'{}'))
|
|
components = []
|
|
for name in ['archipelago', self.archive]:
|
|
raw = (prepare.STAGING / name).read_bytes()
|
|
components.append({'name': name, 'new_version': '2.0.4-alpha',
|
|
'size_bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()})
|
|
self.document = {'version': '2.0.4-alpha', 'application_update': prepare.POLICY,
|
|
'signed_by': prepare.RELEASE_ROOT, 'signature': '0' * 128,
|
|
'components': components}
|
|
self.raw = json.dumps(self.document).encode()
|
|
|
|
def run_prepare(self):
|
|
output = io.StringIO()
|
|
service = ('LoadState=loaded\nUser=archipelago\n'
|
|
'ExecStart={ path=/usr/local/bin/archipelago ; '
|
|
'argv[]=/usr/local/bin/archipelago ; }\n')
|
|
with patch.object(sys, 'argv', ['fixture']), \
|
|
patch.object(sys, 'stdin', io.TextIOWrapper(io.BytesIO(self.raw))), \
|
|
patch.object(sys, 'stdout', output), \
|
|
patch.object(prepare, 'authenticate', return_value='2.0.3-alpha'), \
|
|
patch.object(prepare.subprocess, 'check_output', return_value=service):
|
|
prepare.main()
|
|
return json.loads(output.getvalue())
|
|
|
|
def test_snapshot_replay_and_source_mutation_do_not_change_admitted_bytes(self):
|
|
result = self.run_prepare()
|
|
ready = Path(result['path'])
|
|
self.assertEqual((ready / 'signed-manifest.json').read_bytes(), self.raw)
|
|
self.assertEqual(result['manifest_sha256'], hashlib.sha256(self.raw).hexdigest())
|
|
for name in ['archipelago', self.archive, 'signed-manifest.json']:
|
|
info = (ready / name).stat()
|
|
self.assertEqual(info.st_uid, 0)
|
|
self.assertEqual(stat.S_IMODE(info.st_mode), 0o444)
|
|
(prepare.STAGING / 'archipelago').write_bytes(b'changed mutable staging')
|
|
self.assertEqual(self.run_prepare(), result)
|
|
self.assertEqual((ready / 'archipelago').read_bytes(), b'fixture backend: never executed')
|
|
|
|
def test_existing_snapshot_corruption_is_not_repaired_from_mutable_staging(self):
|
|
result = self.run_prepare()
|
|
binary = Path(result['path']) / 'archipelago'
|
|
binary.write_bytes(b'X' * binary.stat().st_size) # root fault injection
|
|
with self.assertRaisesRegex(ValueError, 'digest changed'):
|
|
self.run_prepare()
|
|
self.assertEqual(binary.read_bytes(), b'X' * binary.stat().st_size)
|
|
|
|
def test_failed_preparation_retains_private_partial_and_can_retry_fresh(self):
|
|
with patch.object(prepare, 'write_manifest', side_effect=OSError('injected fsync/write failure')):
|
|
with self.assertRaisesRegex(OSError, 'injected'):
|
|
self.run_prepare()
|
|
partials = list(prepare.SNAPSHOTS.glob('.partial-*'))
|
|
self.assertEqual(len(partials), 1)
|
|
self.assertEqual(stat.S_IMODE(partials[0].stat().st_mode), 0o700)
|
|
self.assertEqual(len(list(prepare.SNAPSHOTS.glob('[0-9a-f]' * 64))), 0)
|
|
self.run_prepare()
|
|
self.assertTrue(partials[0].exists())
|
|
|
|
def test_lock_symlink_refuses_without_touching_destination(self):
|
|
sentinel = prepare.STAGING / 'sentinel'
|
|
sentinel.write_bytes(b'preserve')
|
|
(prepare.SNAPSHOTS / '.prepare.lock').symlink_to(sentinel)
|
|
with self.assertRaises(OSError):
|
|
self.run_prepare()
|
|
self.assertEqual(sentinel.read_bytes(), b'preserve')
|
|
|
|
def test_dangling_owner_directory_refuses_without_preparing(self):
|
|
marker = Path('/etc/archipelago-d09')
|
|
marker.symlink_to('/nonexistent-fixture-owner')
|
|
try:
|
|
with self.assertRaisesRegex(ValueError, 'protected updater ownership'):
|
|
self.run_prepare()
|
|
self.assertEqual(list(prepare.SNAPSHOTS.iterdir()), [])
|
|
finally:
|
|
marker.unlink()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|