From d7d01a4520f6aee1c517d0982ec15b122e212a46 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 04:04:30 +0300 Subject: [PATCH] feat: add independent VM manager --- bin/hydra | 9 ++ hydra/__init__.py | 1 + hydra/cli.py | 60 +++++++++ hydra/profiles.py | 72 ++++++++++ hydra/qmp.py | 91 +++++++++++++ hydra/runtime.py | 295 ++++++++++++++++++++++++++++++++++++++++ tests/test_profiles.py | 54 ++++++++ tests/test_qmp.py | 141 +++++++++++++++++++ tests/test_runtime.py | 297 +++++++++++++++++++++++++++++++++++++++++ 9 files changed, 1020 insertions(+) create mode 100755 bin/hydra create mode 100644 hydra/__init__.py create mode 100644 hydra/cli.py create mode 100644 hydra/profiles.py create mode 100644 hydra/qmp.py create mode 100644 hydra/runtime.py create mode 100644 tests/test_profiles.py create mode 100644 tests/test_qmp.py create mode 100644 tests/test_runtime.py diff --git a/bin/hydra b/bin/hydra new file mode 100755 index 0000000..6265501 --- /dev/null +++ b/bin/hydra @@ -0,0 +1,9 @@ +#!/usr/bin/env python3 +"""Run HydraOS from this checkout without installing a host package.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from hydra.cli import main + +raise SystemExit(main()) diff --git a/hydra/__init__.py b/hydra/__init__.py new file mode 100644 index 0000000..6cbeed2 --- /dev/null +++ b/hydra/__init__.py @@ -0,0 +1 @@ +"""HydraOS independent VM management.""" diff --git a/hydra/cli.py b/hydra/cli.py new file mode 100644 index 0000000..f998065 --- /dev/null +++ b/hydra/cli.py @@ -0,0 +1,60 @@ +"""Portable source-checkout command line interface.""" + +import argparse +import json +import os +import sys +from pathlib import Path + +from .profiles import Profile +from .runtime import Runtime, RuntimeError + + +def parser() -> argparse.ArgumentParser: + cli = argparse.ArgumentParser(prog='hydra', description='Independent HydraOS VM manager') + cli.add_argument('--config', type=Path, default=Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'hydra') + cli.add_argument('--state', type=Path, default=Path(os.environ.get('XDG_STATE_HOME', Path.home() / '.local/state')) / 'hydra') + cli.add_argument('--runtime', type=Path, default=Path(os.environ.get('XDG_RUNTIME_DIR', f'/tmp/hydra-{os.getuid()}')) / 'hydra') + commands = cli.add_subparsers(dest='command', required=True) + add = commands.add_parser('add', help='Create a VM profile without starting it') + add.add_argument('name') + add.add_argument('--iso', type=Path, required=True) + add.add_argument('--memory', type=int, default=4096) + add.add_argument('--cpus', type=int, default=2) + add.add_argument('--disk-size', default='40G') + add.add_argument('--graphics', choices=('auto', 'software'), default='auto') + add.add_argument('--network', choices=('user', 'bridge'), default='user') + add.add_argument('--bridge', default='hydrabr0') + commands.add_parser('list', help='List configured VMs') + status = commands.add_parser('status', help='Query VM state through private QMP') + status.add_argument('name', nargs='?') + for command in ('start', 'shutdown', 'stop'): + cmd = commands.add_parser(command) + cmd.add_argument('name') + if command == 'stop': + cmd.add_argument('--force', action='store_true', required=True) + return cli + + +def main(argv: list[str] | None = None) -> int: + args = parser().parse_args(argv) + runtime = Runtime(args.config, args.state, args.runtime) + try: + if args.command == 'add': + profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size, + args.graphics, args.network, args.bridge) + runtime.add(profile) + result = {'name': profile.name, 'status': 'added'} + elif args.command == 'list': + result = [{'name': profile.name, 'iso': str(profile.iso)} for profile in runtime.profiles()] + elif args.command == 'status': + result = runtime.status(args.name) if args.name else [runtime.status(p.name) for p in runtime.profiles()] + elif args.command == 'stop': + result = runtime.stop(args.name, force=args.force) + else: + result = getattr(runtime, args.command)(args.name) + print(json.dumps(result, indent=2)) + return 0 + except (RuntimeError, OSError, ValueError) as exc: + print(f'hydra: {exc}', file=sys.stderr) + return 1 diff --git a/hydra/profiles.py b/hydra/profiles.py new file mode 100644 index 0000000..3b3ec12 --- /dev/null +++ b/hydra/profiles.py @@ -0,0 +1,72 @@ +"""Validated, independently editable VM profiles.""" + +import json +import re +import tomllib +from dataclasses import dataclass +from pathlib import Path + + +def validate_name(name: str) -> str: + if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', name): + raise ValueError('VM name must match [a-z][a-z0-9-]{0,31}') + return name + + +@dataclass(frozen=True) +class Profile: + name: str + iso: Path + memory: int = 4096 + cpus: int = 2 + disk_size: str = '40G' + graphics: str = 'auto' + network: str = 'user' + bridge: str = 'hydrabr0' + + def __post_init__(self): + validate_name(self.name) + iso = Path(self.iso).expanduser().resolve() + if not iso.is_file(): + raise ValueError(f'ISO does not exist or is not a file: {iso}') + object.__setattr__(self, 'iso', iso) + for key, upper in (('memory', 262144), ('cpus', 256)): + value = getattr(self, key) + if type(value) is not int or not 1 <= value <= upper: + raise ValueError(f'{key} must be an integer between 1 and {upper}') + if not isinstance(self.disk_size, str) or not re.fullmatch(r'[1-9][0-9]{0,5}[MGT]', self.disk_size): + raise ValueError('disk_size must be a positive size with M, G or T suffix') + if self.graphics not in ('auto', 'software'): + raise ValueError('graphics must be auto or software') + if self.network not in ('user', 'bridge'): + raise ValueError('network must be user or bridge') + if not isinstance(self.bridge, str) or not re.fullmatch(r'[a-zA-Z0-9_-]{1,15}', self.bridge): + raise ValueError('bridge must be a valid Linux interface name') + + +def load_profile(path: Path) -> Profile: + path = Path(path) + with path.open('rb') as stream: + data = tomllib.load(stream) + unknown = data.keys() - Profile.__dataclass_fields__.keys() + if unknown: + raise ValueError(f'Unknown profile fields: {", ".join(sorted(unknown))}') + name = data.pop('name', path.stem) + if name != path.stem: + raise ValueError('Profile name must match its filename') + try: + return Profile(name=name, **data) + except TypeError as exc: + raise ValueError(f'Invalid profile: {exc}') from exc + + +def save_profile(path: Path, profile: Profile) -> None: + path = Path(path) + if path.stem != profile.name: + raise ValueError('Profile name must match its filename') + lines = [] + for field in Profile.__dataclass_fields__: + value = getattr(profile, field) + lines.append(f'{field} = {value if type(value) is int else json.dumps(str(value), ensure_ascii=False)}\n') + with path.open('x', encoding='utf-8') as stream: + stream.write(''.join(lines)) diff --git a/hydra/qmp.py b/hydra/qmp.py new file mode 100644 index 0000000..eb0f687 --- /dev/null +++ b/hydra/qmp.py @@ -0,0 +1,91 @@ +"""Bounded JSON QMP over private Unix sockets, with reply ID matching.""" + +import json +import socket +import time +from pathlib import Path + + +class QMPError(RuntimeError): + pass + + +class QMPClient: + def __init__(self, path: Path, timeout: float = 2.0): + if timeout <= 0: + raise ValueError('QMP timeout must be positive') + self.timeout = timeout + self.socket = socket.socket(socket.AF_UNIX) + self.buffer = bytearray() + self.counter = 0 + deadline = time.monotonic() + timeout + try: + self.socket.settimeout(timeout) + self.socket.connect(str(path)) + greeting = self._receive(deadline) + if not isinstance(greeting.get('QMP'), dict): + raise QMPError('Invalid QMP greeting') + self._command('qmp_capabilities', None, deadline) + except (OSError, QMPError) as exc: + self.close() + if isinstance(exc, QMPError): + raise + raise QMPError(f'QMP connection failed: {exc}') from exc + + def _receive(self, deadline: float) -> dict: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise QMPError('QMP deadline exceeded') + if b'\n' in self.buffer: + line, _, rest = self.buffer.partition(b'\n') + self.buffer = bytearray(rest) + try: + message = json.loads(line) + except (ValueError, UnicodeDecodeError) as exc: + raise QMPError('Malformed QMP JSON') from exc + if not isinstance(message, dict): + raise QMPError('QMP response must be an object') + return message + self.socket.settimeout(remaining) + try: + chunk = self.socket.recv(65536) + except OSError as exc: + raise QMPError(f'QMP receive failed: {exc}') from exc + if not chunk: + raise QMPError('QMP connection closed') + self.buffer.extend(chunk) + if len(self.buffer) > 1048576: + raise QMPError('QMP message exceeds 1 MiB') + + def _command(self, name: str, arguments: dict | None, deadline: float) -> dict: + self.counter += 1 + request = {'execute': name, 'id': self.counter} + if arguments is not None: + request['arguments'] = arguments + try: + self.socket.settimeout(max(0.001, deadline - time.monotonic())) + self.socket.sendall(json.dumps(request).encode() + b'\r\n') + except OSError as exc: + raise QMPError(f'QMP send failed: {exc}') from exc + while True: + response = self._receive(deadline) + if response.get('id') != self.counter: + continue + if 'error' in response: + raise QMPError(f'QMP {name}: {response["error"]}') + if 'return' not in response: + raise QMPError('QMP reply has no return value') + return response['return'] + + def command(self, name: str, arguments: dict | None = None) -> dict: + return self._command(name, arguments, time.monotonic() + self.timeout) + + def close(self): + self.socket.close() + + def __enter__(self): + return self + + def __exit__(self, *args): + self.close() diff --git a/hydra/runtime.py b/hydra/runtime.py new file mode 100644 index 0000000..57a6962 --- /dev/null +++ b/hydra/runtime.py @@ -0,0 +1,295 @@ +"""Independent QEMU lifecycle; all destructive control requires QMP identity.""" + +import contextlib +import errno +import fcntl +import json +import os +import socket +import stat +import subprocess +import tempfile +import time +import uuid +from dataclasses import dataclass +from pathlib import Path + +from .profiles import Profile, load_profile, save_profile, validate_name +from .qmp import QMPClient, QMPError + + +class RuntimeError(Exception): + """A VM lifecycle operation could not be safely completed.""" + + +@dataclass(frozen=True) +class VMPaths: + name: str + disk: Path + log: Path + lock: Path + qmp: Path + spice: Path + pid: Path + metadata: Path + + @property + def uuid(self) -> str: + return str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.disk))) + + +def _qemu_value(value: Path | str) -> str: + # QEMU's key/value parser uses doubled commas for literal commas. + return str(value).replace(',', ',,') + + +def _intel_render_node() -> Path | None: + for node in sorted(Path('/dev/dri').glob('renderD*')): + vendor = Path('/sys/class/drm') / node.name / 'device/vendor' + try: + if vendor.read_text().strip() == '0x8086' and os.access(node, os.R_OK | os.W_OK): + return node + except OSError: + continue + return None + + +def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]: + accel = 'kvm' if os.access('/dev/kvm', os.R_OK | os.W_OK) else 'tcg' + argv = [os.environ.get('HYDRA_QEMU', 'qemu-system-x86_64'), '-name', profile.name, + '-uuid', paths.uuid, '-machine', f'q35,accel={accel}', + '-m', str(profile.memory), '-smp', str(profile.cpus), + '-daemonize', '-pidfile', str(paths.pid), '-nodefaults', + '-chardev', f'socket,id=qmp,path={_qemu_value(paths.qmp)},server=on,wait=off', + '-mon', 'chardev=qmp,mode=control'] + if accel == 'kvm': + argv += ['-cpu', 'host'] + for options in ( + {'driver': 'file', 'filename': str(paths.disk), 'node-name': 'disk-file'}, + {'driver': 'qcow2', 'file': 'disk-file', 'node-name': 'disk'}, + {'driver': 'file', 'filename': str(profile.iso), 'node-name': 'iso-file', 'read-only': True}, + {'driver': 'raw', 'file': 'iso-file', 'node-name': 'iso', 'read-only': True}, + ): + argv += ['-blockdev', json.dumps(options)] + argv += ['-device', 'virtio-blk-pci,drive=disk', '-device', 'ide-cd,drive=iso', + '-boot', 'order=dc', '-device', 'virtio-net-pci,netdev=net', + '-netdev', 'user,id=net' if profile.network == 'user' + else f'bridge,id=net,br={profile.bridge}', + '-device', 'qemu-xhci,id=usb', '-device', 'usb-tablet', '-device', 'usb-kbd'] + argv += ['-audiodev', 'spice,id=audio', '-device', 'ich9-intel-hda', + '-device', 'hda-duplex,audiodev=audio'] + for channel in range(3): + argv += ['-chardev', f'spicevmc,id=usbredir{channel},name=usbredir', + '-device', f'usb-redir,chardev=usbredir{channel}'] + spice = (f'unix=on,addr={_qemu_value(paths.spice)},disable-ticketing=on,' + 'disable-copy-paste=on,disable-agent-file-xfer=on') + render = _intel_render_node() if profile.graphics == 'auto' else None + if render: + spice += f',gl=on,rendernode={_qemu_value(render)}' + argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl'] + else: + argv += ['-display', 'none', '-device', 'virtio-vga'] + return argv + ['-spice', spice] + + +def _private_dir(path: Path) -> None: + if path.is_symlink(): + raise RuntimeError(f'Private directory must not be a symlink: {path}') + path.mkdir(parents=True, mode=0o700, exist_ok=True) + info = path.stat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): + raise RuntimeError(f'Private directory is not owned by this user: {path}') + path.chmod(0o700) + + +class Runtime: + def __init__(self, config: Path, state: Path, runtime: Path, boot_timeout: float = 10): + # Keep final components lexical so symlink checks still see them. + self.config = Path(config).expanduser().absolute() + self.state = Path(state).expanduser().absolute() + self.runtime = Path(runtime).expanduser().absolute() + self.boot_timeout = boot_timeout + + def paths(self, name: str) -> VMPaths: + validate_name(name) + state = self.state / name + run = self.runtime / name + paths = VMPaths(name, state / 'disk.qcow2', state / 'qemu.log', state / 'lock', + run / 'qmp.sock', run / 'spice.sock', run / 'qemu.pid', run / 'identity.json') + if any(len(os.fsencode(path)) >= 108 for path in (paths.qmp, paths.spice)): + raise RuntimeError('Unix socket path is too long; choose a shorter --runtime path') + return paths + + def prepare(self, paths: VMPaths) -> None: + for directory in (self.config, self.state, self.runtime, paths.disk.parent, paths.qmp.parent): + _private_dir(directory) + + def add(self, profile: Profile) -> None: + _private_dir(self.config) + save_profile(self.config / f'{profile.name}.toml', profile) + + def profiles(self) -> list[Profile]: + if not self.config.exists(): + return [] + return [load_profile(path) for path in sorted(self.config.glob('*.toml'))] + + def profile(self, name: str) -> Profile: + validate_name(name) + return load_profile(self.config / f'{name}.toml') + + @contextlib.contextmanager + def _locked(self, paths: VMPaths): + self.prepare(paths) + fd = os.open(paths.lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + yield + finally: + os.close(fd) + + def _owned_process_alive(self, paths: VMPaths) -> bool: + # PID is only a diagnostic hint. Never send a signal to it. + try: + pid = int(paths.pid.read_text()) + args = Path(f'/proc/{pid}/cmdline').read_bytes().split(b'\0') + wanted_uuid = paths.uuid.encode() + wanted_name = paths.name.encode() + return (b'-uuid' in args and args[args.index(b'-uuid') + 1] == wanted_uuid + and b'-name' in args and args[args.index(b'-name') + 1] == wanted_name) + except (OSError, ValueError, IndexError): + return False + + def _connect(self, paths: VMPaths) -> QMPClient: + try: + record = json.loads(paths.metadata.read_text()) + if not isinstance(record, dict) or record.get('name') != paths.name or record.get('uuid') != paths.uuid: + raise RuntimeError('VM identity record does not match selected VM') + client = QMPClient(paths.qmp, timeout=min(2, self.boot_timeout)) + try: + name_result = client.command('query-name') + uuid_result = client.command('query-uuid') + if not isinstance(name_result, dict) or not isinstance(uuid_result, dict): + raise RuntimeError('Malformed QMP identity reply') + name = name_result.get('name') + actual_uuid = uuid_result.get('UUID') + if name != paths.name or actual_uuid != record['uuid']: + raise RuntimeError('QMP VM identity mismatch; refusing control') + except Exception: + client.close() + raise + return client + except (OSError, ValueError, QMPError) as exc: + raise RuntimeError(f'Cannot verify VM identity: {exc}') from exc + + def _listening(self, path: Path) -> bool: + if not path.exists(): + return False + if not stat.S_ISSOCK(path.lstat().st_mode): + raise RuntimeError(f'Refusing non-socket runtime path: {path}') + with socket.socket(socket.AF_UNIX) as probe: + probe.settimeout(min(1, self.boot_timeout)) + try: + probe.connect(str(path)) + return True + except OSError as exc: + if exc.errno in (errno.ENOENT, errno.ECONNREFUSED): + return False + raise RuntimeError(f'Cannot establish socket ownership: {exc}') from exc + + def status(self, name: str) -> dict: + paths = self.paths(name) + try: + if not self._listening(paths.qmp): + if self._owned_process_alive(paths): + return {'name': name, 'status': 'unknown', 'error': 'VM process exists without working QMP'} + return {'name': name, 'status': 'stopped'} + with self._connect(paths) as client: + status = client.command('query-status') + if not isinstance(status, dict) or not isinstance(status.get('status'), str): + raise RuntimeError('Malformed QMP status reply') + return {'name': name, **status} + except (RuntimeError, QMPError) as exc: + return {'name': name, 'status': 'unknown', 'error': str(exc)} + + def _create_disk(self, profile: Profile, paths: VMPaths) -> None: + if paths.disk.exists() or paths.disk.is_symlink(): + if paths.disk.is_symlink() or not paths.disk.is_file(): + raise RuntimeError(f'Refusing non-regular disk: {paths.disk}') + return + fd, temporary = tempfile.mkstemp(prefix='.disk-', suffix='.qcow2', dir=paths.disk.parent) + os.close(fd) + temp = Path(temporary) + temp.unlink() # qemu-img creates a new file; final publication never replaces a disk. + try: + subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'create', '-f', 'qcow2', + str(temp), profile.disk_size], check=True, capture_output=True, + text=True, timeout=30) + temp.chmod(0o600) + os.link(temp, paths.disk) + except (OSError, subprocess.SubprocessError) as exc: + raise RuntimeError(f'Disk creation failed: {exc}') from exc + finally: + temp.unlink(missing_ok=True) + + def start(self, name: str) -> dict: + profile = self.profile(name) + paths = self.paths(name) + with self._locked(paths): + current = self.status(name) + if current['status'] != 'stopped': + raise RuntimeError(f'Cannot start {name}: {current["status"]}') + for path in (paths.qmp, paths.spice): + if self._listening(path): + raise RuntimeError(f'Runtime socket is already in use: {path}') + path.unlink(missing_ok=True) + self._create_disk(profile, paths) + fd, temporary = tempfile.mkstemp(prefix='.identity-', dir=paths.metadata.parent) + try: + with os.fdopen(fd, 'w') as record: + record.write(json.dumps({'name': name, 'uuid': paths.uuid}) + '\n') + os.replace(temporary, paths.metadata) + finally: + Path(temporary).unlink(missing_ok=True) + paths.pid.unlink(missing_ok=True) + try: + log_fd = os.open(paths.log, os.O_CREAT | os.O_APPEND | os.O_WRONLY | os.O_NOFOLLOW, 0o600) + with os.fdopen(log_fd, 'ab') as log: + subprocess.run(build_qemu_argv(profile, paths), stdin=subprocess.DEVNULL, + stdout=log, stderr=log, start_new_session=True, + check=True, timeout=self.boot_timeout, + umask=0o077) + deadline = time.monotonic() + self.boot_timeout + while time.monotonic() < deadline: + current = self.status(name) + if current['status'] not in ('stopped', 'unknown'): + return current + time.sleep(0.05) + raise RuntimeError(f'QEMU did not become ready; inspect {paths.log}') + except (OSError, subprocess.SubprocessError) as exc: + raise RuntimeError(f'QEMU launch failed; inspect {paths.log}: {exc}') from exc + + def shutdown(self, name: str) -> dict: + paths = self.paths(name) + try: + with self._locked(paths), self._connect(paths) as client: + client.command('system_powerdown') + except QMPError as exc: + raise RuntimeError(f'Guest shutdown request failed: {exc}') from exc + return {'name': name, 'status': 'shutdown-requested'} + + def stop(self, name: str, force: bool = False) -> dict: + if not force: + raise RuntimeError('Abrupt stop requires --force; use shutdown for guest ACPI shutdown') + paths = self.paths(name) + with self._locked(paths): + try: + with self._connect(paths) as client: + client.command('quit') + except QMPError as exc: + raise RuntimeError(f'QEMU quit request failed: {exc}') from exc + deadline = time.monotonic() + self.boot_timeout + while time.monotonic() < deadline: + if self.status(name)['status'] == 'stopped': + return {'name': name, 'status': 'stopped'} + time.sleep(0.05) + raise RuntimeError('QEMU accepted quit but has not stopped within the deadline') diff --git a/tests/test_profiles.py b/tests/test_profiles.py new file mode 100644 index 0000000..ce21129 --- /dev/null +++ b/tests/test_profiles.py @@ -0,0 +1,54 @@ +import tempfile +import unittest +from pathlib import Path + +from hydra.profiles import Profile, load_profile, save_profile + + +class ProfilesTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix='hydra-test-') + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.iso = self.root / 'installer with spaces,comma.iso' + self.iso.touch() + + def test_roundtrip_paths_and_defaults(self): + profile = Profile('guest', self.iso) + save_profile(self.root / 'guest.toml', profile) + loaded = load_profile(self.root / 'guest.toml') + self.assertEqual(loaded.iso, self.iso) + self.assertEqual((loaded.memory, loaded.cpus, loaded.disk_size), (4096, 2, '40G')) + self.assertEqual((loaded.graphics, loaded.network, loaded.bridge), ('auto', 'user', 'hydrabr0')) + + def test_invalid_names_resources_and_modes(self): + for name in ('', '../guest', 'Guest', '-guest', 'a' * 33, 'a\n'): + with self.subTest(name=name), self.assertRaises(ValueError): + Profile(name, self.iso) + for field, values in {'memory': (0, -1, True, '4096'), 'cpus': (0, -1, True, 257), + 'disk_size': ('0G', '-1G', '40G,evil', '40'), + 'graphics': ('gl',), 'network': ('public',), + 'bridge': ('../bad', 'a,evil')}.items(): + for value in values: + with self.subTest(field=field, value=value), self.assertRaises(ValueError): + Profile('guest', self.iso, **{field: value}) + + def test_missing_iso_and_duplicate_profile_preserve_existing(self): + with self.assertRaises(ValueError): + Profile('guest', self.root / 'missing.iso') + dest = self.root / 'guest.toml' + save_profile(dest, Profile('guest', self.iso)) + original = dest.read_bytes() + with self.assertRaises(FileExistsError): + save_profile(dest, Profile('guest', self.iso, memory=8192)) + self.assertEqual(dest.read_bytes(), original) + + def test_profile_name_must_match_filename(self): + path = self.root / 'guest.toml' + path.write_text(f'name = "other"\niso = "{self.iso}"\n') + with self.assertRaises(ValueError): + load_profile(path) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_qmp.py b/tests/test_qmp.py new file mode 100644 index 0000000..4488986 --- /dev/null +++ b/tests/test_qmp.py @@ -0,0 +1,141 @@ +import json +import socket +import tempfile +import threading +import time +import unittest +from pathlib import Path + +from hydra.qmp import QMPClient, QMPError + + +class QMPServer: + def __init__(self, path, handler): + self.socket = socket.socket(socket.AF_UNIX) + self.socket.bind(str(path)) + self.socket.listen(4) + self.socket.settimeout(0.1) + self.handler = handler + self.stop = threading.Event() + self.errors = [] + self.thread = threading.Thread(target=self.run, daemon=True) + self.thread.start() + + def run(self): + while not self.stop.is_set(): + try: + conn, _ = self.socket.accept() + except socket.timeout: + continue + except OSError: + break + try: + with conn: + conn.settimeout(1) + self.handler(conn) + except (BrokenPipeError, ConnectionResetError, socket.timeout): + pass + except Exception as exc: + self.errors.append(exc) + + def close(self): + self.stop.set() + self.socket.close() + self.thread.join(2) + + +def send(conn, message): + conn.sendall(json.dumps(message).encode() + b'\r\n') + + +def greeting(conn): + send(conn, {'QMP': {'version': {'qemu': {'major': 10, 'minor': 0, 'micro': 0}, + 'package': ''}, 'capabilities': []}}) + + +class QMPTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix='hq-') + self.addCleanup(self.tmp.cleanup) + self.path = Path(self.tmp.name) / 'qmp' + + def server(self, handler): + srv = QMPServer(self.path, handler) + self.addCleanup(srv.close) + return srv + + def test_handshake_ids_events_and_return(self): + commands = [] + + def handler(conn): + greeting(conn) + stream = conn.makefile('rb') + for index in range(2): + request = json.loads(stream.readline()) + commands.append(request) + send(conn, {'event': 'STOP', 'data': {}}) + send(conn, {'return': {'ignored': True}, 'id': 'unrelated'}) + send(conn, {'return': {} if index == 0 else {'status': 'running'}, 'id': request['id']}) + self.server(handler) + with QMPClient(self.path, timeout=0.5) as client: + self.assertEqual(client.command('query-status'), {'status': 'running'}) + self.assertEqual([cmd['execute'] for cmd in commands], ['qmp_capabilities', 'query-status']) + self.assertNotEqual(commands[0]['id'], commands[1]['id']) + + def test_command_error(self): + def handler(conn): + greeting(conn) + stream = conn.makefile('rb') + for index in range(2): + request = json.loads(stream.readline()) + if index == 0: + send(conn, {'return': {}, 'id': request['id']}) + else: + send(conn, {'error': {'class': 'CommandNotFound', 'desc': 'unsupported'}, 'id': request['id']}) + self.server(handler) + with QMPClient(self.path, timeout=0.5) as client, self.assertRaises(QMPError): + client.command('missing') + + def test_greeting_timeout_is_bounded(self): + self.server(lambda conn: time.sleep(0.3)) + started = time.monotonic() + with self.assertRaises(QMPError): + QMPClient(self.path, timeout=0.08) + self.assertLess(time.monotonic() - started, 0.25) + + def test_event_stream_cannot_extend_command_deadline(self): + def handler(conn): + greeting(conn) + request = json.loads(conn.makefile('rb').readline()) + while True: + send(conn, {'event': 'STOP', 'data': {}}) + time.sleep(0.01) + self.server(handler) + started = time.monotonic() + with self.assertRaises(QMPError): + QMPClient(self.path, timeout=0.08) + self.assertLess(time.monotonic() - started, 0.25) + + def test_malformed_greeting_rejected(self): + self.server(lambda conn: send(conn, {'return': {}})) + with self.assertRaises(QMPError): + QMPClient(self.path) + + def test_command_timeout_after_successful_handshake(self): + def handler(conn): + greeting(conn) + with conn.makefile('rb') as stream: + request = json.loads(stream.readline()) + send(conn, {'return': {}, 'id': request['id']}) + stream.readline() + time.sleep(0.3) + self.server(handler) + with QMPClient(self.path, timeout=0.08) as client: + started = time.monotonic() + with self.assertRaises(QMPError): + client.command('query-status') + self.assertLess(time.monotonic() - started, 0.25) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..40a27d3 --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,297 @@ +import json +import os +import socket +import subprocess +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import patch + +from hydra.profiles import Profile, save_profile +from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv +from test_qmp import QMPServer, greeting, send + + +# External protocol fixtures isolate lifecycle tests from user VMs and real disks. +FAKE_QEMU = r'''#!/usr/bin/env python3 +import json, os, socket, sys +args = sys.argv[1:] +def val(flag): return args[args.index(flag)+1] +if os.environ.get('HYDRA_TEST_FAIL'): + print('intentional boot failure', file=sys.stderr) + sys.exit(1) +path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',') +pid = os.fork() +if pid: + sys.exit(0) +os.setsid() +with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid())) +s = socket.socket(socket.AF_UNIX) +s.bind(path) +s.listen(8) +alive = True +while alive: + c, _ = s.accept() + try: + with c: + c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n') + with c.makefile('rb') as stream: + for line in stream: + cmd = json.loads(line) + name = cmd['execute'] + result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')}, + 'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {}) + c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n') + if name == 'quit': + alive = False + break + except (BrokenPipeError, ConnectionResetError): + pass +s.close() +os.unlink(path) +''' + +FAKE_IMG = r'''#!/usr/bin/env python3 +import sys +with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture') +''' + + +class RuntimeTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix='hr-') + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma', + self.root / 'run,comma', boot_timeout=1) + self.iso = self.root / 'installer space,comma.iso' + self.iso.touch() + self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M') + self.runtime.add(self.profile) + self.qemu = self.root / 'fake qemu' + self.qemu.write_text(FAKE_QEMU) + self.qemu.chmod(0o700) + self.img = self.root / 'fake img' + self.img.write_text(FAKE_IMG) + self.img.chmod(0o700) + env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)}) + env.start() + self.addCleanup(env.stop) + self.addCleanup(self.cleanup_guest) + + def cleanup_guest(self): + try: + if self.runtime.status('guest')['status'] == 'running': + self.runtime.stop('guest', force=True) + except VMError: + pass + + def test_argv_is_private_and_comma_safe(self): + paths = self.runtime.paths('guest') + argv = build_qemu_argv(self.profile, paths) + self.assertEqual(argv[0], str(self.qemu)) + blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev'] + self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs]) + self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs]) + spice = argv[argv.index('-spice')+1] + self.assertIn('unix=on', spice) + self.assertIn('disable-copy-paste=on', spice) + self.assertIn('disable-agent-file-xfer=on', spice) + self.assertIn('run,,comma', spice) + self.assertIn('virtio-blk-pci,drive=disk', argv) + self.assertIn('virtio-net-pci,netdev=net', argv) + self.assertNotIn('-vnc', argv) + self.assertNotIn('-virtfs', argv) + + def test_argv_provides_audio_and_manual_usb_channels(self): + argv = build_qemu_argv(self.profile, self.runtime.paths('guest')) + self.assertIn('spice,id=audio', argv) + self.assertIn('hda-duplex,audiodev=audio', argv) + for channel in range(3): + self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv) + self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv) + self.assertFalse(any('usb-host' in argument for argument in argv)) + + def test_metadata_symlink_does_not_overwrite_user_data(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + valuable = self.root / 'valuable data' + valuable.write_text('preserve me') + paths.metadata.symlink_to(valuable) + self.runtime.start('guest') + self.assertEqual(valuable.read_text(), 'preserve me') + self.assertFalse(paths.metadata.is_symlink()) + + def test_disk_symlink_never_changes_its_target(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + valuable = self.root / 'valuable disk' + valuable.write_text('preserve me') + paths.disk.symlink_to(valuable) + with self.assertRaises(VMError): + self.runtime.start('guest') + self.assertEqual(valuable.read_text(), 'preserve me') + + def test_malformed_identity_record_is_safe_unknown(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.metadata.write_text('[]') + server = QMPServer(paths.qmp, greeting) + self.addCleanup(server.close) + self.assertEqual(self.runtime.status('guest')['status'], 'unknown') + + def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid})) + commands = [] + def handler(conn): + greeting(conn) + for line in conn.makefile('rb'): + req = json.loads(line) + commands.append(req['execute']) + result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {}) + send(conn, {'return': result, 'id': req['id']}) + server = QMPServer(paths.qmp, handler) + self.addCleanup(server.close) + with self.assertRaises(VMError): + self.runtime.stop('guest', force=True) + self.assertNotIn('quit', commands) + + def test_reused_pid_without_qmp_never_controls_a_process(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.pid.write_text(str(os.getpid())) + self.assertEqual(self.runtime.status('guest')['status'], 'stopped') + with patch('os.kill', side_effect=AssertionError('must not signal any PID')): + with self.assertRaises(VMError): + self.runtime.stop('guest', force=True) + + def test_malformed_query_result_is_safe_unknown(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid})) + def handler(conn): + greeting(conn) + for line in conn.makefile('rb'): + req = json.loads(line) + send(conn, {'return': [], 'id': req['id']}) + server = QMPServer(paths.qmp, handler) + self.addCleanup(server.close) + self.assertEqual(self.runtime.status('guest')['status'], 'unknown') + + def test_cli_qmp_failure_reports_error_without_traceback(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid})) + def handler(conn): + greeting(conn) + for line in conn.makefile('rb'): + req = json.loads(line) + result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {}) + if req['execute'] == 'system_powerdown': + send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']}) + else: + send(conn, {'return': result, 'id': req['id']}) + server = QMPServer(paths.qmp, handler) + self.addCleanup(server.close) + run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config), + '--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime), + 'shutdown', 'guest'], capture_output=True, text=True, timeout=4) + self.assertEqual(run.returncode, 1) + self.assertIn('cannot power down', run.stderr) + self.assertNotIn('Traceback', run.stderr) + + def test_disk_and_profile_survive_shutdown_and_force(self): + paths = self.runtime.paths('guest') + paths.disk.parent.mkdir(parents=True, exist_ok=True) + paths.disk.write_bytes(b'existing valuable disk') + self.runtime.start('guest') + self.assertEqual(self.runtime.status('guest')['status'], 'running') + self.runtime.shutdown('guest') + self.assertEqual(self.runtime.status('guest')['status'], 'running') + with self.assertRaises(VMError): + self.runtime.stop('guest', force=False) + self.runtime.stop('guest', force=True) + self.assertEqual(self.runtime.status('guest')['status'], 'stopped') + self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk') + + def test_simultaneous_start_only_launches_one_vm(self): + barrier = threading.Barrier(2) + results = [] + def start(): + barrier.wait() + try: + results.append(self.runtime.start('guest')['status']) + except VMError as exc: + results.append(str(exc)) + threads = [threading.Thread(target=start) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(4) + self.assertFalse(thread.is_alive()) + self.assertEqual(results.count('running'), 1) + self.assertEqual(len(results), 2) + self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture') + + def test_stale_socket_is_replaced_only_when_no_listener(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + stale = socket.socket(socket.AF_UNIX) + stale.bind(str(paths.qmp)) + stale.close() + self.runtime.start('guest') + self.assertEqual(self.runtime.status('guest')['status'], 'running') + + def test_failed_boot_is_reported_and_disk_preserved(self): + with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError): + self.runtime.start('guest') + self.assertEqual(self.runtime.status('guest')['status'], 'stopped') + self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture') + self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text()) + + def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self): + paths = self.runtime.paths('guest') + self.runtime.prepare(paths) + paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid})) + paths.pid.write_text(str(os.getpid())) + commands = [] + def handler(conn): + greeting(conn) + for line in conn.makefile('rb'): + req = json.loads(line) + commands.append(req['execute']) + result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {}) + send(conn, {'return': result, 'id': req['id']}) + server = QMPServer(paths.qmp, handler) + self.addCleanup(server.close) + self.assertEqual(self.runtime.status('guest')['status'], 'unknown') + for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True), + lambda: self.runtime.start('guest')): + with self.assertRaises(VMError): + control() + self.assertNotIn('quit', commands) + self.assertNotIn('system_powerdown', commands) + self.assertFalse(paths.disk.exists()) + + def test_private_runtime_and_socket_length(self): + self.runtime.prepare(self.runtime.paths('guest')) + self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700) + with self.assertRaises(VMError): + Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest') + + def test_cli_lifecycle_and_global_paths(self): + base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state), + '--runtime', str(self.runtime.runtime)] + for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'), + (['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]: + run = subprocess.run(base + args, text=True, capture_output=True, timeout=4) + self.assertEqual(run.returncode, 0, run.stderr) + self.assertIn(expected, run.stdout) + + +if __name__ == '__main__': + unittest.main()