Capture redacted guest serial while waiting for shutdown

dev/iso-smoke.py

@@ -0,0 +1,911 @@
+#!/usr/bin/env python3
+"""Boot/install/reboot acceptance on a fresh, isolated UEFI/KVM VM only.
+
+No existing disk option, host sudo, SSH, host TTY, or GUI is used. Artifacts and
+logs remain in a new private scratch directory even after failure.
+"""
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import inspect
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import select
+import shlex
+import shutil
+import socket
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from hydra.qmp import QMPClient, QMPError
+
+
+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')
+
+
+def login_session_records(uid):
+    """Read logind's public properties, avoiding display-table row semantics."""
+    def query(*args):
+        return subprocess.run(['loginctl', *args], check=True, capture_output=True,
+                              text=True, timeout=5).stdout
+    try:
+        ids = query('show-user', str(uid), '-p', 'Sessions', '--value').split()
+        records = []
+        for ident in ids:
+            output = query('show-session', ident, '--all', '--no-pager',
+                           '-p', 'Id', '-p', 'User', '-p', 'State', '-p', 'Class',
+                           '-p', 'Leader', '-p', 'TTY')
+            record = dict(line.split('=', 1) for line in output.splitlines())
+            if not {'Id', 'User', 'State', 'Class', 'Leader', 'TTY'} <= record.keys():
+                raise SmokeError('Incomplete logind session identity')
+            record['User'], record['Leader'] = int(record['User']), int(record['Leader'])
+            if record['Id'] != ident or record['User'] != uid:
+                raise SmokeError('Logind returned a foreign session identity')
+            records.append(record)
+        return records
+    except (OSError, ValueError, subprocess.SubprocessError) as exc:
+        raise SmokeError('Cannot establish guest login session state: ' + str(exc)) from exc
+
+
+def is_open_login(record):
+    # systemd v258: closing session scopes can remain while user processes
+    # survive. Manager-class records describe bookkeeping, not a PAM login.
+    users = {'user', 'user-early', 'user-incomplete', 'user-light', 'user-early-light'}
+    bookkeeping = {'manager', 'manager-early', 'background', 'background-light',
+                   'greeter', 'lock-screen', 'none'}
+    if record['State'] not in {'opening', 'online', 'active', 'closing'}:
+        raise SmokeError('Unknown logind session state')
+    if record['Class'] not in users | bookkeeping:
+        raise SmokeError('Unknown logind session class')
+    return record['Class'] in users and record['State'] != 'closing'
+
+
+def require_initial_login(uid, ident):
+    opened = [record for record in login_session_records(uid) if is_open_login(record)]
+    if (len(opened) != 1 or opened[0]['Id'] != ident or opened[0]['TTY'] != 'ttyS0'
+            or opened[0]['Leader'] <= 0 or opened[0]['State'] not in {'online', 'active'}):
+        raise SmokeError('Lifecycle test requires exactly its actual serial PAM login')
+    return opened[0]
+
+
+def no_open_logins(context):
+    original = context['login_session']
+    records = login_session_records(context['uid'])
+    for record in records:
+        if record['Id'] == original['Id']:
+            if (record['User'] != original['User'] or record['TTY'] != original['TTY']
+                    or record['Class'] != original['Class']
+                    or record['Leader'] not in {0, original['Leader']}):
+                raise SmokeError('Original PAM session identity was replaced')
+            if record['State'] != 'closing':
+                return False
+        if is_open_login(record):
+            return False
+    return True
+
+
+def capture_logout_snapshot(context):
+    if not no_open_logins(context):
+        raise SmokeError('An open login overlaps the logout sample')
+    during = lifecycle_snapshot(context)
+    if not no_open_logins(context):
+        raise SmokeError('A login opened while the logout sample was running')
+    verify_lifecycle_survival(context['before'], during)
+    return during
+
+
+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 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')
+    login_session = require_initial_login(os.getuid(), os.environ['XDG_SESSION_ID'])
+    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'))),
+               'login_session': login_session,
+               '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_open_logins(context):
+            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)
+        during = capture_logout_snapshot(context)
+        write_report(observation_path, {'status': 'passed', 'no_open_logins': 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_open_logins') is not True:
+            raise SmokeError('No-open-login 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_open_logins': 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,
+               login_session_records, is_open_login, require_initial_login,
+               no_open_logins, capture_logout_snapshot)
+    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 open logins.
+    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)
+        os.close(fd)
+    except OSError as exc:
+        raise SmokeError('Test disk must be a newly created regular file') from exc
+
+
+def owned_file(path, scratch):
+    path, scratch = Path(path), Path(scratch)
+    try:
+        info = path.lstat()
+    except OSError as exc:
+        raise SmokeError('Missing disposable drive') from exc
+    if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1
+            or path.parent.resolve() != scratch.resolve()
+            or info.st_uid != os.getuid()):
+        raise SmokeError('Drive must be an owned, unlinked regular file directly in scratch')
+
+
+def qemu_path(path):
+    # QEMU keyval arguments use commas as separators. Refuse ambiguous paths.
+    if any(c in str(path) for c in ',\n\r\x00'):
+        raise SmokeError('QEMU file paths cannot contain comma or control characters')
+    return str(path)
+
+
+def qemu_argv(qemu, scratch, disk, code, variables, iso, memory):
+    scratch = Path(scratch)
+    owned_file(disk, scratch)
+    owned_file(variables, scratch)
+    argv = [qemu, '-machine', 'q35,accel=kvm', '-cpu', 'host', '-smp', '2',
+            '-m', str(memory), '-display', 'none', '-monitor', 'none',
+            '-no-reboot', '-nic', 'user,model=virtio-net-pci',
+            '-drive', f'if=pflash,format=raw,readonly=on,file={qemu_path(code)}',
+            '-drive', f'if=pflash,format=raw,file={qemu_path(variables)}',
+            '-drive', f'if=virtio,format=qcow2,file={qemu_path(disk)}',
+            '-chardev', f'socket,id=console,path={qemu_path(scratch / "serial.sock")},server=on,wait=off',
+            '-serial', 'chardev:console', '-qmp', f'unix:{qemu_path(scratch / "qmp.sock")},server=on,wait=off']
+    if iso is not None:
+        if not Path(iso).is_file():
+            raise SmokeError('Trusted ISO must exist as a regular file')
+        argv += ['-drive', f'file={qemu_path(iso)},media=cdrom,format=raw,readonly=on',
+                 '-boot', 'order=d,strict=on']
+    else:
+        argv += ['-boot', 'order=c,strict=on']
+    return argv
+
+
+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 {chunks} | base64 -d)\"; "
+            f"rc=$?; printf '\\036{token}:END:%s\\037' \"$rc\"")
+
+
+def parse_frame(data, token):
+    begin = b'\x1e' + token.encode() + b':BEGIN\x1f'
+    start = data.find(begin)
+    if start < 0:
+        return None
+    end = re.search(rb'\x1e' + token.encode() + rb':END:(\d+)\x1f', data[start + len(begin):])
+    if end is None:
+        return None
+    output = data[start + len(begin):start + len(begin) + end.start()]
+    return int(end.group(1)), output
+
+
+def installer_command(token):
+    # A literal control-byte return frame cannot be forged by terminal echo.
+    # Keep the installer on this controlling TTY for getpass, then mark every
+    # shell return, including an unexpected status zero without completion.
+    return ('sudo -n hydra-install --disk /dev/vda --user hydratest --execute '
+            '--confirm-erase /dev/vda; rc=$?; '
+            f"printf '\\036{token}:INSTALL_EXIT:%s\\037' \"$rc\"")
+
+
+class Console:
+    def __init__(self, connection, logfile, passwords):
+        self.sock = connection
+        self.log = open(logfile, 'ab', buffering=0)
+        self.passwords = [p.encode() for p in passwords if p]
+        self.pending_log = b''
+        self.buffer = b''
+        self.closed = False
+
+    def record(self, data, final=False):
+        self.pending_log += data
+        for password in self.passwords:
+            self.pending_log = self.pending_log.replace(password, b'[REDACTED]')
+        keep = 0
+        if not final:
+            for password in self.passwords:
+                for length in range(1, len(password)):
+                    if self.pending_log.endswith(password[:length]):
+                        keep = max(keep, length)
+        if keep:
+            self.log.write(self.pending_log[:-keep])
+            self.pending_log = self.pending_log[-keep:]
+        else:
+            self.log.write(self.pending_log)
+            self.pending_log = b''
+
+    def close(self):
+        if not self.closed:
+            self.record(b'', final=True)
+            self.log.close()
+            self.sock.close()
+            self.closed = True
+
+    def send(self, data):
+        self.sock.sendall(data.encode() if isinstance(data, str) else data)
+
+    def read(self, timeout):
+        ready, _, _ = select.select([self.sock], [], [], max(0, timeout))
+        if not ready:
+            return b''
+        data = self.sock.recv(65536)
+        if not data:
+            raise SmokeError('Guest console closed')
+        self.record(data)
+        self.buffer += data
+        if len(self.buffer) > 16 * 1024 * 1024:
+            raise SmokeError('Console exceeded bounded capture size')
+        return data
+
+    def expect(self, pattern, timeout, failure_pattern=None):
+        deadline = time.monotonic() + timeout
+        while True:
+            match = re.search(pattern, self.buffer)
+            failure = re.search(failure_pattern, self.buffer) if failure_pattern else None
+            if failure and (not match or failure.start() < match.start()):
+                evidence = failure.group()
+                for password in self.passwords:
+                    evidence = evidence.replace(password, b'[REDACTED]')
+                raise SmokeError('Installer exited before expected console state: ' +
+                                 evidence.decode(errors='replace'))
+            if match:
+                self.buffer = self.buffer[match.end():]
+                return match
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Timed out waiting for guest console state')
+            self.read(min(remaining, 1))
+
+    def expect_installer(self, pattern, timeout, token):
+        failure = (rb'hydra-install: [^\r\n]*[\r\n]|\x1e' +
+                   re.escape(token.encode()) + rb':INSTALL_EXIT:[0-9]+\x1f')
+        return self.expect(pattern, timeout, failure_pattern=failure)
+
+    def finish_installer(self, token, timeout=30):
+        # Completion text can arrive before the process exits and readline
+        # resumes. Input sent during that gap can be discarded by the TTY.
+        deadline = time.monotonic() + timeout
+        returned = self.expect(rb'\x1e' + re.escape(token.encode()) +
+                               rb':INSTALL_EXIT:([0-9]+)\x1f', timeout)
+        status = int(returned.group(1))
+        if status:
+            raise SmokeError(f'Installer returned status {status} after completion')
+        self.expect(rb'hydra-live[^\r\n]*[#$] ', max(0, deadline - time.monotonic()))
+
+    def command(self, command, timeout=60):
+        token = secrets.token_hex(12)
+        self.buffer = b''
+        self.send(frame_command(command, token) + '\n')
+        deadline = time.monotonic() + timeout
+        while True:
+            parsed = parse_frame(self.buffer, token)
+            if parsed is not None:
+                status, output = parsed
+                self.buffer = b''
+                if status:
+                    raise SmokeError(f'Guest probe failed with status {status}: {output.decode(errors="replace")[-3000:]}')
+                return output.replace(b'\r\n', b'\n')
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Timed out waiting for framed guest command')
+            self.read(min(remaining, 1))
+
+
+def connect(path, process, timeout=30):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise SmokeError(f'Test QEMU exited with status {process.returncode}')
+        connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        try:
+            connection.connect(str(path))
+            return connection
+        except OSError:
+            connection.close()
+            time.sleep(0.1)
+    raise SmokeError('Timed out connecting to private test socket')
+
+
+class VM:
+    def __init__(self, args, scratch, disk, iso, password, phase):
+        self.scratch = scratch
+        for name in ('serial.sock', 'qmp.sock'):
+            (scratch / name).unlink(missing_ok=True)
+        self.stderr = open(scratch / f'{phase}-qemu.log', 'ab', buffering=0)
+        self.process = subprocess.Popen(qemu_argv(args.qemu, scratch, disk,
+            args.ovmf_code, scratch / 'OVMF_VARS.fd', iso, args.memory),
+            stdin=subprocess.DEVNULL, stdout=self.stderr, stderr=self.stderr)
+        self.console = None
+        self.qmp = None
+        try:
+            self.console = Console(connect(scratch / 'serial.sock', self.process),
+                                   scratch / f'{phase}-serial.log', [password])
+            ready = connect(scratch / 'qmp.sock', self.process)
+            ready.close()
+            self.qmp = QMPClient(scratch / 'qmp.sock', timeout=5)
+            self.kvm = self.qmp.command('query-kvm')
+            if not self.kvm.get('enabled'):
+                raise SmokeError('Disposable QEMU is not using KVM')
+        except BaseException:
+            self.close()
+            raise
+
+    def wait_poweroff(self, timeout):
+        deadline = time.monotonic() + timeout
+        serial_open = True
+        while self.process.poll() is None:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise SmokeError('Guest failed to power off in time')
+            if serial_open:
+                try:
+                    self.console.read(min(remaining, 0.25))
+                except SmokeError as exc:
+                    if str(exc) != 'Guest console closed':
+                        raise
+                    serial_open = False
+            else:
+                try:
+                    self.process.wait(timeout=remaining)
+                except subprocess.TimeoutExpired as exc:
+                    raise SmokeError('Guest failed to power off in time') from exc
+        # Capture bytes queued immediately before exit, even if poll observed
+        # the child exiting before the first serial read.
+        while serial_open:
+            try:
+                if not self.console.read(0):
+                    break
+            except SmokeError as exc:
+                if str(exc) != 'Guest console closed':
+                    raise
+                serial_open = False
+        rc = self.process.returncode
+        if rc:
+            raise SmokeError(f'Test QEMU failed during poweroff ({rc})')
+
+    def close(self):
+        # Only the Popen child owned by this harness is signalled.
+        if self.qmp:
+            if self.process.poll() is None:
+                try:
+                    self.qmp.command('quit')
+                    self.process.wait(timeout=5)
+                except (QMPError, subprocess.TimeoutExpired):
+                    pass
+            self.qmp.close()
+        if self.process.poll() is None:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait(timeout=10)
+        if self.console:
+            self.console.close()
+        self.stderr.close()
+
+
+LIVE_CHECK = r'''set -eu
+[ "$(id -un)" = hydra-live ]
+test -e /sys/firmware/efi
+test -f /etc/hydra/live-mode/active
+findmnt -n -o FSTYPE / | grep -qx overlay
+test -f /run/initramfs/live/LiveOS/squashfs.img
+sudo -n true
+for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
+for service in NetworkManager hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
+'''
+
+INSTALLED_CHECK = r'''set -eu
+assert_no_match() {
+    if grep "$@"; then
+        printf 'Forbidden live configuration remains\n' >&2
+        exit 1
+    else
+        status=$?
+        [ "$status" -eq 1 ] || exit "$status"
+    fi
+}
+[ "$(findmnt -n -o FSTYPE /)" = ext4 ]
+[ "$(readlink -f "$(findmnt -n -o SOURCE /)")" = /dev/vda2 ]
+assert_no_match -Eq 'root=live:|rd\.live' /proc/cmdline
+test ! -e /etc/hydra/live-mode
+test ! -L /etc/hydra/live-mode
+if getent passwd hydra-live; then
+    printf 'Live account remains\n' >&2
+    exit 1
+else
+    status=$?
+    [ "$status" -eq 2 ] || exit "$status"
+fi
+remaining=$(find /etc/systemd/system /etc/sudoers.d \( -name 90-hydra-live.conf -o -name 90-hydra-live \) -print)
+[ -z "$remaining" ]
+test ! -e /etc/greetd/live.toml
+test ! -L /etc/greetd/live.toml
+for directory in /etc/systemd/system/getty@*.service.d /etc/systemd/system/serial-getty@*.service.d /etc/sudoers.d; do
+    if [ -d "$directory" ]; then
+        assert_no_match -R -E -- '--autologin|NOPASSWD' "$directory"
+    elif [ -e "$directory" ] || [ -L "$directory" ]; then
+        printf 'Unexpected non-directory configuration path\n' >&2
+        exit 1
+    fi
+done
+getent shadow root | cut -d: -f2 | grep -Eq '^[!*]'
+getent shadow hydratest | cut -d: -f2 | grep -Eq '^\$'
+for service in NetworkManager greetd hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
+[ "$(systemctl show hydra-bridge-permissions -p Result --value)" = success ]
+[ "$(systemctl show hydra-bridge-permissions -p ExecMainStatus --value)" = 0 ]
+ip -4 -o addr show hydrabr0 | grep -F '10.203.0.1/24'
+ip -4 route show default | grep '^default '
+nft list table ip hydra_nat
+nmcli -t -f STATE general | grep -E '^connected'
+getent ahostsv4 gentoo.org
+for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
+hydra --help
+qemu-system-x86_64 --version
+qemu-img --version
+test -r /etc/greetd/config.toml
+find /usr/share -name OVMF_CODE.fd -print -quit | grep .
+'''
+
+NETWORK_CHECK = r'''set -eu
+[ "$(cat /proc/sys/net/ipv4/ip_forward)" = 1 ]
+python3 - <<'NETWORK'
+import pathlib, secrets, subprocess, time
+tag = secrets.token_hex(3)
+lan, guest = 'hl' + tag, 'hv' + tag
+uplink, lanpeer = 'hu' + tag, 'hp' + tag
+bridgepeer, guestpeer = 'hb' + tag, 'hg' + tag
+created = []
+links = []
+server = None
+def run(*args):
+    return subprocess.run(args, check=True, capture_output=True, text=True, timeout=20).stdout
+try:
+    for name in (lan, guest):
+        run('ip', 'netns', 'add', name)
+        created.append(name)
+    run('ip', 'link', 'add', uplink, 'type', 'veth', 'peer', 'name', lanpeer)
+    links.append(uplink)
+    run('ip', 'link', 'set', lanpeer, 'netns', lan)
+    run('ip', 'addr', 'add', '198.18.0.1/30', 'dev', uplink)
+    run('ip', 'link', 'set', uplink, 'up')
+    run('ip', '-n', lan, 'addr', 'add', '198.18.0.2/30', 'dev', lanpeer)
+    run('ip', '-n', lan, 'link', 'set', lanpeer, 'up')
+    run('ip', '-n', lan, 'link', 'set', 'lo', 'up')
+    run('ip', '-n', lan, 'route', 'add', '10.203.0.0/24', 'via', '198.18.0.1')
+    run('ip', 'link', 'add', bridgepeer, 'type', 'veth', 'peer', 'name', guestpeer)
+    links.append(bridgepeer)
+    run('ip', 'link', 'set', guestpeer, 'netns', guest)
+    run('ip', 'link', 'set', bridgepeer, 'master', 'hydrabr0')
+    run('ip', 'link', 'set', bridgepeer, 'up')
+    run('ip', '-n', guest, 'addr', 'add', '10.203.0.250/24', 'dev', guestpeer)
+    run('ip', '-n', guest, 'link', 'set', guestpeer, 'up')
+    run('ip', '-n', guest, 'link', 'set', 'lo', 'up')
+    run('ip', '-n', guest, 'route', 'add', 'default', 'via', '10.203.0.1')
+    port = 43117
+    server_code = "import socket; s=socket.socket(); s.settimeout(15); s.bind(('10.203.0.250',43117)); s.listen(1); c,a=s.accept(); c.sendall(b'HYDRA_LAN_TO_VM_OK'); c.close(); s.close()"
+    server = subprocess.Popen(['ip', 'netns', 'exec', guest, 'python3', '-c', server_code], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    client_code = "import socket,time; deadline=time.monotonic()+10\nwhile True:\n try:\n  s=socket.create_connection(('10.203.0.250',43117),1); break\n except OSError:\n  if time.monotonic()>deadline: raise\n  time.sleep(.1)\nexpected=b'HYDRA_LAN_TO_VM_OK'; received=b''\nwhile len(received)<len(expected):\n remaining=deadline-time.monotonic()\n if remaining<=0: raise TimeoutError('TCP marker deadline exceeded')\n s.settimeout(remaining)\n part=s.recv(len(expected)-len(received))\n if not part: raise RuntimeError('Truncated TCP marker')\n received+=part\nassert received==expected; s.close(); print('PASS: routed new LAN-to-VM TCP connection on arbitrary port 43117')"
+    print(run('ip', 'netns', 'exec', lan, 'python3', '-c', client_code), end='')
+    assert server.wait(timeout=10) == 0
+finally:
+    if server is not None and server.poll() is None:
+        server.terminate()
+        try: server.wait(timeout=5)
+        except subprocess.TimeoutExpired: server.kill(); server.wait(timeout=5)
+    for link in reversed(links):
+        subprocess.run(['ip', 'link', 'del', link], check=False, capture_output=True, timeout=10)
+    for name in reversed(created):
+        subprocess.run(['ip', 'netns', 'del', name], check=False, capture_output=True, timeout=10)
+NETWORK
+'''
+
+# A nested VM hardware smoke proves KVM initialisation/QMP, not guest OS boot.
+NESTED_CHECK = r'''set -eu
+if ! test -c /dev/kvm; then printf 'SKIP: guest /dev/kvm unavailable\n'; exit 0; fi
+python3 - <<'NESTED'
+import json, pathlib, socket, subprocess, tempfile, time
+with tempfile.TemporaryDirectory(prefix='hydra-nested-') as tmp:
+    root = pathlib.Path(tmp)
+    disk = root / 'fresh.qcow2'
+    subprocess.run(['qemu-img', 'create', '-f', 'qcow2', str(disk), '64M'], check=True, timeout=30)
+    log = open(root / 'qemu.log', 'wb')
+    proc = subprocess.Popen(['qemu-system-x86_64', '-machine', 'q35,accel=kvm', '-cpu', 'host', '-m', '128', '-S', '-display', 'none', '-nodefaults', '-nic', 'bridge,br=hydrabr0,model=virtio-net-pci', '-drive', 'if=virtio,format=qcow2,file=' + str(disk), '-qmp', 'unix:' + str(root / 'qmp') + ',server=on,wait=off'], stdin=subprocess.DEVNULL, stdout=log, stderr=log)
+    sock = socket.socket(socket.AF_UNIX)
+    sock.settimeout(10)
+    try:
+        deadline = time.monotonic() + 15
+        while True:
+            try: sock.connect(str(root / 'qmp')); break
+            except OSError:
+                if proc.poll() is not None or time.monotonic() >= deadline: raise RuntimeError('Nested KVM start failed')
+                time.sleep(.1)
+        stream = sock.makefile('rwb', buffering=0)
+        assert 'QMP' in json.loads(stream.readline())
+        for request in ({'execute':'qmp_capabilities'}, {'execute':'query-kvm'}, {'execute':'quit'}):
+            stream.write(json.dumps(request).encode() + b'\n')
+            while True:
+                response = json.loads(stream.readline())
+                if 'return' in response or 'error' in response: break
+            assert 'error' not in response, response
+            if request['execute'] == 'query-kvm':
+                assert response['return']['enabled'], response
+                print('PASS: normal-user nested KVM VM and bridge helper; fresh qcow2; private QMP')
+        assert proc.wait(timeout=10) == 0
+    finally:
+        sock.close()
+        if proc.poll() is None:
+            proc.terminate()
+            try: proc.wait(timeout=5)
+            except subprocess.TimeoutExpired: proc.kill(); proc.wait(timeout=5)
+        log.close()
+NESTED
+'''
+
+
+def digest(path):
+    result = hashlib.sha256()
+    with open(path, 'rb') as source:
+        for block in iter(lambda: source.read(1024 * 1024), b''):
+            result.update(block)
+    return result.hexdigest()
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--iso', required=True, type=Path)
+    parser.add_argument('--ovmf-code', required=True, type=Path)
+    parser.add_argument('--ovmf-vars', required=True, type=Path)
+    parser.add_argument('--qemu', default='qemu-system-x86_64')
+    parser.add_argument('--qemu-img', default='qemu-img')
+    parser.add_argument('--scratch-parent', type=Path, default=Path(tempfile.gettempdir()))
+    parser.add_argument('--memory', type=int, default=4096)
+    parser.add_argument('--boot-timeout', type=int, default=360)
+    parser.add_argument('--install-timeout', type=int, default=1800)
+    args = parser.parse_args(argv)
+    for path in (args.iso, args.ovmf_code, args.ovmf_vars):
+        if not path.is_file():
+            parser.error(f'Missing input file: {path}')
+    if args.memory < 2048 or args.boot_timeout <= 0 or args.install_timeout <= 0:
+        parser.error('Memory must be >=2048 MiB and timeouts positive')
+    args.iso = args.iso.resolve()
+    args.ovmf_code = args.ovmf_code.resolve()
+    args.ovmf_vars = args.ovmf_vars.resolve()
+    scratch = Path(tempfile.mkdtemp(prefix='hydra-iso-smoke-', dir=args.scratch_parent))
+    if len(str(scratch / 'serial.sock').encode()) >= 104:
+        parser.error('Scratch path too long for Unix sockets; use /tmp')
+    print(f'Acceptance artifacts: {scratch}', flush=True)
+    report = {'schema': 1, 'status': 'running', 'scratch': str(scratch),
+              'iso': str(args.iso), 'iso_sha256': digest(args.iso),
+              'checks': {}, 'physical_hardware': 'pending',
+              'nested_scope': 'Normal-user KVM VM launch, bridge helper and QMP; no nested guest OS boot',
+              'ovmf_code_sha256': digest(args.ovmf_code),
+              'ovmf_vars_input_sha256': digest(args.ovmf_vars)}
+    password = secrets.token_urlsafe(24)
+    sentinel = secrets.token_hex(32)
+    vm = None
+    def checkpoint(name, output):
+        report['checks'][name] = {'status': 'skipped' if output.startswith(b'SKIP:') else 'passed',
+                                  'output': output.decode(errors='replace')}
+        (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
+    try:
+        disk = scratch / 'fresh-install.qcow2'
+        reserve_disk(disk)
+        subprocess.run([args.qemu_img, 'create', '-f', 'qcow2', str(disk), '32G'], check=True, timeout=30,
+                       stdout=subprocess.DEVNULL)
+        shutil.copyfile(args.ovmf_vars, scratch / 'OVMF_VARS.fd')
+        os.chmod(scratch / 'OVMF_VARS.fd', 0o600)
+        vm = VM(args, scratch, disk, args.iso, password, 'live')
+        checkpoint('live_qmp_kvm', json.dumps(vm.kvm).encode())
+        console = vm.console
+        console.expect(rb'hydra-live[^\r\n]*[#$] ', args.boot_timeout)
+        checkpoint('live_boot', console.command(LIVE_CHECK, 90))
+        console.buffer = b''
+        installer_token = secrets.token_hex(12)
+        console.send(installer_command(installer_token) + '\n')
+        console.expect_installer(rb'Installed account password: ', 120, installer_token)
+        console.send(password + '\n')
+        console.expect_installer(rb'Repeat password: ', 30, installer_token)
+        console.send(password + '\n')
+        console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                 args.install_timeout, installer_token)
+        console.finish_installer(installer_token)
+        checkpoint('real_installer', b'Installer completed via guest controlling TTY')
+        console.command("sudo -n mkdir -p /mnt/hydra-validation && sudo -n mount /dev/vda2 /mnt/hydra-validation && printf %s " +
+                        shlex.quote(sentinel) + " | sudo -n tee /mnt/hydra-validation/var/lib/hydra-acceptance-sentinel >/dev/null && sudo -n umount /mnt/hydra-validation", 60)
+        console.send('sudo -n poweroff\n')
+        vm.wait_poweroff(90)
+        vm.close()
+        vm = None
+        checkpoint('live_poweroff', b'QEMU exited before ISO-free reboot')
+        vm = VM(args, scratch, disk, None, password, 'installed')
+        checkpoint('installed_qmp_kvm', json.dumps(vm.kvm).encode())
+        console = vm.console
+        console.expect(rb'login: ', args.boot_timeout)
+        console.send('hydratest\n')
+        console.expect(rb'Password: ', 30)
+        console.send(password + '\n')
+        console.expect(rb'hydratest[^\r\n]*[#$] ', 60)
+        output = console.command('id -un; if sudo -n true 2>/dev/null; then exit 1; fi', 30)
+        if output.strip() != b'hydratest':
+            raise SmokeError('Installed account identity mismatch')
+        checkpoint('installed_password_login', output)
+        console.buffer = b''
+        prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
+        console.send(f'sudo -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)
+        observed = console.command('cat /var/lib/hydra-acceptance-sentinel', 30)
+        if observed != sentinel.encode():
+            raise SmokeError('Persistence sentinel did not survive installed disk boot')
+        checkpoint('disk_persistence', b'Sentinel content verified from command output frame')
+        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')
+        report['status'] = 'passed'
+        return 0
+    except (SmokeError, QMPError, OSError, subprocess.SubprocessError) as exc:
+        report['status'] = 'failed'
+        report['error'] = str(exc).replace(password, '[REDACTED]')
+        print(f'Acceptance failed: {report["error"]}', file=sys.stderr)
+        return 1
+    finally:
+        if vm:
+            vm.close()
+        report['completed_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
+        (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
+        print(f'Report: {scratch / "report.json"}', flush=True)
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

docs/install.md

@@ -0,0 +1,85 @@
+# Install HydraOS
+
+HydraOS targets x86_64 UEFI systems. Disable Secure Boot for this first image.
+Boot the ISO and open the recovery terminal. The live session uses a locked
+`hydra-live` account with local sudo access; changes in the live filesystem are
+volatile. The image provides no default SSH login or universal installed password.
+
+`hydra-install` lists disks without selecting one. Inspect the devices, then ask
+for a plan using the whole disk you intend to erase:
+
+```sh
+hydra-install
+hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER
+```
+
+Replace the placeholders with an actual whole disk and your desired account
+name. The installer rejects mounted devices, the running root, the live medium,
+and their backing disks. These checks help prevent mistakes; verify the chosen
+disk and preserve backups before executing the plan. Physical disk installation
+requires the operator's explicit choice.
+
+To execute, enter the identical resolved whole-disk path in the confirmation:
+
+```sh
+sudo hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER \
+  --execute --confirm-erase /dev/CHOSEN_DISK
+```
+
+Execution erases the disk, creates a 1 GiB FAT32 EFI partition and an ext4 root,
+extracts the prepared payload, and prompts twice on the local terminal for your
+installed account password. Passwords have no command-line or configuration
+option. The installed root account stays locked. Defaults are Europe/Moscow and
+English UTF-8, with Russian UTF-8 also available; `--timezone` accepts an IANA
+zone present in the payload.
+
+The installer removes the live account, local password-free sudo rule, and
+live autologin overrides, then builds a normal disk initramfs and fallback UEFI
+loader. Wait for `Installation completed`, power off, remove the ISO, and boot
+the installed disk. Log in with the account and password you supplied. On
+failure, read the printed live diagnostic path; when a target root was mounted,
+a nonsecret snapshot is also saved at `/var/log/hydra-install.log` on that disk.
+
+## Disposable VM acceptance
+
+Run the repository harness with QEMU/KVM, `qemu-img`, Python 3, and an OVMF
+CODE/VARS pair. Use trusted image and firmware files. For example:
+
+```sh
+python3 dev/iso-smoke.py --iso /absolute/path/to/hydra-os.iso \
+  --ovmf-code /absolute/path/to/OVMF_CODE.fd \
+  --ovmf-vars /absolute/path/to/OVMF_VARS.fd
+```
+
+The harness creates a private directory under `/tmp`, a new 32 GiB sparse
+qcow2 disk, and a private copy of OVMF variables. There is no existing disk
+argument. QEMU attaches the ISO read-only; serial and QMP use private Unix
+sockets. Tool overrides `--qemu` and `--qemu-img` support temporary development
+environments. `--scratch-parent`, `--memory`, `--boot-timeout`, and
+`--install-timeout` control retained artifacts and bounded resource use.
+
+The test boots the live image, runs the real installer through the guest's
+controlling terminal using a generated password, writes a persistence sentinel,
+shuts down, and launches a second QEMU with the ISO absent. It verifies password
+login, disk root, removed live overrides, services, bridge/NAT configuration,
+DNS, recovery tools, and a routed new TCP connection from a synthetic LAN to a
+synthetic VM endpoint. Network namespaces and veth interfaces exist only inside
+the disposable HydraOS guest. When guest KVM is available, it launches a further
+fresh-disk QEMU as the installed account to exercise KVM, the bridge helper,
+and private QMP; this checks VM launch, not a nested operating system boot.
+
+`report.json` records each observed check, ISO/firmware SHA256, failures, and
+skipped nested virtualization. Serial/QEMU logs and the qcow2 remain in the
+printed scratch directory for review; passwords are redacted even across serial
+read boundaries. Cleanup controls only the QEMU process created by this run.
+A passing unit suite establishes harness behavior, not ISO acceptance. An
+actual successful report from the built ISO is required for boot/install/reboot
+acceptance.
+
+## Checks requiring the laptop
+
+Disposable QEMU results cannot establish ASUS FX517ZC hardware support. Before
+using HydraOS daily, boot the live ISO and observe Wi-Fi, touchpad, audio,
+brightness, suspend/resume, USB device selection, input capture, and actual
+display refresh. Physical installation and these checks remain pending until
+explicitly performed and recorded.

tests/test_iso_smoke.py

@@ -0,0 +1,634 @@
+"""Regression tests for disposable-VM isolation and trustworthy serial evidence."""
+import importlib.util
+from pathlib import Path
+import socket
+import select
+import subprocess
+import sys
+import time
+import tempfile
+import threading
+import unittest
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'dev/iso-smoke.py'
+
+
+class SmokeTests(unittest.TestCase):
+    def setUp(self):
+        self.assertTrue(SCRIPT.is_file(), 'disposable ISO acceptance harness is missing')
+        spec = importlib.util.spec_from_file_location('iso_smoke', SCRIPT)
+        self.mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.mod)
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.code = self.root / 'code.fd'
+        self.variables = self.root / 'vars.fd'
+        self.code.write_bytes(b'firmware')
+        self.variables.write_bytes(b'fresh vars')
+
+    def test_existing_disk_or_symlink_cannot_be_a_test_drive(self):
+        disk = self.root / 'disk.qcow2'
+        disk.write_bytes(b'original')
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.reserve_disk(disk)
+        link = self.root / 'link.qcow2'
+        link.symlink_to(disk)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.reserve_disk(link)
+        self.assertEqual(disk.read_bytes(), b'original')
+
+    def test_qemu_only_accepts_owned_regular_disk_and_readonly_iso(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        iso = self.root / 'trusted.iso'
+        iso.write_bytes(b'ISO')
+        argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                                  self.variables, iso, 2048)
+        drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
+        self.assertTrue(any('format=qcow2' in item and str(disk) in item for item in drives))
+        self.assertTrue(any('media=cdrom' in item and 'readonly=on' in item for item in drives))
+        self.assertIn('accel=kvm', ' '.join(argv))
+        disk.unlink()
+        disk.symlink_to(iso)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, iso, 2048)
+
+    def test_installed_reboot_has_no_optical_medium(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                                  self.variables, None, 2048)
+        self.assertNotIn('media=cdrom', ' '.join(argv))
+        self.assertNotIn('-cdrom', argv)
+
+    def test_hardlinked_disk_is_refused(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        (self.root / 'alias').hardlink_to(disk)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, None, 2048)
+
+    def test_result_frame_ignores_command_echo_and_preserves_failure(self):
+        command = self.mod.frame_command('printf sentinel; false', 'testtoken')
+        result = subprocess.run(['/bin/sh', '-c', command], capture_output=True)
+        echo = command.encode() + b'\r\n'
+        self.assertIsNone(self.mod.parse_frame(echo, 'testtoken'))
+        self.assertEqual(self.mod.parse_frame(echo + result.stdout, 'testtoken'),
+                         (1, b'sentinel'))
+
+    def test_console_redacts_password_even_when_split_across_reads(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        log = self.root / 'serial.log'
+        console = self.mod.Console(one, log, ['split-secret'])
+        self.addCleanup(console.close)
+        two.sendall(b'hello split-')
+        console.read(0.1)
+        two.sendall(b'secret goodbye')
+        console.read(0.1)
+        console.close()
+        self.assertNotIn(b'split-secret', log.read_bytes())
+        self.assertIn(b'[REDACTED]', log.read_bytes())
+
+    def test_console_timeout_is_bounded_and_partial_frame_not_success(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'serial.log', [])
+        self.addCleanup(console.close)
+        two.sendall(b'\\036token:END:0\\037')
+        with self.assertRaises(self.mod.SmokeError):
+            console.command('true', timeout=0.05)
+
+    def installer_console(self):
+        one, two = socket.socketpair()
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'installer.log', ['test-secret'])
+        self.addCleanup(console.close)
+        return console, two
+
+    def test_captured_installer_refusal_fails_before_password_timeout(self):
+        console, guest = self.installer_console()
+        guest.sendall(b'\x1b]3008;type=command\x1b\\hydra-install: Command failed (1): '
+                      b'findmnt --json --target /LiveOS/squashfs.img '
+                      b'--output SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS: \r\n')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*findmnt'):
+            console.expect_installer(rb'Installed account password: ', 0.05, 'testtoken')
+
+    def test_installer_shell_return_without_completion_is_failure_even_zero(self):
+        for status in (0, 1, 127):
+            with self.subTest(status=status):
+                console, guest = self.installer_console()
+                guest.sendall(b'\x1etesttoken:INSTALL_EXIT:' + str(status).encode() + b'\x1f')
+                with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*INSTALL_EXIT'):
+                    console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                             0.05, 'testtoken')
+
+    def test_installer_echo_cannot_fake_exit_and_complete_prompt_still_required(self):
+        console, guest = self.installer_console()
+        guest.sendall(self.mod.installer_command('testtoken').encode() + b'\r\n'
+                      b'Installed account password: ')
+        self.assertEqual(console.expect_installer(rb'Installed account password: ',
+                         0.05, 'testtoken').group(), b'Installed account password: ')
+        guest.sendall(b'Installation completed. Diagnostic')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Timed out'):
+            console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                     0.05, 'testtoken')
+        guest.sendall(b' log: /var/log/hydra-install.log\r\n')
+        self.assertIn(b'/var/log/hydra-install.log', console.expect_installer(
+            rb'Installation completed\. Diagnostic log: [^\r\n]+', 0.05, 'testtoken').group())
+
+    def test_installer_wrapper_marks_actual_shell_return_and_redacts_error(self):
+        console, guest = self.installer_console()
+        fake_sudo = self.root / 'sudo'
+        fake_sudo.write_text('#!/bin/sh\nprintf "hydra-install: test-secret refused\\n"\nexit 1\n')
+        fake_sudo.chmod(0o700)
+        result = subprocess.run(['/bin/sh', '-c', self.mod.installer_command('testtoken')],
+                                env={'PATH': str(self.root)}, capture_output=True, check=True)
+        self.assertIn(b'\x1etesttoken:INSTALL_EXIT:1\x1f', result.stdout)
+        guest.sendall(result.stdout)
+        with self.assertRaisesRegex(self.mod.SmokeError, r'Installer.*\[REDACTED\] refused') as caught:
+            console.expect_installer(rb'Repeat password: ', 0.05, 'testtoken')
+        self.assertNotIn('test-secret', str(caught.exception))
+
+    def test_installer_followup_waits_for_exit_and_new_shell_prompt(self):
+        console, guest = self.installer_console()
+        premature = []
+        def respond():
+            guest.sendall(b'Installation completed. Diagnostic log: /run/installer.log\r\n')
+            premature.append(bool(select.select([guest], [], [], 0.05)[0]))
+            guest.sendall(b'\x1etesttoken:INSTALL_EXIT:0\x1f')
+            premature.append(bool(select.select([guest], [], [], 0.05)[0]))
+            guest.sendall(b'\x1b[?2004hhydra-live@hydra-os ~ $ ')
+            data = b''
+            while not data.endswith(b'\n'):
+                data += guest.recv(65536)
+            result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
+            guest.sendall(result.stdout)
+        worker = threading.Thread(target=respond, daemon=True)
+        worker.start()
+        console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
+                                 1, 'testtoken')
+        console.finish_installer('testtoken', timeout=1)
+        self.assertEqual(console.command('printf ready', timeout=1), b'ready')
+        worker.join(timeout=1)
+        self.assertEqual(premature, [False, False], 'command reached installer before shell was ready')
+
+    def test_installer_completion_does_not_override_nonzero_return(self):
+        console, guest = self.installer_console()
+        guest.sendall(b'\x1etesttoken:INSTALL_EXIT:1\x1fhydra-live@hydra-os ~ $ ')
+        with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*status 1'):
+            console.finish_installer('testtoken', timeout=0.05)
+
+    def shutdown_vm(self, program):
+        one, two = socket.socketpair()
+        vm = self.mod.VM.__new__(self.mod.VM)
+        vm.console = self.mod.Console(one, self.root / 'shutdown.log', ['split-secret'])
+        self.addCleanup(vm.console.close)
+        code = ('import socket,sys,time; s=socket.socket(fileno=int(sys.argv[1])); ' + program)
+        vm.process = subprocess.Popen([sys.executable, '-c', code, str(two.fileno())],
+                                      pass_fds=(two.fileno(),), stdin=subprocess.DEVNULL)
+        two.close()
+        def reap():
+            if vm.process.poll() is None:
+                vm.process.terminate()
+            vm.process.wait(timeout=2)
+        self.addCleanup(reap)
+        return vm
+
+    def test_poweroff_drains_and_redacts_serial_until_owned_process_exits(self):
+        vm = self.shutdown_vm("s.sendall(b'Stopping split-'); time.sleep(.02); "
+                              "s.sendall(b'secret service\\nPower down\\n'); s.close()")
+        vm.wait_poweroff(1)
+        output = (self.root / 'shutdown.log').read_bytes()
+        self.assertIn(b'Stopping [REDACTED] service\nPower down\n', output)
+        self.assertNotIn(b'split-secret', output)
+
+    def test_poweroff_preserves_failure_cause_and_original_timeout(self):
+        for program, error in (("s.sendall(b'Stop job stuck\\n'); s.close(); time.sleep(1)",
+                                'Guest failed to power off in time'),
+                               ("s.sendall(b'QEMU shutdown failed\\n'); s.close(); sys.exit(7)",
+                                'Test QEMU failed during poweroff')):
+            with self.subTest(program=program):
+                vm = self.shutdown_vm(program)
+                started = time.monotonic()
+                with self.assertRaisesRegex(self.mod.SmokeError, error):
+                    vm.wait_poweroff(.15)
+                self.assertLess(time.monotonic() - started, .5)
+                output = (self.root / 'shutdown.log').read_bytes()
+                self.assertIn(b'Stop job stuck' if 'time.sleep' in program else b'QEMU shutdown failed', output)
+                if 'time.sleep' in program:
+                    self.assertIsNone(vm.process.poll(), 'timeout must not kill or falsely pass the owned process')
+
+    def test_frame_real_shell_roundtrip_does_not_accept_sentinel_in_echo(self):
+        one, two = socket.socketpair()
+        self.addCleanup(one.close)
+        self.addCleanup(two.close)
+        console = self.mod.Console(one, self.root / 'serial.log', [])
+        self.addCleanup(console.close)
+        def guest():
+            data = b''
+            while not data.endswith(b'\n'):
+                data += two.recv(65536)
+            two.sendall(data)
+            result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
+            two.sendall(result.stdout)
+        thread = threading.Thread(target=guest)
+        thread.start()
+        self.assertEqual(console.command("printf 'persisted-data'", timeout=2), b'persisted-data')
+        thread.join(timeout=2)
+
+    def test_firmware_vars_must_be_private_copy_in_scratch(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        self.variables.unlink()
+        self.variables.symlink_to(self.code)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, disk, self.code,
+                               self.variables, None, 2048)
+
+    def test_drive_outside_scratch_and_ambiguous_qemu_paths_are_refused(self):
+        disk = self.root / 'disk.qcow2'
+        self.mod.reserve_disk(disk)
+        other = self.root / 'other'
+        other.mkdir()
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', other, disk, self.code,
+                               self.variables, None, 2048)
+        comma = self.root / 'disk,cache=unsafe.qcow2'
+        self.mod.reserve_disk(comma)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.qemu_argv('qemu', self.root, comma, self.code,
+                               self.variables, None, 2048)
+
+    def probe_fixture(self, case, live=False):
+        # Run the production shell against ordinary scratch files. Commands that
+        # query guest state are boundary fixtures; grep/find/test are real tools.
+        import os
+        root = self.root / 'guest'
+        for directory in ('etc/hydra', 'etc/greetd', 'etc/sudoers.d',
+                          'etc/systemd/system/getty@tty2.service.d',
+                          'etc/systemd/system/serial-getty@ttyS0.service.d',
+                          'usr/share/firmware', 'proc', 'sys/firmware/efi',
+                          'run/initramfs/live/LiveOS'):
+            (root / directory).mkdir(parents=True, exist_ok=True)
+        (root / 'etc/greetd/config.toml').write_text('normal authentication')
+        (root / 'usr/share/firmware/OVMF_CODE.fd').write_text('firmware')
+        (root / 'proc/cmdline').write_text('root=UUID=disk ro' if case != 'live-cmdline' else 'root=live:CDLABEL=HYDRA_OS rd.live.image')
+        if live or case == 'live-marker':
+            (root / 'etc/hydra/live-mode').mkdir(exist_ok=True)
+            (root / 'etc/hydra/live-mode/active').touch()
+        if live:
+            (root / 'run/initramfs/live/LiveOS/squashfs.img').touch()
+        files = {
+            'live-dropin': ('etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf', 'live'),
+            'live-sudoers': ('etc/sudoers.d/90-hydra-live', 'live'),
+            'autologin': ('etc/systemd/system/serial-getty@ttyS0.service.d/other.conf', 'ExecStart=agetty --autologin hydratest'),
+            'nopasswd': ('etc/sudoers.d/other', 'hydratest ALL=(ALL) NOPASSWD: ALL'),
+            'live-greetd': ('etc/greetd/live.toml', 'live'),
+        }
+        if case in files:
+            path, text = files[case]
+            (root / path).write_text(text)
+        if case == 'config-read-error':
+            unreadable = root / 'etc/sudoers.d/unreadable'
+            unreadable.write_text('configuration')
+            unreadable.chmod(0)
+        if case == 'cmdline-read-error':
+            (root / 'proc/cmdline').unlink()
+        commands = self.root / 'fixture-bin'
+        commands.mkdir(exist_ok=True)
+        dispatcher = commands / 'fixture'
+        dispatcher.write_text('''#!/usr/bin/env python3
+import os, pathlib, sys
+name = pathlib.Path(sys.argv[0]).name
+args = sys.argv[1:]
+case = os.environ['SMOKE_CASE']
+if name == 'id': print('hydra-live')
+elif name == 'findmnt': print('/dev/vda2' if 'SOURCE' in args else ('overlay' if os.environ['SMOKE_LIVE']=='1' else 'ext4'))
+elif name == 'systemctl':
+    if args[0] == 'is-active':
+        states = [unit != case for unit in args[1:]]
+        print('\\n'.join('active' if state else 'failed' for state in states))
+        sys.exit(0 if any(states) else 3)
+    print('success' if 'Result' in args else '0')
+elif name == 'getent':
+    if args[:2] == ['passwd','hydra-live']:
+        if case == 'account-lookup-error': sys.exit(3)
+        if case != 'live-account': sys.exit(2)
+        print('hydra-live:x:1000:1000::/home/hydra-live:/bin/bash')
+    elif args[0] == 'shadow': print(args[1] + ':' + ('!' if args[1]=='root' else '$6$hash') + ':1::::::')
+    else: print('127.0.0.1 STREAM example')
+elif name == 'ip': print('10.203.0.1/24' if 'addr' in args else 'default via 10.0.2.2')
+elif name == 'nmcli': print('connected')
+else: print('fixture tool')
+''')
+        dispatcher.chmod(0o755)
+        for command in ('id', 'findmnt', 'systemctl', 'getent', 'ip', 'nft',
+                        'nmcli', 'sudo', 'hydra', 'hydra-install',
+                        'qemu-system-x86_64', 'qemu-img', 'remote-viewer', 'foot', 'labwc'):
+            path = commands / command
+            if not path.exists():
+                path.symlink_to(dispatcher)
+        probe = self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK
+        for path in ('/etc/', '/usr/share', '/proc/cmdline', '/sys/firmware/efi', '/run/initramfs/live/LiveOS/squashfs.img'):
+            probe = probe.replace(path, str(root) + path)
+        import shutil
+        for command in ('python3', 'sh', 'grep', 'find', 'cut', 'readlink', 'base64'):
+            target = shutil.which(command)
+            self.assertIsNotNone(target)
+            (commands / command).symlink_to(target) if not (commands / command).exists() else None
+        env = dict(os.environ, PATH=str(commands),
+                   SMOKE_CASE=case, SMOKE_LIVE='1' if live else '0')
+        result = subprocess.run(['/bin/sh', '-c', self.mod.frame_command(probe, 'probecheck')],
+                                env=env, capture_output=True, timeout=10)
+        parsed = self.mod.parse_frame(result.stdout, 'probecheck')
+        self.assertIsNotNone(parsed, result.stderr)
+        return parsed[0]
+
+    def test_installed_probe_rejects_each_retained_live_artifact(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        for case in ('live-cmdline', 'live-account', 'live-dropin', 'live-sudoers',
+                     'autologin', 'nopasswd', 'live-marker', 'live-greetd'):
+            with self.subTest(case=case):
+                # Give each negative case an independent clean filesystem.
+                import shutil
+                shutil.rmtree(self.root / 'guest')
+                self.assertNotEqual(self.probe_fixture(case), 0)
+
+    def test_installed_probe_does_not_treat_account_query_error_as_absence(self):
+        self.assertNotEqual(self.probe_fixture('account-lookup-error'), 0)
+
+    def test_installed_probe_rejects_unreadable_configuration(self):
+        self.assertNotEqual(self.probe_fixture('config-read-error'), 0)
+
+    def test_installed_probe_rejects_unreadable_boot_arguments(self):
+        self.assertNotEqual(self.probe_fixture('cmdline-read-error'), 0)
+
+    def test_installed_probe_rejects_each_failed_required_service(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        for service in ('NetworkManager', 'greetd', 'hydra-network', 'hydra-dnsmasq'):
+            with self.subTest(service=service):
+                self.assertNotEqual(self.probe_fixture(service), 0)
+
+    def test_live_probe_rejects_each_failed_required_service(self):
+        self.assertEqual(self.probe_fixture('clean', live=True), 0)
+        for service in ('NetworkManager', 'hydra-network', 'hydra-dnsmasq'):
+            with self.subTest(service=service):
+                self.assertNotEqual(self.probe_fixture(service, live=True), 0)
+
+    def tcp_probe(self, fragments):
+        # Execute the exact Python client embedded in the guest network probe,
+        # redirecting only its address to a fresh loopback socket.
+        import ast
+        import time
+        source = self.mod.NETWORK_CHECK.split("<<'NETWORK'\n", 1)[1].rsplit('\nNETWORK\n', 1)[0]
+        assignments = (node for node in ast.walk(ast.parse(source)) if isinstance(node, ast.Assign))
+        value = next(node.value for node in assignments
+                     if any(isinstance(target, ast.Name) and target.id == 'client_code' for target in node.targets))
+        client = ast.parse(ast.literal_eval(value))
+        listener = socket.socket()
+        self.addCleanup(listener.close)
+        listener.bind(('127.0.0.1', 0))
+        listener.listen(1)
+        listener.settimeout(3)
+        host, port = listener.getsockname()
+        class Redirect(ast.NodeTransformer):
+            def visit_Tuple(self, node):
+                if isinstance(node.elts[0], ast.Constant) and node.elts[0].value == '10.203.0.250':
+                    return ast.Tuple(elts=[ast.Constant(host), ast.Constant(port)], ctx=ast.Load())
+                return self.generic_visit(node)
+        client = ast.fix_missing_locations(Redirect().visit(client))
+        def server():
+            connection, _ = listener.accept()
+            with connection:
+                for fragment in fragments:
+                    try:
+                        connection.sendall(fragment)
+                    except BrokenPipeError:
+                        break
+                    time.sleep(.05)
+        thread = threading.Thread(target=server)
+        thread.start()
+        result = subprocess.run(['python3', '-c', ast.unparse(client)], capture_output=True, timeout=5)
+        thread.join(timeout=3)
+        self.assertFalse(thread.is_alive())
+        return result
+
+    def test_tcp_probe_accepts_complete_marker_across_fragmented_reads(self):
+        result = self.tcp_probe([b'HYDRA_', b'LAN_TO_', b'VM_OK'])
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    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}}
+                        if request['execute'] == 'query-kvm' and 'on_snapshot' in context:
+                            context.pop('on_snapshot')()
+                        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(), '<guest-lifecycle-probe>', 'exec')
+
+    def session_fixture(self, records):
+        import json
+        import os
+        from unittest.mock import patch
+        source = self.root / 'session-records.json'
+        source.write_text(json.dumps(records))
+        commands = self.root / 'session-fixture-bin'
+        commands.mkdir()
+        command = commands / 'loginctl'
+        command.write_text('''#!/usr/bin/env python3
+import json, os, sys
+records = json.load(open(os.environ['SMOKE_SESSION_RECORDS']))
+args = sys.argv[1:]
+if args[0] == 'show-user':
+    print(' '.join(record['Id'] for record in records if str(record['User']) == args[1]))
+elif args[0] == 'show-session':
+    record = next((record for record in records if record['Id'] == args[1]), None)
+    if record is None: sys.exit(1)
+    for key, value in record.items(): print(key + '=' + str(value))
+else: sys.exit(1)
+''')
+        command.chmod(0o755)
+        environment = patch.dict(os.environ, {'PATH': str(commands) + ':' + os.environ['PATH'],
+                                              'SMOKE_SESSION_RECORDS': str(source)})
+        environment.start()
+        self.addCleanup(environment.stop)
+        return source
+
+    def session_record(self, ident='serial1', state='active', cls='user', leader=1234):
+        import os
+        return {'Id': ident, 'User': os.getuid(), 'State': state, 'Class': cls,
+                'Leader': leader, 'TTY': 'ttyS0' if cls == 'user' else ''}
+
+    def test_logout_sessions_accept_closing_scope_and_manager_with_surviving_vm(self):
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        self.session_fixture([dict(original, State='closing'),
+                              self.session_record('manager1', 'active', 'manager')])
+        self.assertTrue(self.mod.no_open_logins(context))
+        self.assertEqual(self.mod.capture_logout_snapshot(context), context['before'])
+
+    def test_logout_sessions_initial_login_ignores_manager_but_rejects_open_replacement(self):
+        import json
+        import os
+        original = self.session_record()
+        context = {'uid': os.getuid(), 'login_session': original}
+        source = self.session_fixture([original, self.session_record('manager1', 'active', 'manager')])
+        self.assertEqual(self.mod.require_initial_login(os.getuid(), 'serial1'), original)
+        for state in ('online', 'active'):
+            with self.subTest(state=state):
+                source.write_text(json.dumps([dict(original, State='closing'),
+                                             self.session_record('new-login', state)]))
+                self.assertFalse(self.mod.no_open_logins(context))
+        source.write_text(json.dumps([dict(original, State='closing', Leader=9999)]))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.no_open_logins(context)
+
+    def test_logout_sessions_reject_login_appearing_during_snapshot(self):
+        import json
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        source = self.session_fixture([dict(original, State='closing')])
+        context['on_snapshot'] = lambda: source.write_text(json.dumps([
+            dict(original, State='closing'), self.session_record('relogin')]))
+        self.assertTrue(self.mod.no_open_logins(context))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.capture_logout_snapshot(context)
+
+    def test_guest_tools_accept_gentoo_remote_viewer_without_virt_viewer(self):
+        self.assertEqual(self.probe_fixture('clean'), 0)
+        self.assertEqual(self.probe_fixture('clean', live=True), 0)