Probe installed user VM and audio continuity across logout

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')

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(), '<guest-lifecycle-probe>', 'exec')