476 lines
26 KiB
Python
476 lines
26 KiB
Python
import importlib.util
|
|
import bz2
|
|
import json
|
|
import os
|
|
import shlex
|
|
import sys
|
|
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_export_filters_saved_environment_proxies_but_retains_package_metadata(self):
|
|
self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
|
|
# Portable stand-in for the external Portage filter API, not the scrubber.
|
|
helper = self.file('usr/lib/portage/python3.14/filter-bash-environment.py',
|
|
"import re\ndef filter_bash_environment(pattern, source, output):\n"
|
|
" for line in source:\n"
|
|
" match = re.match(rb'declare -x ([A-Za-z_][A-Za-z0-9_]*)=', line)\n"
|
|
" if not match or not pattern.fullmatch(match.group(1)):\n"
|
|
" output.write(line)\n")
|
|
path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
|
|
path.parent.mkdir(parents=True)
|
|
package_metadata = b'declare -x LICENSE="GPL-2"\ndeclare -x EAPI="8"\n'
|
|
path.write_bytes(bz2.compress(package_metadata + b'declare -x HTTPS_PROXY="https://user:secret@builder.invalid"\n'
|
|
b'declare -x HTTP_PROXY="http://127.0.0.1:1234"\n'
|
|
b'declare -x NO_PROXY="localhost"\n'))
|
|
result = image.sanitize_saved_environments(self.root)
|
|
self.assertEqual(bz2.decompress(path.read_bytes()), package_metadata)
|
|
self.assertEqual(result['filtered_files'], 1)
|
|
self.assertEqual(result['variable_names'], ['HTTPS_PROXY', 'HTTP_PROXY', 'NO_PROXY'])
|
|
|
|
def test_saved_environment_filter_refuses_missing_portage_helper(self):
|
|
self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
|
|
path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
|
|
path.parent.mkdir(parents=True)
|
|
original = bz2.compress(b'declare -x HTTPS_PROXY="private"\n')
|
|
path.write_bytes(original)
|
|
with self.assertRaises(image.ImageError):
|
|
image.sanitize_saved_environments(self.root)
|
|
self.assertEqual(path.read_bytes(), original)
|
|
|
|
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_installed_helper_uses_relocated_payload_and_opens_empty_recovery(self):
|
|
self.kernel()
|
|
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
|
|
terminal = Path(self.tmp.name) / 'terminal'
|
|
terminal.write_text('#!/bin/sh\nshift\nexec "$@"\n')
|
|
terminal.chmod(0o755)
|
|
manifest = Path(self.tmp.name) / 'session.json'
|
|
manifest.write_text(json.dumps({'config': str(Path(self.tmp.name) / 'config'),
|
|
'state': str(Path(self.tmp.name) / 'state'),
|
|
'runtime': str(Path(self.tmp.name) / 'runtime'),
|
|
'terminal': str(terminal), 'names': []}))
|
|
script = r"""import json, pathlib, subprocess, sys
|
|
sys.path.insert(0, sys.argv[1])
|
|
from hydra.session import _module_argv
|
|
helper = _module_argv('clients', pathlib.Path(sys.argv[2]))
|
|
help_result = subprocess.run(helper[:2] + ['--help'], text=True, capture_output=True)
|
|
print(help_result.stdout)
|
|
print(help_result.stderr, file=sys.stderr)
|
|
if help_result.returncode:
|
|
sys.exit(help_result.returncode)
|
|
result = subprocess.run(helper, input=':quit\n', text=True, capture_output=True, timeout=10)
|
|
print(result.stdout)
|
|
print(result.stderr, file=sys.stderr)
|
|
sys.exit(result.returncode)
|
|
"""
|
|
result = subprocess.run([sys.executable, '-I', '-c', script, str(self.root / 'usr/lib/hydra'), str(manifest)],
|
|
cwd=self.tmp.name, text=True, capture_output=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn('HydraOS local recovery', result.stdout)
|
|
self.assertIn('add', result.stdout)
|
|
|
|
def test_installed_updates_exclude_unadapted_qemu_and_viewer_binaries(self):
|
|
self.kernel()
|
|
source = Path(__file__).resolve().parents[1]
|
|
self.file('etc/portage/make.conf', (source / 'image/portage/make.conf').read_text())
|
|
image.configure_rootfs(self.root, source)
|
|
settings = subprocess.run(['bash', '-euc', 'source "$1"; printf "%s" "$EMERGE_DEFAULT_OPTS"',
|
|
'--', str(self.root / 'etc/portage/make.conf')], text=True, capture_output=True, check=True)
|
|
defaults = shlex.split(settings.stdout)
|
|
exclusions = [item.split('=', 1)[1].split() for item in defaults if item.startswith('--usepkg-exclude=')]
|
|
self.assertEqual(exclusions, [['app-emulation/qemu', 'app-emulation/virt-viewer']])
|
|
self.assertIn('--getbinpkg', defaults)
|
|
|
|
def test_export_enables_pulse_socket_without_starting_user_services(self):
|
|
source = Path(__file__).resolve().parents[1]
|
|
self.assertTrue(callable(getattr(image, 'enable_user_audio', None)), 'user audio enable is missing')
|
|
units = self.root / 'usr/lib/systemd/user'
|
|
units.mkdir(parents=True)
|
|
for name, section in [('pipewire.socket', 'Socket'), ('pipewire-pulse.socket', 'Socket'),
|
|
('wireplumber.service', 'Service')]:
|
|
target = 'sockets.target' if section == 'Socket' else 'default.target'
|
|
(units / name).write_text(f'[{section}]\n[Install]\nWantedBy={target}\n')
|
|
image.enable_user_audio(self.root)
|
|
for target, name in [('sockets.target', 'pipewire.socket'), ('sockets.target', 'pipewire-pulse.socket'),
|
|
('default.target', 'wireplumber.service')]:
|
|
link = self.root / 'etc/systemd/user' / (target + '.wants') / name
|
|
self.assertTrue(link.is_symlink(), name)
|
|
self.assertEqual(os.readlink(link), '/usr/lib/systemd/user/' + name)
|
|
|
|
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 inventory(self, records=None):
|
|
records = records if records is not None else [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]
|
|
self.builder.work.mkdir(exist_ok=True)
|
|
artifacts = self.builder.work / 'artifacts'
|
|
artifacts.mkdir(exist_ok=True)
|
|
path = artifacts / 'packages.json'
|
|
path.write_text(json.dumps(records))
|
|
original = self.builder.complete
|
|
self.builder.complete = lambda step: True if step == 'prepare' else original(step)
|
|
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
|
|
return path
|
|
|
|
def test_changed_inventory_invalidates_package_checkpoint(self):
|
|
path = self.inventory()
|
|
self.assertTrue(self.builder.complete('packages'))
|
|
path.write_text('[{"package":"app-emulation/qemu-8.0", "use": [], "build_time": null}]')
|
|
self.assertFalse(self.builder.complete('packages'))
|
|
|
|
def test_malformed_inventory_with_matching_digest_is_refused(self):
|
|
for data in ['{', '{}', '[]', '[{"package":"wrong"}]',
|
|
'[{"package":"app-emulation/qemu-9.2.4", "use":"spice", "build_time":"123"}]',
|
|
'[{"package":"app-emulation/qemu-9.2.4", "use":[], "build_time":42}]']:
|
|
with self.subTest(data=data):
|
|
path = self.inventory()
|
|
path.write_text(data)
|
|
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
|
|
self.assertFalse(self.builder.complete('packages'))
|
|
|
|
def test_changed_inventory_invalidates_completed_rootfs_and_iso(self):
|
|
path = self.inventory()
|
|
artifacts = path.parent
|
|
self.builder.active_package_job = lambda: False
|
|
self.builder.assembly = lambda snapshot=None: 'isolated-assembler'
|
|
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
|
|
def pod(*args, **kwargs):
|
|
if args[0] == 'export':
|
|
snapshot = artifacts / 'builder-rootfs.tar'
|
|
self.assertTrue(snapshot.is_file(), 'raw snapshot must be created privately before export')
|
|
self.assertEqual(snapshot.stat().st_mode & 0o777, 0o600)
|
|
snapshot.write_bytes(b'exported fixture')
|
|
elif args[0] == 'exec' and 'configure-rootfs.sh' in args[-1]:
|
|
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
|
|
(artifacts / name).write_bytes(b'artifact')
|
|
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
|
|
'packages': [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]}))
|
|
elif args[0] == 'exec' and 'grub-mkrescue' in args[-1]:
|
|
(artifacts / 'hydra-os.iso').write_bytes(b'ISO fixture')
|
|
(artifacts / 'iso-inspection.txt').write_text('HYDRA_OS\nEl Torito boot img : 1 UEFI\n')
|
|
else:
|
|
self.fail('Unexpected Podman operation: ' + repr(args))
|
|
self.builder.pod = pod
|
|
self.builder.rootfs()
|
|
self.builder.iso()
|
|
self.assertTrue(self.builder.complete('rootfs'))
|
|
self.assertTrue(self.builder.complete('iso'))
|
|
path.write_text('[{"package":"app-emulation/qemu-9.2.5", "use":[], "build_time":null}]')
|
|
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
|
|
self.assertTrue(self.builder.complete('packages'))
|
|
self.assertFalse(self.builder.complete('rootfs'))
|
|
self.assertFalse(self.builder.complete('iso'))
|
|
|
|
def test_export_inventory_must_match_validated_package_checkpoint(self):
|
|
from hydra.build import BuildError
|
|
path = self.inventory()
|
|
self.builder.complete = lambda step: step == 'packages'
|
|
self.builder.active_package_job = lambda: False
|
|
self.builder.assembly = lambda snapshot: 'isolated-assembler'
|
|
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
|
|
def pod(*args, **kwargs):
|
|
artifacts = self.builder.work / 'artifacts'
|
|
if args[0] == 'export':
|
|
(artifacts / 'builder-rootfs.tar').write_bytes(b'exported fixture')
|
|
elif args[0] == 'exec':
|
|
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
|
|
(artifacts / name).write_bytes(b'artifact')
|
|
# Actual exported package differs from the checkpoint inventory.
|
|
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
|
|
'packages': [{'package': 'app-emulation/qemu-9.2.5', 'use': ['spice'], 'build_time': '456'}]}))
|
|
else:
|
|
self.fail('Unexpected Podman operation: ' + repr(args))
|
|
self.builder.pod = pod
|
|
with self.assertRaisesRegex(BuildError, 'inventory'):
|
|
self.builder.rootfs()
|
|
self.assertFalse(self.builder.manifest()['rootfs']['complete'])
|
|
|
|
def test_corrupt_local_package_archive_invalidates_fetch_checkpoint(self):
|
|
import hashlib
|
|
source = self.project / 'image'
|
|
source.mkdir()
|
|
(source / 'sources.json').write_text(json.dumps({
|
|
'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'stage').hexdigest()},
|
|
'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()},
|
|
'local_packages': [{'url': 'https://example.test/brightnessctl.tar.gz',
|
|
'filename': 'brightnessctl.tar.gz', 'sha256': hashlib.sha256(b'original-source').hexdigest()}]}))
|
|
with self.builder.locked():
|
|
(self.builder.cache / 'stage.tar').write_bytes(b'stage')
|
|
(self.builder.cache / 'repo.tar').write_bytes(b'repo')
|
|
archive = self.builder.cache / 'brightnessctl.tar.gz'
|
|
archive.write_bytes(b'original-source')
|
|
self.builder.record('fetch')
|
|
self.assertTrue(self.builder.complete('fetch'))
|
|
archive.write_bytes(b'corrupted-source')
|
|
self.assertFalse(self.builder.complete('fetch'))
|
|
|
|
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()
|