178 lines
8.8 KiB
Python
178 lines
8.8 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
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 PublicUpdatePreparationTests(unittest.TestCase):
|
|
def component(self, raw=b'qualified fixture bytes'):
|
|
return {'name': 'archipelago', 'new_version': '2.0.4-alpha',
|
|
'size_bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()}
|
|
|
|
def test_copy_matches_signed_size_and_hash_and_makes_read_only_snapshot(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
raw = b'qualified fixture bytes'
|
|
(root / 'archipelago').write_bytes(raw)
|
|
fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
prepare.copy_component(fd, root / 'snapshot', self.component(raw))
|
|
finally:
|
|
os.close(fd)
|
|
self.assertEqual((root / 'snapshot').read_bytes(), raw)
|
|
self.assertEqual(stat.S_IMODE((root / 'snapshot').stat().st_mode), 0o444)
|
|
|
|
def test_source_symlink_is_refused_before_copy(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'outside').write_bytes(b'qualified fixture bytes')
|
|
(root / 'archipelago').symlink_to(root / 'outside')
|
|
fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
with self.assertRaises(OSError):
|
|
prepare.copy_component(fd, root / 'snapshot', self.component())
|
|
finally:
|
|
os.close(fd)
|
|
self.assertFalse((root / 'snapshot').exists())
|
|
|
|
def test_corrupt_copy_is_not_published_read_only(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'archipelago').write_bytes(b'X' * self.component()['size_bytes'])
|
|
fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
prepare.copy_component(fd, root / 'snapshot', self.component())
|
|
finally:
|
|
os.close(fd)
|
|
self.assertEqual(stat.S_IMODE((root / 'snapshot').stat().st_mode), 0o600)
|
|
|
|
def archive(self, path, extras):
|
|
with tarfile.open(path, 'w:gz') as archive:
|
|
for name, kind, mode in [('index.html', tarfile.REGTYPE, 0o644),
|
|
('archipelago-runtime/RELEASE-PAYLOAD.json', tarfile.REGTYPE, 0o644), *extras]:
|
|
entry = tarfile.TarInfo(name)
|
|
entry.type = kind
|
|
entry.mode = mode
|
|
entry.size = 1 if kind == tarfile.REGTYPE else 0
|
|
if kind == tarfile.SYMTYPE:
|
|
entry.linkname = '/etc/shadow'
|
|
archive.addfile(entry, io.BytesIO(b'x') if entry.size else None)
|
|
|
|
def test_archive_refuses_escape_links_duplicates_and_setuid(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
for index, extra in enumerate([
|
|
('../outside', tarfile.REGTYPE, 0o644),
|
|
('link', tarfile.SYMTYPE, 0o777),
|
|
('index.html', tarfile.REGTYPE, 0o644),
|
|
('unsafe', tarfile.REGTYPE, 0o4755)]):
|
|
path = Path(directory) / str(index)
|
|
self.archive(path, [extra])
|
|
with self.assertRaises(ValueError):
|
|
prepare.validate_archive(path)
|
|
good = Path(directory) / 'good'
|
|
self.archive(good, [])
|
|
prepare.validate_archive(good)
|
|
|
|
def test_archive_refuses_file_directory_collisions_in_either_order(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
for index, entries in enumerate([
|
|
[('parent', tarfile.REGTYPE, 0o644), ('parent/child', tarfile.REGTYPE, 0o644)],
|
|
[('parent/child', tarfile.REGTYPE, 0o644), ('parent', tarfile.REGTYPE, 0o644)],
|
|
[('archipelago-runtime', tarfile.REGTYPE, 0o644)]]):
|
|
path = Path(directory) / str(index)
|
|
self.archive(path, entries)
|
|
with self.assertRaises(ValueError):
|
|
prepare.validate_archive(path)
|
|
|
|
def test_archive_required_entries_must_be_files(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'archive'
|
|
with tarfile.open(path, 'w:gz') as archive:
|
|
for name in ['index.html', 'archipelago-runtime/RELEASE-PAYLOAD.json']:
|
|
entry = tarfile.TarInfo(name)
|
|
entry.type = tarfile.DIRTYPE
|
|
archive.addfile(entry)
|
|
with self.assertRaises(ValueError):
|
|
prepare.validate_archive(path)
|
|
|
|
def test_partial_or_dangling_protected_owner_refuses_before_service_probe(self):
|
|
for marker in ['/etc/archipelago-d09', '/etc/archipelago-c01',
|
|
'/var/lib/archipelago-d09', '/run/archipelago-d09']:
|
|
def lstat(path):
|
|
if path == marker:
|
|
return object() # Any inode, including a dangling symlink.
|
|
raise FileNotFoundError(path)
|
|
with self.subTest(marker=marker), patch.object(prepare.os, 'lstat', side_effect=lstat), \
|
|
patch.object(prepare.subprocess, 'check_output') as command:
|
|
with self.assertRaises(ValueError):
|
|
prepare.require_legacy_service()
|
|
command.assert_not_called()
|
|
|
|
def test_ownership_probe_errors_fail_closed(self):
|
|
with patch.object(prepare.os, 'lstat', side_effect=PermissionError('fixture')), \
|
|
patch.object(prepare.subprocess, 'check_output') as command:
|
|
with self.assertRaises(PermissionError):
|
|
prepare.require_legacy_service()
|
|
command.assert_not_called()
|
|
|
|
def test_service_probe_accepts_only_standard_legacy_command(self):
|
|
command = 'LoadState=loaded\nUser=archipelago\nExecStart={ path=/usr/local/bin/archipelago ; argv[]=/usr/local/bin/archipelago ; }\n'
|
|
for output, accepted in [(command, True),
|
|
(command.replace('argv[]=/usr/local/bin/archipelago ;',
|
|
'argv[]=/usr/local/bin/archipelago --managed ;'), False),
|
|
(command.replace('User=archipelago', 'User=root'), False),
|
|
(command.replace('LoadState=loaded', 'LoadState=masked'), False)]:
|
|
with self.subTest(output=output), \
|
|
patch.object(prepare.os, 'lstat', side_effect=FileNotFoundError), \
|
|
patch.object(prepare.subprocess, 'check_output', return_value=output):
|
|
if accepted:
|
|
prepare.require_legacy_service()
|
|
else:
|
|
with self.assertRaises(ValueError):
|
|
prepare.require_legacy_service()
|
|
|
|
def test_manifest_write_is_exclusive_and_readonly(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'manifest'
|
|
prepare.write_manifest(path, b'original signed bytes')
|
|
self.assertEqual(path.read_bytes(), b'original signed bytes')
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o444)
|
|
with self.assertRaises(FileExistsError):
|
|
prepare.write_manifest(path, b'replacement')
|
|
|
|
def test_duplicate_json_fields_are_rejected(self):
|
|
import json
|
|
with self.assertRaises(ValueError):
|
|
json.loads('{"version": "2.0.4-alpha", "version": "2.0.3-alpha"}',
|
|
object_pairs_hook=prepare.pairs)
|
|
|
|
def test_document_shape_refuses_os_policy_unsigned_and_boolean_version(self):
|
|
raw = self.component()
|
|
frontend = dict(raw, name='archipelago-frontend-2.0.4-alpha.tar.gz')
|
|
document = {'version': '2.0.4-alpha', 'application_update': dict(prepare.POLICY),
|
|
'signed_by': prepare.RELEASE_ROOT, 'signature': '0' * 128,
|
|
'components': [raw, frontend]}
|
|
# Shape validation only. authenticate() separately performs real signature verification.
|
|
prepare.validate_document(document, '2.0.3-alpha')
|
|
for bad in [dict(document, os_update={}), dict(document, signature=None),
|
|
dict(document, application_update={'version': True, 'mode': 'signed-application-only'})]:
|
|
with self.assertRaises(ValueError):
|
|
prepare.validate_document(bad, '2.0.3-alpha')
|
|
with self.assertRaises(ValueError):
|
|
prepare.validate_document(document, '2.0.4-alpha')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|