135 lines
6.6 KiB
Python
135 lines
6.6 KiB
Python
"""Unprivileged filesystem/control tests, NOT installed or OS-transaction UAT."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SPEC = importlib.util.spec_from_file_location('bootstrap', Path(__file__).parents[1] / 'os_updater_bootstrap.py')
|
|
B = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(B)
|
|
|
|
|
|
class BootstrapTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
for name in ('etc', 'usr/local/bin', 'var/lib', 'payload', 'evidence', 'wallet'):
|
|
(self.root / name).mkdir(parents=True)
|
|
(self.root / 'etc/machine-id').write_bytes(b'fixture-machine\n')
|
|
(self.root / 'usr/local/bin/archipelago').write_bytes(b'old native')
|
|
(self.root / 'payload/owner.py').write_bytes(b'# inert fixture owner\n')
|
|
(self.root / 'evidence/inventory.json').write_bytes(b'public inventory')
|
|
(self.root / 'wallet/sentinel').write_bytes(b'never restored')
|
|
self.plan = {'schema': 1, 'machine_id_sha256': B.digest(b'fixture-machine\n'),
|
|
'current': {'version': '2.0.0-alpha+uat', 'source_sha256': 'a'*64,
|
|
'backend_sha256': B.digest(b'old native')},
|
|
'candidate': {'version': '2.0.1-alpha', 'source_sha256': 'b'*64,
|
|
'backend_sha256': 'c'*64},
|
|
'admin': {'uid': 1001, 'service_uid': 1000, 'access_receipt_sha256': B.digest(b'public inventory')},
|
|
'observations': {'evidence/inventory.json': B.digest(b'public inventory')},
|
|
'files': {'owner.py': B.digest(b'# inert fixture owner\n')}}
|
|
self.plan['candidate']['source_sha256'] = B.digest(json.dumps(
|
|
self.plan['files'], sort_keys=True, separators=(',', ':')).encode())
|
|
|
|
def prepare(self):
|
|
approval = self.root / 'approval.json'
|
|
raw = json.dumps(self.plan).encode()
|
|
approval.write_bytes(raw)
|
|
# Emulate root custody only in this unprivileged filesystem fixture.
|
|
# Production secure() is tested separately and is never bypassed by CLI.
|
|
def fixture_open(path, uid=0, directory=False):
|
|
return os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
with patch.object(B.os, 'geteuid', return_value=0), patch.object(B, 'secure', fixture_open):
|
|
return B.prepare(self.root, approval, B.digest(raw), self.root / 'payload')
|
|
|
|
def test_atomic_inert_generation_preserves_old_code_and_data(self):
|
|
result = self.prepare()
|
|
self.assertEqual(result['state'], 'PREPARED_NOT_ACTIVATED')
|
|
prepared = self.root / B.BASE / 'prepared'
|
|
self.assertEqual(json.loads((prepared / 'control.json').read_bytes()), result)
|
|
self.assertEqual((self.root / 'wallet/sentinel').read_bytes(), b'never restored')
|
|
self.assertEqual((self.root / 'usr/local/bin/archipelago').read_bytes(), b'old native')
|
|
with self.assertRaisesRegex(ValueError, 'existing bootstrap'):
|
|
self.prepare()
|
|
|
|
def test_unresolved_ledgers_refuse_without_rewriting(self):
|
|
for name in B.BLOCKERS:
|
|
with self.subTest(name=name):
|
|
p = self.root / name
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_bytes(b'unresolved')
|
|
with self.assertRaisesRegex(ValueError, 'existing authority'):
|
|
self.prepare()
|
|
self.assertEqual(p.read_bytes(), b'unresolved')
|
|
p.unlink()
|
|
|
|
def test_interruption_retained_and_retry_refused(self):
|
|
original = B.write
|
|
def fail_control(path, raw):
|
|
if path.name == 'control.json':
|
|
raise OSError('injected disk failure')
|
|
original(path, raw)
|
|
with patch.object(B, 'write', fail_control), self.assertRaises(OSError):
|
|
self.prepare()
|
|
self.assertFalse((self.root / B.BASE / 'prepared').exists())
|
|
self.assertEqual(len(list((self.root / B.BASE).glob('pending-*'))), 1)
|
|
with self.assertRaisesRegex(ValueError, 'unresolved bootstrap'):
|
|
self.prepare()
|
|
|
|
def test_downgrade_and_same_version_refused(self):
|
|
for version in ('1.9.0-alpha', '2.0.0-alpha', '2.0.0-alpha+uat'):
|
|
self.plan['candidate']['version'] = version
|
|
with self.assertRaisesRegex(ValueError, 'downgrade'):
|
|
self.prepare()
|
|
|
|
def test_source_inventory_and_backend_drift(self):
|
|
for name in ('payload/owner.py', 'evidence/inventory.json', 'usr/local/bin/archipelago'):
|
|
with self.subTest(name=name):
|
|
p = self.root / name
|
|
before = p.read_bytes()
|
|
p.write_bytes(b'changed')
|
|
with self.assertRaisesRegex(ValueError, 'drift'):
|
|
self.prepare()
|
|
p.write_bytes(before)
|
|
self.assertFalse((self.root / B.BASE).exists())
|
|
|
|
def test_shared_admin_and_traversal_refuse(self):
|
|
self.plan['admin']['uid'] = 1000
|
|
with self.assertRaisesRegex(ValueError, 'shared'):
|
|
self.prepare()
|
|
self.plan['admin']['uid'] = 1001
|
|
self.plan['files'] = {'../wallet/sentinel': 'a'*64}
|
|
with self.assertRaisesRegex(ValueError, 'payload names'):
|
|
self.prepare()
|
|
|
|
def test_real_custody_rejects_untrusted_input(self):
|
|
with self.assertRaisesRegex(ValueError, 'custody'):
|
|
B.read(self.root / 'payload/owner.py')
|
|
|
|
def test_duplicate_keys_refused(self):
|
|
with self.assertRaisesRegex(ValueError, 'duplicate'):
|
|
B.decode(b'{"schema":1,"schema":1}')
|
|
|
|
def test_unauthenticated_approval_and_nonroot_refuse(self):
|
|
with patch.object(B.os, 'geteuid', return_value=1000):
|
|
with self.assertRaisesRegex(ValueError, 'root administrator'):
|
|
B.prepare(self.root, self.root / 'absent', '0'*64, self.root / 'payload')
|
|
with patch.object(B.os, 'geteuid', return_value=0), patch.object(
|
|
B, 'secure', side_effect=lambda p, **kw: os.open(p, os.O_RDONLY)), patch.object(
|
|
B, 'read', return_value=b'not the approved bytes'):
|
|
with self.assertRaisesRegex(ValueError, 'authenticated approval digest'):
|
|
B.prepare(self.root, self.root / 'absent', '0'*64, self.root / 'payload')
|
|
|
|
def test_candidate_source_inventory_pin_refuses(self):
|
|
self.plan['candidate']['source_sha256'] = '0'*64
|
|
with self.assertRaisesRegex(ValueError, 'source inventory digest'):
|
|
self.prepare()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|