88 lines
4.1 KiB
Python
88 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Harness safety regressions only; these are never installed-OTA qualification."""
|
|
import importlib.util
|
|
import json
|
|
import hashlib
|
|
from pathlib import Path
|
|
import tempfile
|
|
import tarfile
|
|
import shutil
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('installed_vm', Path(__file__).with_name('os_ota_installed_vm.py'))
|
|
vm = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(vm)
|
|
|
|
|
|
class PrivateDiskBoundary(unittest.TestCase):
|
|
@unittest.skipUnless(shutil.which('/usr/sbin/mkfs.ext4') and shutil.which('/usr/sbin/debugfs'), 'requires ext4 tools')
|
|
def test_real_ext4_injection_requires_exact_readback(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root=Path(directory)
|
|
disk=root/'fixture.raw'
|
|
with disk.open('wb') as stream:
|
|
stream.truncate(8*1024**2)
|
|
vm.run(['/usr/sbin/mkfs.ext4','-F',disk],root/'mkfs.log')
|
|
source=root/'source'
|
|
source.write_bytes(b'private disk injection\n')
|
|
vm.inject(disk,source,'/fixture',root/'inject.log')
|
|
self.assertEqual((root/'inject.readback').read_bytes(),source.read_bytes())
|
|
vm.inject(disk,source,'/private-fixture',root/'private.log',private=True)
|
|
self.assertEqual((root/'private.readback').stat().st_mode & 0o777,0o600)
|
|
self.assertEqual((root/'private.readback').read_bytes(),source.read_bytes())
|
|
with self.assertRaises(FileNotFoundError):
|
|
vm.inject(disk,source,'/missing-parent/fixture',root/'failed.log')
|
|
|
|
def test_guest_source_snapshot_is_root_owned_and_hashes_captured_bytes(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root=Path(directory)
|
|
(root/'scripts').mkdir()
|
|
source=root/'scripts/fixture.py'
|
|
source.write_bytes(b'print("private fixture")\n')
|
|
source.chmod(0o777)
|
|
(root/'scripts/alias.py').symlink_to(source)
|
|
(root/'scripts/excluded.key').write_text('not source')
|
|
destination=root/'source.tar'
|
|
manifest=vm.source_archive(root,destination)
|
|
with tarfile.open(destination) as archive:
|
|
self.assertEqual(archive.getnames(),['scripts/fixture.py'])
|
|
entry=archive.getmembers()[0]
|
|
self.assertEqual((entry.uid,entry.gid,entry.mode),(0,0,0o755))
|
|
self.assertEqual(hashlib.sha256(archive.extractfile(entry).read()).hexdigest(),manifest[entry.name])
|
|
|
|
def test_devices_and_unreceipted_files_are_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
vm.owned_disk(Path('/dev/null'))
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
disk = Path(directory)/'installed.raw'
|
|
disk.touch()
|
|
with self.assertRaises(ValueError):
|
|
vm.owned_disk(disk)
|
|
|
|
def test_failed_or_other_disk_receipts_do_not_appoint_disk(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
disk = Path(directory)/'installed.raw'
|
|
disk.touch()
|
|
receipt = disk.parent/'result.json'
|
|
for value in (dict(action='prepare',passed=False,disk=str(disk)),
|
|
dict(action='boot',passed=True,disk=str(disk)),
|
|
dict(action='prepare',passed=True,disk='/dev/vda')):
|
|
receipt.write_text(json.dumps(value))
|
|
with self.assertRaises(ValueError):
|
|
vm.owned_disk(disk)
|
|
receipt.write_text(json.dumps(dict(action='prepare',passed=True,disk=str(disk))))
|
|
self.assertEqual(vm.owned_disk(disk),disk)
|
|
alias = disk.parent/'alias.raw'
|
|
alias.symlink_to(disk)
|
|
with self.assertRaises(ValueError):
|
|
vm.owned_disk(alias)
|
|
|
|
def test_debugfs_command_injection_is_rejected_before_execution(self):
|
|
for name in ('/tmp/a b','/tmp/a\nwrite /dev/sda','/tmp/a"b','/tmp/a\\b'):
|
|
with self.assertRaises(ValueError):
|
|
vm.inject(Path('/dev/null'),Path(name),'/opt/fixture',Path('/dev/null'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|