From 972b82c24b345eb6629101466c71515d67174b27 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 04:49:54 +0300 Subject: [PATCH] Probe installed user VM and audio continuity across logout --- dev/iso-smoke.py | 258 +++++++++++++++++++++++++++++++++++++++- tests/test_iso_smoke.py | 129 ++++++++++++++++++++ 2 files changed, 386 insertions(+), 1 deletion(-) diff --git a/dev/iso-smoke.py b/dev/iso-smoke.py index 32c99f4..bc44fce 100644 --- a/dev/iso-smoke.py +++ b/dev/iso-smoke.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse import base64 import hashlib +import inspect import json import os from pathlib import Path @@ -32,6 +33,257 @@ class SmokeError(RuntimeError): pass +def lifecycle_socket(path, *, connect_socket=True): + path = Path(path) + info = path.lstat() + if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.getuid(): + raise SmokeError('Lifecycle socket is not owned by the installed user') + identity = [info.st_dev, info.st_ino, info.st_uid, info.st_mode] + if connect_socket: + with socket.socket(socket.AF_UNIX) as connection: + connection.settimeout(2) + connection.connect(str(path)) + after = path.lstat() + if identity != [after.st_dev, after.st_ino, after.st_uid, after.st_mode]: + raise SmokeError('Lifecycle socket changed while connecting') + return identity + + +def lifecycle_snapshot(context): + """Prove live identity and owned runtime endpoints, never trust PID alone.""" + try: + if context['uid'] != os.getuid(): + raise SmokeError('Lifecycle probe is not the installed normal user') + runtime = Path(context['runtime']) + run = runtime / context['name'] + directories = {} + for directory in (runtime, run): + info = directory.lstat() + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() + or stat.S_IMODE(info.st_mode) != 0o700): + raise SmokeError('Lifecycle runtime must remain private and owned') + directories[str(directory)] = [info.st_dev, info.st_ino, info.st_mode] + metadata = json.loads((run / 'identity.json').read_text()) + if metadata != {'name': context['name'], 'uuid': context['uuid']}: + raise SmokeError('Lifecycle metadata is stale or foreign') + pid = int((run / 'qemu.pid').read_text()) + process = Path('/proc') / str(pid) + if process.stat().st_uid != os.getuid(): + raise SmokeError('Lifecycle process belongs to another user') + fields = (process / 'stat').read_text().rsplit(')', 1)[1].split() + if fields[0] == 'Z': + raise SmokeError('Lifecycle process has exited') + argv = (process / 'cmdline').read_bytes().split(b'\0') + for option, expected in ((b'-name', context['name']), (b'-uuid', context['uuid'])): + if argv[argv.index(option) + 1] != expected.encode(): + raise SmokeError('Lifecycle PID is not the owned VM identity') + endpoints = {'qmp': lifecycle_socket(run / 'qmp.sock', connect_socket=False)} + with QMPClient(run / 'qmp.sock', timeout=2) as client: + if (client.command('query-name') != {'name': context['name']} + or client.command('query-uuid') != {'UUID': context['uuid']}): + raise SmokeError('Lifecycle QMP identity is foreign') + if client.command('query-status').get('status') != 'running': + raise SmokeError('Lifecycle VM is not running') + if not client.command('query-kvm').get('enabled'): + raise SmokeError('Lifecycle VM did not initialise nested KVM') + endpoints['spice'] = lifecycle_socket(run / 'spice.sock') + audio = Path(context['user_runtime']) + endpoints['pipewire'] = lifecycle_socket(audio / 'pipewire-0') + endpoints['pulse'] = lifecycle_socket(audio / 'pulse/native') + services = {} + for service in context.get('audio_services', []): + subprocess.run(['systemctl', '--user', 'is-active', service], check=True, + capture_output=True, timeout=5) + services[service] = subprocess.run(['systemctl', '--user', 'show', service, + '-p', 'InvocationID', '-p', 'ExecMainPID'], check=True, + capture_output=True, text=True, timeout=5).stdout + return {'name': context['name'], 'uuid': context['uuid'], 'pid': pid, + 'process_start': fields[19], 'directories': directories, + 'sockets': endpoints, 'audio_services': services} + except (OSError, ValueError, KeyError, IndexError, QMPError, subprocess.SubprocessError) as exc: + raise SmokeError('Lifecycle runtime or process was lost: ' + str(exc)) from exc + + +def verify_lifecycle_survival(before, after): + if before != after: + raise SmokeError('VM identity, private runtime, or audio endpoints changed across logout') + + +LIFECYCLE_BODY = r''' +def run(*args): + return subprocess.run(args, check=True, capture_output=True, text=True, timeout=40).stdout + +def cli(context, *args): + return json.loads(run('hydra', '--config', context['config'], '--state', context['state'], + '--runtime', context['runtime'], *args)) + +def no_user_sessions(uid): + rows = run('loginctl', 'list-sessions', '--no-legend', '--no-pager').splitlines() + return not any(len(row.split()) > 1 and row.split()[1] == str(uid) for row in rows) + +def write_report(path, data): + temporary = path.with_suffix('.tmp') + temporary.write_text(json.dumps(data, indent=2) + '\n') + temporary.chmod(0o600) + temporary.replace(path) + +mode, base_arg = sys.argv[1:] +base = Path(base_arg) +context_path = base / 'context.json' +observation_path = base / 'logout-observation.json' +if mode == 'start': + if not Path('/dev/kvm').exists(): + print('SKIP: installed guest nested KVM unavailable') + raise SystemExit(0) + if os.getuid() == 0 or not os.access('/dev/kvm', os.R_OK | os.W_OK): + raise SmokeError('Lifecycle test needs installed-user nested KVM access') + user_runtime = Path(os.environ['XDG_RUNTIME_DIR']) + if user_runtime != Path('/run/user') / str(os.getuid()): + raise SmokeError('Lifecycle test needs the real login runtime directory') + if run('loginctl', 'show-user', str(os.getuid()), '-p', 'Linger', '--value').strip() != 'no': + raise SmokeError('Lingering would mask the installed logout policy') + rows = run('loginctl', 'list-sessions', '--no-legend', '--no-pager').splitlines() + if sum(len(row.split()) > 1 and row.split()[1] == str(os.getuid()) for row in rows) != 1: + raise SmokeError('Lifecycle test requires exactly the serial user login') + name = 'logout-' + base.name.rsplit('-', 1)[1] + state = base / 'state' + runtime = user_runtime / base.name + context = {'uid': os.getuid(), 'name': name, 'config': str(base / 'config'), + 'state': str(state), 'runtime': str(runtime), 'user_runtime': str(user_runtime), + 'uuid': str(uuid.uuid5(uuid.NAMESPACE_URL, str(state / name / 'disk.qcow2'))), + 'audio_services': ['pipewire.service', 'pipewire-pulse.service', 'wireplumber.service']} + run('systemctl', '--user', 'start', *context['audio_services']) + cli(context, 'add', name, '--memory', '256', '--cpus', '1', '--disk-size', '64M', + '--graphics', 'software', '--network', 'bridge', '--bridge', 'hydrabr0') + if cli(context, 'start', name).get('status') != 'running': + raise SmokeError('Real installed hydra CLI failed to start lifecycle VM') + try: + disk = Path(context['state']) / name / 'disk.qcow2' + info = disk.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid(): + raise SmokeError('Lifecycle VM disk is not a new owned regular qcow2') + context['before'] = lifecycle_snapshot(context) + except Exception: + # CLI control is scoped to the newly created private identity. + cli(context, 'stop', name, '--force') + raise + write_report(context_path, context) + with open(base / 'monitor.log', 'ab', buffering=0) as log: + subprocess.Popen([sys.executable, __file__, 'monitor', str(base)], + stdin=subprocess.DEVNULL, stdout=log, stderr=log, + start_new_session=True) + print('READY: normal-user Hydra VM and audio sockets recorded before real logout') +elif mode == 'monitor': + context = json.loads(context_path.read_text()) + try: + deadline = time.monotonic() + 35 + while not no_user_sessions(context['uid']): + if time.monotonic() >= deadline: + raise SmokeError('Serial session did not actually log out') + time.sleep(.2) + # Exceed systemd's usual 10-second user-manager stop delay. No helper + # login or lingering is allowed to keep the tested user session alive. + time.sleep(15) + if not no_user_sessions(context['uid']): + raise SmokeError('Another user login masked the logout interval') + during = lifecycle_snapshot(context) + verify_lifecycle_survival(context['before'], during) + write_report(observation_path, {'status': 'passed', 'no_user_sessions': True, + 'survival_seconds': 15, 'during_logout': during}) + except Exception as exc: + write_report(observation_path, {'status': 'failed', 'error': str(exc)}) + raise +elif mode == 'finish': + context = json.loads(context_path.read_text()) + try: + if os.environ.get('XDG_RUNTIME_DIR') != context['user_runtime']: + raise SmokeError('Relogin replaced the user runtime path') + deadline = time.monotonic() + 10 + while not observation_path.is_file(): + if time.monotonic() >= deadline: + raise SmokeError('Logout monitor died or did not record survival') + time.sleep(.2) + observation = json.loads(observation_path.read_text()) + if observation.get('status') != 'passed' or observation.get('no_user_sessions') is not True: + raise SmokeError('No-session observation failed: ' + json.dumps(observation)) + after = lifecycle_snapshot(context) + verify_lifecycle_survival(context['before'], after) + if cli(context, 'status', context['name']).get('status') != 'running': + raise SmokeError('Real installed hydra CLI could not reconnect after login') + print(json.dumps({'status': 'passed', 'real_logout': True, 'no_user_sessions': True, + 'survival_seconds': observation['survival_seconds'], 'same_vm': after, + 'audio_scope': 'Socket and service continuity only; signal/playback untested'})) + finally: + # Real product CLI verifies metadata and live QMP name/UUID before quit. + # Never signal a saved PID or control a VM outside these private paths. + cleanup_status = cli(context, 'status', context['name']).get('status') + if cleanup_status == 'running': + run_path = Path(context['runtime']) / context['name'] + if json.loads((run_path / 'identity.json').read_text()) != {'name': context['name'], 'uuid': context['uuid']}: + raise SmokeError('Refusing lifecycle cleanup for a foreign identity') + with QMPClient(run_path / 'qmp.sock', timeout=2) as client: + if (client.command('query-name') != {'name': context['name']} + or client.command('query-uuid') != {'UUID': context['uuid']}): + raise SmokeError('Refusing lifecycle cleanup for foreign live QMP') + if cli(context, 'stop', context['name'], '--force').get('status') != 'stopped': + raise SmokeError('Owned lifecycle VM did not stop') + elif cleanup_status != 'stopped': + raise SmokeError('Cannot safely verify the lifecycle VM for cleanup') +else: + raise SmokeError('Unknown lifecycle mode') +''' + + +def lifecycle_guest_script(): + helpers = (lifecycle_socket, lifecycle_snapshot, verify_lifecycle_survival) + return ('import json, os, socket, stat, subprocess, sys, time, uuid\n' + 'from pathlib import Path\nsys.path.insert(0, "/usr/lib/hydra")\n' + 'from hydra.qmp import QMPClient, QMPError\n' + 'class SmokeError(RuntimeError): pass\n' + + '\n'.join(inspect.getsource(helper) for helper in helpers) + LIFECYCLE_BODY) + + +def installed_admin(console, password): + console.buffer = b'' + prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode() + console.send(f'sudo -k -p "$(printf %s {prompt_encoded} | base64 -d)" -i\n') + console.expect(rb'HYDRA_SUDO_PASSWORD: ', 30) + console.send(password + '\n') + console.expect(rb'root[^\r\n]*[#$] ', 30) + + +def installed_logout_probe(console, password): + # Exit only the guest sudo shell, then work in its actual getty PAM login. + console.send('exit\n') + console.expect(rb'hydratest[^\r\n]*[#$] ', 30) + if console.command('id -un', 10).strip() != b'hydratest': + raise SmokeError('Logout probe did not return to installed user login') + base = '/home/hydratest/hydra-logout-' + secrets.token_hex(4) + script = base + '/probe.py' + preparation = (f'umask 077; mkdir {shlex.quote(base)} && cat > {shlex.quote(script)} ' + "<<'HYDRA_LIFECYCLE_SOURCE'\n" + lifecycle_guest_script() + '\nHYDRA_LIFECYCLE_SOURCE\n' + + f'python3 {shlex.quote(script)} start {shlex.quote(base)}') + ready = console.command(preparation, 120) + if ready.startswith(b'SKIP:'): + installed_admin(console, password) + return ready + if b'READY:' not in ready: + raise SmokeError('Lifecycle VM did not become ready before logout') + console.buffer = b'' + console.send('exit\n') + console.expect(rb'login: ', 30) + # This is intentionally a period at the actual login prompt with no + # replacement login. The guest monitor independently confirms no sessions. + time.sleep(20) + console.send('hydratest\n') + console.expect(rb'Password: ', 30) + console.send(password + '\n') + console.expect(rb'hydratest[^\r\n]*[#$] ', 60) + result = console.command(f'python3 {shlex.quote(script)} finish {shlex.quote(base)}', 90) + installed_admin(console, password) + return result + + def reserve_disk(path): try: fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) @@ -85,9 +337,12 @@ def frame_command(command, token): if not re.fullmatch(r'[a-zA-Z0-9]+', token): raise SmokeError('Invalid framing token') encoded = base64.b64encode(command.encode()).decode() + # Linux canonical serial TTY input is bounded per physical line. Continue + # the shell expression across short lines for substantial guest probes. + chunks = ' \\\n'.join(shlex.quote(encoded[i:i + 768]) for i in range(0, len(encoded), 768)) or "''" # Literal control bytes appear only in output, never in the echoed command. return (f"printf '\\036{token}:BEGIN\\037'; " - f"sh -c \"$(printf %s {shlex.quote(encoded)} | base64 -d)\"; " + f"sh -c \"$(printf %s {chunks} | base64 -d)\"; " f"rc=$?; printf '\\036{token}:END:%s\\037' \"$rc\"") @@ -508,6 +763,7 @@ def main(argv=None): checkpoint('installed_contract', console.command(INSTALLED_CHECK, 120)) checkpoint('lan_to_vm_tcp', console.command(NETWORK_CHECK, 120)) checkpoint('nested_kvm', console.command('runuser -u hydratest -- sh -c ' + shlex.quote(NESTED_CHECK), 90)) + checkpoint('installed_user_logout', installed_logout_probe(console, password)) console.send('poweroff\n') vm.wait_poweroff(90) checkpoint('installed_poweroff', b'Normal installed guest shutdown') diff --git a/tests/test_iso_smoke.py b/tests/test_iso_smoke.py index 457dc6d..fcb73d7 100644 --- a/tests/test_iso_smoke.py +++ b/tests/test_iso_smoke.py @@ -297,3 +297,132 @@ else: print('fixture tool') def test_tcp_probe_rejects_truncated_marker(self): self.assertNotEqual(self.tcp_probe([b'HYDRA_LAN_TO_']).returncode, 0) + + def lifecycle_fixture(self): + import json + import os + import uuid + name = 'logout-test' + runtime = self.root / 'private' + run = runtime / name + run.mkdir(parents=True, mode=0o700) + runtime.chmod(0o700) + audio = self.root / 'user-runtime' + (audio / 'pulse').mkdir(parents=True) + context = {'name': name, 'uuid': str(uuid.uuid4()), 'runtime': str(runtime), + 'user_runtime': str(audio), 'uid': os.getuid()} + process = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)', + '-name', name, '-uuid', context['uuid']]) + self.addCleanup(lambda: process.wait(timeout=3)) + self.addCleanup(lambda: process.poll() is None and process.kill()) + (run / 'qemu.pid').write_text(str(process.pid)) + (run / 'identity.json').write_text(json.dumps({'name': name, 'uuid': context['uuid']})) + listeners = [] + for path in (run / 'qmp.sock', run / 'spice.sock', audio / 'pipewire-0', audio / 'pulse/native'): + listener = socket.socket(socket.AF_UNIX) + listener.bind(str(path)) + listener.listen(5) + listener.settimeout(.1) + listeners.append(listener) + self.addCleanup(listener.close) + stopped = threading.Event() + def qmp_server(): + while not stopped.is_set(): + try: + connection, _ = listeners[0].accept() + except socket.timeout: + continue + except OSError: + break + with connection, connection.makefile('rwb', buffering=0) as stream: + stream.write(b'{"QMP": {}}\n') + for line in stream: + request = json.loads(line) + replies = {'query-name': {'name': context.get('qmp_name', name)}, + 'query-uuid': {'UUID': context['uuid']}, + 'query-status': {'status': 'running'}, + 'query-kvm': {'enabled': True, 'present': True}} + stream.write(json.dumps({'return': replies.get(request['execute'], {}), + 'id': request['id']}).encode() + b'\n') + thread = threading.Thread(target=qmp_server) + thread.start() + self.addCleanup(lambda: thread.join(timeout=3)) + self.addCleanup(stopped.set) + return context, process, run + + def test_logout_probe_detects_lost_private_socket(self): + context, process, run = self.lifecycle_fixture() + before = self.mod.lifecycle_snapshot(context) + self.assertEqual(self.mod.lifecycle_snapshot(context), before) + (run / 'spice.sock').unlink() + with self.assertRaises(self.mod.SmokeError): + self.mod.lifecycle_snapshot(context) + + def test_logout_probe_rejects_stale_metadata_and_foreign_qmp_identity(self): + import json + context, process, run = self.lifecycle_fixture() + self.mod.lifecycle_snapshot(context) + (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': 'stale'})) + with self.assertRaises(self.mod.SmokeError): + self.mod.lifecycle_snapshot(context) + (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': context['uuid']})) + context['qmp_name'] = 'foreign-process' + with self.assertRaises(self.mod.SmokeError): + self.mod.lifecycle_snapshot(context) + + def test_logout_probe_rejects_dead_process_and_changed_runtime_socket(self): + context, process, run = self.lifecycle_fixture() + before = self.mod.lifecycle_snapshot(context) + audio = Path(context['user_runtime']) / 'pipewire-0' + audio.unlink() + replacement = socket.socket(socket.AF_UNIX) + self.addCleanup(replacement.close) + replacement.bind(str(audio)) + replacement.listen(1) + after = self.mod.lifecycle_snapshot(context) + with self.assertRaises(self.mod.SmokeError): + self.mod.verify_lifecycle_survival(before, after) + process.terminate() + process.wait(timeout=3) + with self.assertRaises(self.mod.SmokeError): + self.mod.lifecycle_snapshot(context) + + def test_logout_probe_accepts_verified_same_runtime_and_rejects_process_restart(self): + context, process, run = self.lifecycle_fixture() + before = self.mod.lifecycle_snapshot(context) + self.mod.verify_lifecycle_survival(before, self.mod.lifecycle_snapshot(context)) + process.terminate() + process.wait(timeout=3) + replacement = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)', + '-name', context['name'], '-uuid', context['uuid']]) + self.addCleanup(lambda: replacement.wait(timeout=3)) + self.addCleanup(lambda: replacement.poll() is None and replacement.kill()) + (run / 'qemu.pid').write_text(str(replacement.pid)) + changed = self.mod.lifecycle_snapshot(context) + with self.assertRaises(self.mod.SmokeError): + self.mod.verify_lifecycle_survival(before, changed) + + def test_large_guest_probe_survives_serial_tty_canonical_line_limit(self): + import os + import pty + import termios + master, slave = pty.openpty() + self.addCleanup(os.close, master) + self.addCleanup(os.close, slave) + options = termios.tcgetattr(slave) + options[3] &= ~termios.ECHO + termios.tcsetattr(slave, termios.TCSANOW, options) + process = subprocess.Popen(['/bin/sh'], stdin=slave, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + self.addCleanup(process.stdout.close) + self.addCleanup(process.stderr.close) + self.addCleanup(lambda: process.wait(timeout=3)) + self.addCleanup(lambda: process.poll() is None and process.kill()) + payload = 'guest-proof-' * 1200 + command = self.mod.frame_command('printf %s ' + payload, 'longprobe') + os.write(master, (command + '\nexit\n').encode()) + output, error = process.communicate(timeout=3) + self.assertEqual(self.mod.parse_frame(output, 'longprobe'), (0, payload.encode()), error) + + def test_generated_guest_lifecycle_probe_compiles(self): + compile(self.mod.lifecycle_guest_script(), '', 'exec')