207 lines
10 KiB
Python
207 lines
10 KiB
Python
"""Offline publication boundary fixtures. No publisher, signer or live service."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import release_payload as p
|
|
REAL_RUN = subprocess.run
|
|
|
|
|
|
class PublicationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.root = self.base / 'source'
|
|
(self.root / 'core/archipelago').mkdir(parents=True)
|
|
(self.root / 'neode-ui').mkdir()
|
|
self.version = '2.0.1'
|
|
(self.root / 'core/archipelago/Cargo.toml').write_text('[package]\nversion="2.0.1"\n')
|
|
(self.root / 'neode-ui/package.json').write_text('{"version":"2.0.1"}')
|
|
(self.root / 'core/Cargo.lock').write_text('fixture lock')
|
|
self.git('init')
|
|
self.git('add', '.')
|
|
self.git('commit', '-m', 'disposable fixture')
|
|
self.git('tag', '-a', 'v2.0.1', '-m', 'disposable fixture')
|
|
self.backend = self.base / 'archipelago'
|
|
self.backend.write_bytes(Path('/usr/bin/true').read_bytes())
|
|
self.document = self.base / 'manifest.json'
|
|
self.manifest = dict(version=self.version, source_commit=self.git('rev-parse', 'HEAD'),
|
|
signed_by=p.RELEASE_ROOT, signature='forged',
|
|
components=[dict(name='archipelago', sha256=p.digest(self.backend.read_bytes()),
|
|
size_bytes=self.backend.stat().st_size)])
|
|
self.document.write_text(json.dumps(self.manifest))
|
|
# /usr/bin/false is an explicitly rejecting fixture authority, not a
|
|
# production verifier. Only protected input reads are replaced in tests.
|
|
self.verifier = Path('/usr/bin/false').read_bytes()
|
|
self.custody = {'/authority/evidence': b'independent build log fixture'}
|
|
self.receipt = dict(schema='archipelago-production-build-v1', clean=True,
|
|
release_root=p.RELEASE_ROOT, target=p.TARGET, build_profile='release',
|
|
test_root_override=False, source_commit=self.manifest['source_commit'],
|
|
source_tree=self.git('rev-parse', 'HEAD^{tree}'),
|
|
cargo_lock_sha256=p.digest(b'fixture lock'),
|
|
backend_sha256=p.digest(self.backend.read_bytes()), version=self.version,
|
|
build_command='fixture only', evidence=dict(path='/authority/evidence',
|
|
sha256=p.digest(self.custody['/authority/evidence'])))
|
|
self.appointment = dict(schema='archipelago-release-appointment-v1', release_root=p.RELEASE_ROOT,
|
|
verifier=dict(path='/authority/verifier', sha256=p.digest(self.verifier)),
|
|
releases={})
|
|
self.custody['/authority/verifier'] = self.verifier
|
|
self.pin_receipts()
|
|
|
|
def git(self, *args):
|
|
return subprocess.check_output(['git', '-C', str(self.root), '-c', 'user.name=Fixture',
|
|
'-c', 'user.email=fixture@example.invalid', *args], stderr=subprocess.DEVNULL, text=True).strip()
|
|
|
|
def pin_receipts(self):
|
|
verifier_receipt = dict(self.receipt, backend_sha256=p.digest(self.verifier))
|
|
for name, receipt in (('verifier-receipt', verifier_receipt), ('candidate-receipt', self.receipt)):
|
|
raw = json.dumps(receipt).encode()
|
|
path = '/authority/' + name
|
|
self.custody[path] = raw
|
|
reference = dict(path=path, sha256=p.digest(raw))
|
|
if name == 'verifier-receipt':
|
|
self.appointment['verifier']['build_receipt'] = reference
|
|
else:
|
|
self.appointment['releases'][self.version] = reference
|
|
|
|
def read_custody(self, path):
|
|
if path == p.APPOINTMENT:
|
|
return json.dumps(self.appointment).encode()
|
|
return self.custody[str(path)]
|
|
|
|
def verify(self, **kwargs):
|
|
with patch.object(p, 'protected_bytes', side_effect=self.read_custody):
|
|
p.verify_production(self.root, self.backend, self.version, self.document, **kwargs)
|
|
|
|
def test_true_header_and_forged_manifest_cannot_approve_itself(self):
|
|
p.check_native(self.backend) # The exact review counterexample still is ELF.
|
|
# Even explicitly appointing these candidate bytes as a build does not
|
|
# let them override the independent (rejecting) verifier's decision.
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
self.verify()
|
|
|
|
def test_candidate_substitution_fails_before_execution(self):
|
|
self.receipt['backend_sha256'] = '0' * 64
|
|
self.pin_receipts()
|
|
with patch.object(p.subprocess, 'run', wraps=REAL_RUN) as execute:
|
|
with self.assertRaisesRegex(ValueError, 'appointed clean build'):
|
|
self.verify()
|
|
self.assertFalse(any('pass_fds' in call.kwargs for call in execute.call_args_list))
|
|
|
|
def test_wrong_verifier_hash(self):
|
|
self.appointment['verifier']['sha256'] = '0' * 64
|
|
with self.assertRaisesRegex(ValueError, 'verifier hash'):
|
|
self.verify()
|
|
|
|
def test_verifier_replaced_by_true(self):
|
|
self.custody['/authority/verifier'] = self.backend.read_bytes()
|
|
with self.assertRaisesRegex(ValueError, 'verifier hash'):
|
|
self.verify()
|
|
|
|
def test_even_root_appointment_cannot_select_candidate_itself(self):
|
|
self.verifier = self.backend.read_bytes()
|
|
self.custody['/authority/verifier'] = self.verifier
|
|
self.appointment['verifier']['sha256'] = p.digest(self.verifier)
|
|
self.pin_receipts()
|
|
with self.assertRaisesRegex(ValueError, 'own appointed verifier'):
|
|
self.verify()
|
|
|
|
def test_unsigned_and_wrong_root_refused_without_execution(self):
|
|
for changes in ({'signature': ''}, {'signed_by': 'test-root'}):
|
|
self.document.write_text(json.dumps(dict(self.manifest, **changes)))
|
|
with patch.object(p.subprocess, 'run', wraps=REAL_RUN) as execute:
|
|
with self.assertRaisesRegex(ValueError, 'unsigned or wrong-root'):
|
|
self.verify()
|
|
self.assertFalse(any('pass_fds' in call.kwargs for call in execute.call_args_list))
|
|
|
|
def test_wrong_appointment_root(self):
|
|
self.appointment['release_root'] = 'test-root'
|
|
with self.assertRaisesRegex(ValueError, 'root appointment'):
|
|
self.verify()
|
|
|
|
def test_test_root_or_unsigned_uat_receipt_refused(self):
|
|
for key, value in (('release_root', 'test-root'), ('test_root_override', True),
|
|
('schema', 'archipelago-unsigned-uat-input-v2'), ('clean', False)):
|
|
original = self.receipt[key]
|
|
self.receipt[key] = value
|
|
self.pin_receipts()
|
|
with self.subTest(key=key), self.assertRaisesRegex(ValueError, 'production-root build'):
|
|
self.verify()
|
|
self.receipt[key] = original
|
|
|
|
def test_commit_tree_lock_version_binding(self):
|
|
for key, value in (('source_commit', '0'*40), ('source_tree', '0'*40),
|
|
('cargo_lock_sha256', '0'*64), ('version', '1.0.0')):
|
|
original = self.receipt[key]
|
|
self.receipt[key] = value
|
|
self.pin_receipts()
|
|
with self.subTest(key=key), self.assertRaisesRegex(ValueError, 'appointed clean build'):
|
|
self.verify()
|
|
self.receipt[key] = original
|
|
|
|
def test_receipt_and_evidence_tampering(self):
|
|
self.custody['/authority/candidate-receipt'] += b' '
|
|
with self.assertRaisesRegex(ValueError, 'receipt hash'):
|
|
self.verify()
|
|
self.pin_receipts()
|
|
self.custody['/authority/evidence'] += b'changed'
|
|
with self.assertRaisesRegex(ValueError, 'evidence hash'):
|
|
self.verify()
|
|
|
|
def test_dirty_source_refused(self):
|
|
(self.root / 'unreviewed').write_text('dirty')
|
|
with self.assertRaisesRegex(ValueError, 'clean reviewed'):
|
|
self.verify()
|
|
|
|
def test_iso_uses_independent_rejecting_verifier(self):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
self.verify(iso=True)
|
|
|
|
def test_positive_byte_pinning_and_sanitized_execution(self):
|
|
def inspect(argv, **kwargs):
|
|
if 'pass_fds' not in kwargs:
|
|
return REAL_RUN(argv, **kwargs)
|
|
verifier, document = kwargs['pass_fds']
|
|
self.assertEqual(Path(f'/proc/self/fd/{verifier}').read_bytes(), self.verifier)
|
|
self.assertEqual(Path(f'/proc/self/fd/{document}').read_bytes(), self.document.read_bytes())
|
|
with self.assertRaises(OSError):
|
|
os.write(verifier, b'alteration')
|
|
self.assertEqual(argv[1:3], ['ceremony', 'verify'])
|
|
self.assertNotIn('LD_PRELOAD', kwargs['env'])
|
|
with patch.object(p.subprocess, 'run', side_effect=inspect) as execute:
|
|
self.verify()
|
|
self.assertEqual(sum('pass_fds' in call.kwargs for call in execute.call_args_list), 1)
|
|
|
|
def test_mutable_custody_and_candidate_appointment_refused(self):
|
|
# Real filesystem custody walker, with no mock: even read-only JSON
|
|
# below a user-writable ancestor is not an independent appointment.
|
|
self.document.chmod(0o444)
|
|
with self.assertRaisesRegex(ValueError, 'mutable custody'):
|
|
p.protected_bytes(self.document)
|
|
with patch.object(p, 'APPOINTMENT', self.document):
|
|
with self.assertRaisesRegex(ValueError, 'mutable custody'):
|
|
p.appointed_verifier()
|
|
|
|
def test_no_environment_appointment_override(self):
|
|
with patch.dict(os.environ, RELEASE_VERIFIER='/usr/bin/true',
|
|
RELEASE_APPOINTMENT=str(self.document)):
|
|
with patch.object(p, 'protected_bytes', side_effect=FileNotFoundError) as read:
|
|
with self.assertRaises(FileNotFoundError):
|
|
p.appointed_verifier()
|
|
read.assert_called_once_with(Path('/etc/archipelago/release-publication/appointment.json'))
|
|
|
|
def test_duplicate_keys_refused(self):
|
|
with self.assertRaisesRegex(ValueError, 'duplicate'):
|
|
p.strict_json(b'{"signature":"good","signature":"bad"}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|