import importlib.util import json from pathlib import Path import subprocess import tempfile import tomllib import unittest try: from hydra import image except ImportError: image = None class ImageTests(unittest.TestCase): def setUp(self): self.assertIsNotNone(image, 'image pipeline is missing') self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) / 'rootfs' self.root.mkdir() def file(self, relative, content='preserved'): path = self.root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) return path def kernel(self, version='6.12.58-gentoo-dist'): (self.root / 'lib/modules' / version).mkdir(parents=True) self.file('boot/vmlinuz-' + version, 'kernel') return version def test_kernel_from_payload_not_running_host(self): self.kernel() self.assertEqual(image.kernel_version(self.root), '6.12.58-gentoo-dist') def test_ambiguous_or_missing_kernel_refused(self): with self.assertRaises(image.ImageError): image.kernel_version(self.root) self.kernel('6.1-one') self.kernel('6.1-two') with self.assertRaises(image.ImageError): image.kernel_version(self.root) def test_module_directory_without_kernel_refused(self): self.kernel() (self.root / 'boot/vmlinuz-6.12.58-gentoo-dist').unlink() with self.assertRaises(image.ImageError): image.kernel_version(self.root) def test_export_sanitization_preserves_update_inputs(self): preserved = ['var/db/pkg/sys-apps/portage-3/USE', 'var/lib/portage/world', 'var/db/repos/gentoo/profiles/repo_name', 'usr/share/openpgp-keys/gentoo-release.asc', 'etc/portage/patches/app-emulation/qemu/patch.patch', 'etc/portage/make.conf'] removed = ['etc/portage/gnupg/private-keys-v1.d/key', 'etc/portage/gnupg.hydra-untrusted/key', 'var/lib/hydra-build/packages.log', 'root/.ssh/id_ed25519', 'root/.bash_history', 'var/cache/binpkgs/package.tbz2', 'var/tmp/portage/build/log', 'run/container-state', 'var/log/emerge.log', 'var/lib/systemd/random-seed', 'hydra-source/private', 'hydra-cache/private'] for name in preserved + removed: self.file(name) self.file('etc/machine-id', 'builder-id') self.file('etc/resolv.conf', 'nameserver 127.0.0.1') image.sanitize_rootfs(self.root) for name in preserved: self.assertEqual((self.root / name).read_text(), 'preserved', name) for name in removed: self.assertFalse((self.root / name).exists(), name) self.assertEqual((self.root / 'etc/machine-id').read_text(), '') self.assertEqual((self.root / 'etc/hostname').read_text(), 'hydra-os\n') self.assertTrue((self.root / 'etc/resolv.conf').is_symlink()) def test_sanitization_cannot_follow_exported_symlink_to_host(self): outside = Path(self.tmp.name) / 'host' outside.mkdir() secret = outside / 'machine-id' secret.write_text('host-id') (self.root / 'etc').symlink_to(outside) with self.assertRaises(image.ImageError): image.sanitize_rootfs(self.root) self.assertEqual(secret.read_text(), 'host-id') def test_installed_configuration_restores_package_sandbox_defaults(self): self.kernel() self.file('etc/portage/make.conf', 'FEATURES="getbinpkg binpkg-request-signature -usersandbox -sandbox -network-sandbox -ipc-sandbox -pid-sandbox"\n') image.configure_rootfs(self.root, Path(__file__).resolve().parents[1]) features = (self.root / 'etc/portage/make.conf').read_text() for option in ('-usersandbox', '-sandbox', '-network-sandbox', '-ipc-sandbox', '-pid-sandbox'): self.assertNotIn(option, features) self.assertIn('binpkg-request-signature', features) self.assertEqual((self.root / 'usr/share/hydra/build-manifest.json').is_file(), True) def test_rootfs_configuration_separates_live_overrides_and_installed_auth(self): source = Path(__file__).resolve().parents[1] image.copy_overlay(self.root, source / 'image/rootfs') installed = tomllib.loads((self.root / 'etc/greetd/config.toml').read_text()) live = tomllib.loads((self.root / 'etc/greetd/live.toml').read_text()) self.assertNotIn('initial_session', installed) self.assertEqual(live['initial_session']['user'], 'hydra-live') manifest = json.loads((self.root / 'etc/hydra/live-mode/remove-on-install.json').read_text()) for name in manifest: self.assertTrue((self.root / name.lstrip('/')).is_file(), name) self.assertIn('/etc/sudoers.d/90-hydra-live', manifest) self.assertIn('/etc/greetd/live.toml', manifest) self.assertNotIn('root=live:', (self.root / 'etc/dracut.conf.d/50-hydra.conf').read_text()) defaults = tomllib.loads((self.root / 'etc/hydra/defaults.toml').read_text()) self.assertEqual(defaults, dict(network='bridge', bridge='hydrabr0', width=1920, height=1080, refresh_millihz=144000)) def test_overlay_cannot_write_through_host_symlink(self): outside = Path(self.tmp.name) / 'host' outside.mkdir() (self.root / 'etc').symlink_to(outside) with self.assertRaises(image.ImageError): image.copy_overlay(self.root, Path(__file__).resolve().parents[1] / 'image/rootfs') self.assertEqual(list(outside.iterdir()), []) def test_live_kernel_args_match_label_and_never_copy_payload_to_ram(self): args = image.live_kernel_args() self.assertIn('root=live:CDLABEL=HYDRA_OS', args) self.assertIn('rd.live.overlay.overlayfs=1', args) self.assertNotIn('rd.live.ram=1', args) cfg = (Path(__file__).resolve().parents[1] / 'image/grub.cfg').read_text() linux = next(line.strip() for line in cfg.splitlines() if line.strip().startswith('linux ')) self.assertEqual(linux.split()[2:], args) def test_manifest_records_actual_kernel_packages_and_source_content(self): self.kernel() self.file('var/db/pkg/gui-wm/labwc-0.20.0/USE', 'X png') self.file('var/db/pkg/gui-wm/labwc-0.20.0/BUILD_TIME', '123') source = Path(__file__).resolve().parents[1] result = image.payload_manifest(self.root, source) self.assertEqual(result['kernel'], '6.12.58-gentoo-dist') self.assertEqual(result['packages'], [{'package': 'gui-wm/labwc-0.20.0', 'use': ['X', 'png'], 'build_time': '123'}]) self.assertIn('hydra/session.py', result['source_files']) self.assertIn('image/rootfs/etc/hydra/defaults.toml', result['source_files']) def test_artifact_checkpoint_checks_bytes_not_flags(self): path = self.file('artifacts/squashfs.img', 'actual-payload') state = {'artifacts': {'squashfs.img': image.digest(path)}} self.assertTrue(image.artifacts_valid(path.parent, state)) path.write_text('changed-payload') self.assertFalse(image.artifacts_valid(path.parent, state)) path.unlink() self.assertFalse(image.artifacts_valid(path.parent, state)) self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {}})) self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {'../escape': '0' * 64}})) class ImageBuildTests(unittest.TestCase): def setUp(self): from hydra.build import Builder self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.project = Path(self.tmp.name) self.builder = Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman') def test_changed_canonical_patch_invalidates_package_input_identity(self): patches = self.project / 'patches' patches.mkdir() patch = patches / 'qemu-virtio-refresh-rate.patch' patch.write_text('old patch') before = self.builder.input_id() patch.write_text('new patch') self.assertNotEqual(self.builder.input_id(), before) def test_running_package_job_blocks_export(self): from hydra.build import BuildError self.builder.active_package_job = lambda: True self.builder.pod = lambda *args, **kw: self.fail('Running builder was modified or exported') self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing') with self.assertRaisesRegex(BuildError, 'running'): self.builder.rootfs() def test_unverified_packages_block_export(self): from hydra.build import BuildError self.builder.active_package_job = lambda: False self.builder.pod = lambda *args, **kw: self.fail('Unverified builder was exported') self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing') with self.assertRaises(BuildError): self.builder.rootfs() def test_iso_requires_checked_rootfs_artifacts(self): from hydra.build import BuildError self.builder.pod = lambda *args, **kw: self.fail('ISO build started without checked rootfs') self.assertTrue(callable(getattr(self.builder, 'iso', None)), 'iso pipeline is missing') with self.assertRaises(BuildError): self.builder.iso() class NetworkTests(unittest.TestCase): def setUp(self): self.assertIsNotNone(image, 'network preparation is missing') def test_default_route_does_not_conflict(self): image.check_network_conflicts([{'dst': 'default', 'gateway': '192.168.1.1', 'dev': 'eth0'}, {'dst': '192.168.1.0/24', 'dev': 'eth0'}], []) def test_overlapping_lan_vpn_address_or_route_fails_closed(self): for routes, addresses in [([{'dst': '10.203.0.0/16', 'dev': 'tun0'}], []), ([], [{'ifname': 'wlan0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.77', 'prefixlen': 24}]}]), ([{'dst': '10.203.0.8/32', 'dev': 'eth0'}], [])]: with self.subTest(routes=routes, addresses=addresses), self.assertRaises(image.ImageError): image.check_network_conflicts(routes, addresses) def test_existing_owned_bridge_route_allowed(self): image.check_network_conflicts([{'dst': '10.203.0.0/24', 'dev': 'hydrabr0'}], [{'ifname': 'hydrabr0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.1', 'prefixlen': 24}]}]) def test_malformed_data_not_silently_accepted(self): for routes, addresses in [('bad', []), ([{'dst': 'nonsense', 'dev': 'eth0'}], []), ([], [{'ifname': 'eth0', 'addr_info': [{'family': 'inet', 'local': 'bad', 'prefixlen': 24}]}])]: with self.subTest(routes=routes), self.assertRaises(image.ImageError): image.check_network_conflicts(routes, addresses) def test_network_conflict_stops_before_any_mutation(self): calls = [] def run(argv, **kwargs): calls.append(argv) if argv == ['ip', '-json', '-4', 'route', 'show', 'table', 'all']: return subprocess.CompletedProcess(argv, 0, json.dumps([{'dst': '10.203.0.0/24', 'dev': 'eth0'}]), '') if argv == ['ip', '-json', '-4', 'address', 'show']: return subprocess.CompletedProcess(argv, 0, '[]', '') self.fail('Unexpected operation before conflict validation: ' + repr(argv)) with self.assertRaises(image.ImageError): image.configure_network(run) self.assertEqual(len(calls), 2) def test_manual_bridge_has_nat_without_forward_filter(self): changes = [] def run(argv, **kwargs): responses = { ('ip', '-json', '-4', 'route', 'show', 'table', 'all'): '[]', ('ip', '-json', '-4', 'address', 'show'): '[]', ('ip', '-json', '-details', 'link', 'show'): '[]', } if tuple(argv) in responses: return subprocess.CompletedProcess(argv, 0, responses[tuple(argv)], '') if argv[0] == 'nmcli' and argv[1] == '--get-values': return subprocess.CompletedProcess(argv, 10, '', 'not found') if argv == ['nft', 'list', 'table', 'ip', 'hydra_nat']: return subprocess.CompletedProcess(argv, 1, '', 'not found') changes.append((argv, kwargs)) return subprocess.CompletedProcess(argv, 0, '', '') image.configure_network(run) addition = next(argv for argv, kwargs in changes if argv[:3] == ['nmcli', 'connection', 'add']) self.assertEqual(addition[addition.index('ipv4.method') + 1], 'manual') rules = next(kwargs['input'] for argv, kwargs in changes if argv == ['nft', '-f', '-']) self.assertIn('masquerade', rules) self.assertNotIn('hook forward', rules) self.assertNotIn('reject', rules) self.assertNotIn('flush ruleset', rules) def test_foreign_bridge_name_refused(self): calls = [] def run(argv, **kwargs): calls.append(argv) if argv[0] == 'ip': output = '[{"ifname": "hydrabr0", "linkinfo": {"info_kind": "bridge"}}]' if 'link' in argv else '[]' return subprocess.CompletedProcess(argv, 0, output, '') if argv[0] == 'nmcli' and argv[1] == '--get-values': return subprocess.CompletedProcess(argv, 10, '', 'not found') self.fail('Foreign interface triggered a network mutation: ' + repr(argv)) with self.assertRaises(image.ImageError): image.configure_network(run) def test_changed_owned_connection_refused(self): def run(argv, **kwargs): if argv[0] == 'ip': return subprocess.CompletedProcess(argv, 0, '[]', '') if argv[0] == 'nmcli' and argv[1] == '--get-values': return subprocess.CompletedProcess(argv, 0, image.CONNECTION_UUID + '\nbridge\nhydrabr0\nshared\n10.203.0.1/24\n', '') self.fail('Changed connection triggered a network mutation: ' + repr(argv)) with self.assertRaises(image.ImageError): image.configure_network(run) if __name__ == '__main__': unittest.main()