#!/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()