Verify packaged terminal UX in installed ISO acceptance

dev/iso-smoke.py

@@ -308,6 +308,253 @@ def lifecycle_guest_script():
             '\n'.join(inspect.getsource(helper) for helper in helpers) + LIFECYCLE_BODY)
 
 
+def terminal_ux_iso():
+    """An empty ISO9660 filesystem, intentionally without an El Torito boot image."""
+    def both(value, width):
+        return value.to_bytes(width, 'little') + value.to_bytes(width, 'big')
+
+    def directory_record(identifier):
+        record = bytearray(34)
+        record[0] = 34
+        record[2:10] = both(20, 4)
+        record[10:18] = both(2048, 4)
+        record[18:25] = bytes((126, 10, 13, 0, 0, 0, 0))
+        record[25] = 2
+        record[28:32] = both(1, 2)
+        record[32] = 1
+        record[33] = identifier
+        return record
+
+    data = bytearray(21 * 2048)
+    primary = bytearray(2048)
+    primary[:7] = b'\x01CD001\x01'
+    primary[8:40] = b'HYDRA OS'.ljust(32, b' ')
+    primary[40:72] = b'HYDRA_UX_PROBE'.ljust(32, b' ')
+    primary[80:88] = both(21, 4)
+    primary[120:124] = both(1, 2)
+    primary[124:128] = both(1, 2)
+    primary[128:132] = both(2048, 2)
+    primary[132:140] = both(10, 4)
+    primary[140:144] = (18).to_bytes(4, 'little')
+    primary[148:152] = (19).to_bytes(4, 'big')
+    primary[156:190] = directory_record(0)
+    primary[190:813] = b' ' * 623
+    primary[813:830] = b'2026101300000000\x00'
+    primary[830:847] = b'2026101300000000\x00'
+    primary[847:881] = b'0000000000000000\x00' * 2
+    primary[881] = 1
+    data[16 * 2048:17 * 2048] = primary
+    data[17 * 2048:17 * 2048 + 7] = b'\xffCD001\x01'
+    for sector, order in ((18, 'little'), (19, 'big')):
+        data[sector * 2048:sector * 2048 + 10] = (
+            b'\x01\x00' + (20).to_bytes(4, order) + (1).to_bytes(2, order) + b'\x00\x00')
+    data[20 * 2048:20 * 2048 + 68] = directory_record(0) + directory_record(1)
+    return bytes(data)
+
+
+def verify_terminal_ux_branding(output, errors):
+    rendered = re.sub(r'\x1b\[[0-?]*[ -/]*[@-~]', '', output)
+    if errors.strip() or re.search(r'\b(error|failed|unknown module)\b', rendered, re.I):
+        raise SmokeError('Installed default Fastfetch emitted errors: ' + (errors or rendered)[-1500:])
+    if not re.search(r'\bOS\s*:\s*Hydra OS\b', rendered):
+        raise SmokeError('Installed default Fastfetch did not render the Hydra OS pretty name')
+    if 'H Y D R A  O S' not in rendered or rendered.count('(o >') != 3:
+        raise SmokeError('Installed default Fastfetch did not render the three-headed Hydra logo')
+    return rendered
+
+
+def terminal_ux_snapshot(context):
+    """Verify CLI-owned live process/QMP identity, excluding OS boot or audio claims."""
+    try:
+        if context['uid'] != os.getuid():
+            raise SmokeError('Terminal UX process is not owned by the installed normal user')
+        runtime = Path(context['runtime'])
+        run = runtime / context['name']
+        for directory in (runtime, run):
+            info = directory.lstat()
+            if (not stat.S_ISDIR(info.st_mode) or info.st_uid != context['uid']
+                    or stat.S_IMODE(info.st_mode) != 0o700):
+                raise SmokeError('Terminal UX runtime is not private and owned')
+        if json.loads((run / 'identity.json').read_text()) != {
+                'name': context['name'], 'uuid': context['uuid']}:
+            raise SmokeError('Terminal UX metadata belongs to a different VM')
+        pid = int((run / 'qemu.pid').read_text())
+        process = Path('/proc') / str(pid)
+        fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
+        if process.stat().st_uid != context['uid'] or fields[0] == 'Z':
+            raise SmokeError('Terminal UX VM process exited or changed owner')
+        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('Terminal UX PID belongs to a different VM')
+        lifecycle_socket(run / 'qmp.sock', connect_socket=False)
+        lifecycle_socket(run / 'spice.sock')
+        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('Terminal UX QMP belongs to a different VM')
+            if client.command('query-status').get('status') != 'running':
+                raise SmokeError('Terminal UX VM is not running')
+            if not client.command('query-kvm').get('enabled'):
+                raise SmokeError('Terminal UX VM did not initialise nested KVM')
+        return {'name': context['name'], 'uuid': context['uuid'], 'pid': pid,
+                'process_start': fields[19], 'kvm': True}
+    except (OSError, ValueError, KeyError, IndexError, QMPError) as exc:
+        raise SmokeError('Terminal UX cannot establish owned live identity: ' + str(exc)) from exc
+
+
+TERMINAL_UX_BODY = r'''
+if os.getuid() == 0 or subprocess.check_output(['id', '-un'], text=True).strip() != 'hydratest':
+    raise SmokeError('Terminal UX probe must run as the installed normal hydratest user')
+if not os.access('/dev/kvm', os.R_OK | os.W_OK):
+    raise SmokeError('Terminal UX needs installed-user nested KVM access')
+if not Path('/usr/bin/hydra').is_file():
+    raise SmokeError('Installed product launcher /usr/bin/hydra is missing')
+fetch = subprocess.run(['fastfetch', '--pipe', 'false', '--show-errors'],
+                       capture_output=True, text=True, timeout=30)
+if fetch.returncode:
+    raise SmokeError('Installed default Fastfetch failed: ' + fetch.stderr[-1500:])
+fastfetch_output = verify_terminal_ux_branding(fetch.stdout, fetch.stderr)
+config_hash = hashlib.sha256(Path('/etc/fastfetch/config.jsonc').read_bytes()).hexdigest()
+logo_hash = hashlib.sha256(Path('/usr/share/hydra/fastfetch.txt').read_bytes()).hexdigest()
+base = Path(tempfile.mkdtemp(prefix='hux-', dir='/tmp'))
+name = 'ux-' + secrets.token_hex(4)
+context = {'name': name, 'uid': os.getuid(), 'config': str(base / 'config'),
+           'state': str(base / 'state'), 'runtime': str(base / 'run')}
+context['uuid'] = str(uuid.uuid5(uuid.NAMESPACE_URL, str(base / 'state' / name / 'disk.qcow2')))
+environment = dict(os.environ, XDG_CACHE_HOME=str(base / 'cache'))
+environment.pop('HYDRA_SESSION_MANIFEST', None)
+command = ['/usr/bin/hydra', '--config', context['config'], '--state', context['state'],
+           '--runtime', context['runtime']]
+
+def cli(*arguments, input=None):
+    result = subprocess.run(command + list(arguments), input=input, env=environment,
+                            capture_output=True, text=True, timeout=60)
+    if result.returncode:
+        raise SmokeError('Installed hydra ' + arguments[0] + ' failed: ' +
+                         (result.stdout + result.stderr)[-2000:])
+    return result.stdout
+
+def cli_json(*arguments):
+    return json.loads(cli(*arguments))
+
+fixture = base / 'fixture.iso'
+fixture.write_bytes(terminal_ux_iso())
+iso_hash = hashlib.sha256(fixture.read_bytes()).hexdigest()
+requests = []
+class ISOHandler(http.server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        requests.append(self.path)
+        if self.path != '/fixture.iso':
+            self.send_error(404)
+            return
+        payload = fixture.read_bytes()
+        self.send_response(200)
+        self.send_header('Content-Type', 'application/x-iso9660-image')
+        self.send_header('Content-Length', str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+    def log_message(self, *args):
+        pass
+server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), ISOHandler)
+worker = threading.Thread(target=server.serve_forever, daemon=True)
+worker.start()
+url = 'http://127.0.0.1:' + str(server.server_port) + '/fixture.iso'
+result = None
+try:
+    if cli_json('list'):
+        raise SmokeError('Terminal UX private configuration was not initially empty')
+    empty_menu = cli('menu', input=':quit\n')
+    if ('Hydra OS' not in empty_menu or 'No VM profiles' not in empty_menu
+            or list(base.glob('run/*/qemu.pid')) or cli_json('list')):
+        raise SmokeError('Installed empty menu was unreadable or started a VM')
+    run_output = cli('run', url, '--name', name, '--memory', '512', '--cpus', '1',
+                     '--disk-size', '64M', '--sha256', iso_hash, '--no-view')
+    if requests != ['/fixture.iso']:
+        raise SmokeError('Installed hydra run did not download exactly its loopback ISO')
+    before = terminal_ux_snapshot(context)
+    profiles = cli_json('list')
+    if len(profiles) != 1 or profiles[0]['name'] != name:
+        raise SmokeError('Installed hydra run did not create exactly the requested private profile')
+    cached_iso = Path(profiles[0]['iso'])
+    cache = base / 'cache' / 'hydra' / 'iso'
+    info = cached_iso.lstat()
+    if (cached_iso.parent != cache or not stat.S_ISREG(info.st_mode)
+            or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600
+            or hashlib.sha256(cached_iso.read_bytes()).hexdigest() != iso_hash
+            or list(cache.glob('*.part'))):
+        raise SmokeError('Installed hydra run cache is unsafe, incomplete or has a different checksum')
+    profile = tomllib.loads((base / 'config' / (name + '.toml')).read_text())
+    if (profile['memory'], profile['cpus'], profile['disk_size']) != (512, 1, '64M'):
+        raise SmokeError('Installed hydra run lost its explicit allocations')
+    downloaded = cli('download', url, '--sha256', iso_hash)
+    if (Path(downloaded.splitlines()[-1]) != cached_iso or requests != ['/fixture.iso']
+            or hashlib.sha256(cached_iso.read_bytes()).hexdigest() != iso_hash):
+        raise SmokeError('Installed hydra download did not reuse the complete cached ISO')
+    cli('open', name, '--no-view')
+    after_open = terminal_ux_snapshot(context)
+    if before != after_open:
+        raise SmokeError('Installed hydra open restarted or replaced its running VM')
+    menu = cli('menu', input=':quit\n')
+    after_menu = terminal_ux_snapshot(context)
+    if ('Hydra OS' not in menu or name not in menu or 'running' not in menu
+            or '512 MiB' not in menu or '"status":' in menu or before != after_menu):
+        raise SmokeError('Installed running menu was unreadable or changed VM lifetime')
+    result = {'status': 'passed', 'uid': os.getuid(), 'normal_user': 'hydratest',
+              'launcher': '/usr/bin/hydra', 'fixture_bytes': fixture.stat().st_size,
+              'iso_sha256': iso_hash, 'cache_reused': True, 'http_requests': requests,
+              'before': before, 'after_open': after_open, 'after_menu': after_menu,
+              'memory_mib': profile['memory'], 'cpus': profile['cpus'],
+              'disk_size': profile['disk_size'], 'network': profile['network'],
+              'empty_menu': empty_menu, 'running_menu': menu, 'run_output': run_output,
+              'download_output': downloaded, 'fastfetch_command': ['fastfetch', '--pipe', 'false', '--show-errors'],
+              'fastfetch_default_config_sha256': config_hash, 'fastfetch_logo_sha256': logo_hash,
+              'fastfetch_output': fastfetch_output,
+              'scope': 'Packaged installed CLI, cache, branding and live nested QMP/KVM; nonboot ISO fixture, no nested OS boot or performance claim'}
+finally:
+    try:
+        status = cli_json('status', name).get('status')
+        if status == 'running':
+            # Never signal a saved PID: verify the exact private live identity
+            # and let the product CLI perform its own identity-checked QMP quit.
+            terminal_ux_snapshot(context)
+            if cli_json('stop', name, '--force').get('status') != 'stopped':
+                raise SmokeError('Owned terminal UX fixture VM did not stop')
+        elif status != 'stopped':
+            raise SmokeError('Refusing terminal UX cleanup of an unverified VM identity at ' + str(base))
+    finally:
+        server.shutdown()
+        server.server_close()
+        worker.join(timeout=5)
+if result is not None:
+    result['cleanup'] = 'Only the verified private fixture VM stopped through hydra stop --force'
+    print(json.dumps(result))
+    shutil.rmtree(base)
+'''
+
+
+def terminal_ux_guest_script():
+    helpers = (lifecycle_socket, terminal_ux_iso, verify_terminal_ux_branding, terminal_ux_snapshot)
+    return ('import hashlib, http.server, json, os, re, secrets, shutil, socket, stat, subprocess, sys, tempfile, threading, tomllib, 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) + TERMINAL_UX_BODY)
+
+
+def installed_terminal_ux_probe(console):
+    preparation = ("python3 - <<'HYDRA_TERMINAL_UX_SOURCE'\n" + terminal_ux_guest_script() +
+                   '\nHYDRA_TERMINAL_UX_SOURCE')
+    output = console.command(preparation, 180)
+    try:
+        result = json.loads(output)
+    except ValueError as exc:
+        raise SmokeError('Installed terminal UX did not return framed JSON evidence') from exc
+    if result.get('status') != 'passed':
+        raise SmokeError('Installed terminal UX did not verify its packaged CLI')
+    return output
+
+
 def installed_admin(console, password):
     console.buffer = b''
     prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
@@ -875,6 +1122,7 @@ def main(argv=None):
         if output.strip() != b'hydratest':
             raise SmokeError('Installed account identity mismatch')
         checkpoint('installed_password_login', output)
+        checkpoint('installed_terminal_ux', installed_terminal_ux_probe(console))
         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')

dev/smoke.py

@@ -178,7 +178,7 @@ def main():
                                     env=viewer_env, stdout=vl, stderr=vl)
             processes.append(proc)
             log = root / f'{name}-{number}.log'
-            wait_for(lambda: f'set_title("HydraOS: {name}")' in log.read_text(), f'{name} viewer did not connect')
+            wait_for(lambda: f'set_title("Hydra OS: {name}")' in log.read_text(), f'{name} viewer did not connect')
             return proc, log
 
         for name in names:
@@ -220,8 +220,8 @@ def main():
         post_usb_switches = [switch(106, alog, blog), switch(105, blog, alog)]
         assert alpha.poll() is None and beta.poll() is None
         assert identities == {name: manager.paths(name).pid.read_text().strip() for name in names}, 'switch restarted a guest'
-        assert alog.read_text().count('set_title("HydraOS: alpha")') == 1, 'switch reconnected alpha'
+        assert alog.read_text().count('set_title("Hydra OS: alpha")') == 1, 'switch reconnected alpha'
-        assert blog.read_text().count('set_title("HydraOS: beta")') == 1, 'switch reconnected beta'
+        assert blog.read_text().count('set_title("Hydra OS: beta")') == 1, 'switch reconnected beta'
         with manager._connect(manager.paths('alpha')) as qmp:
             usb_after = qmp.command('human-monitor-command', {'command-line': 'info usb'})
         assert usb_after == usb_before, 'opening selector attached an unselected USB device'

tests/test_iso_smoke.py

@@ -632,3 +632,51 @@ else: sys.exit(1)
     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)
+
+
+    def test_terminal_ux_fixture_has_complete_iso_volume_and_nonboot_directory(self):
+        self.assertTrue(hasattr(self.mod, 'terminal_ux_iso'), 'Installed UX ISO fixture is missing')
+        from hydra.media import validate_iso, MediaError
+        iso = self.root / 'ux.iso'
+        data = self.mod.terminal_ux_iso()
+        iso.write_bytes(data)
+        self.assertEqual(validate_iso(iso), iso)
+        self.assertEqual(len(data), 21 * 2048)
+        self.assertEqual(data[16 * 2048:16 * 2048 + 7], b'\x01CD001\x01')
+        self.assertEqual(data[17 * 2048:17 * 2048 + 7], b'\xffCD001\x01')
+        self.assertEqual(data[20 * 2048 + 25], 2, 'Root directory must be a directory record')
+        self.assertNotIn(b'EL TORITO SPECIFICATION', data, 'Fixture must not imply an OS boot')
+        iso.write_bytes(data[:-2048])
+        with self.assertRaises(MediaError):
+            validate_iso(iso)
+
+    def test_terminal_ux_branding_rejects_wrong_os_missing_logo_and_render_errors(self):
+        self.assertTrue(hasattr(self.mod, 'verify_terminal_ux_branding'), 'Default Fastfetch verification is missing')
+        rendered = ' .-(o >    .-(o >    .-(o >\nH Y D R A  O S\nOS: Hydra OS x86_64\n'
+        self.mod.verify_terminal_ux_branding('\x1b[36m' + rendered + '\x1b[0m', '')
+        for output, error in ((rendered.replace('OS: Hydra OS', 'OS: Gentoo'), ''),
+                              (rendered.replace('(o >', 'head'), ''),
+                              (rendered, 'Unknown JSON key: invalid'),
+                              (rendered + 'Memory: Error loading module\n', '')):
+            with self.subTest(output=output, error=error):
+                with self.assertRaises(self.mod.SmokeError):
+                    self.mod.verify_terminal_ux_branding(output, error)
+
+    def test_terminal_ux_snapshot_checks_live_qmp_identity_and_same_process(self):
+        self.assertTrue(hasattr(self.mod, 'terminal_ux_snapshot'), 'Installed UX identity verification is missing')
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.terminal_ux_snapshot(context)
+        self.assertEqual(before['pid'], process.pid)
+        self.assertEqual(before['uuid'], context['uuid'])
+        self.assertEqual(self.mod.terminal_ux_snapshot(context), before)
+        context['qmp_name'] = 'foreign-process'
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.terminal_ux_snapshot(context)
+        context.pop('qmp_name')
+        process.terminate(); process.wait(timeout=3)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.terminal_ux_snapshot(context)
+
+    def test_installed_terminal_ux_script_compiles_as_standalone_guest_probe(self):
+        self.assertTrue(hasattr(self.mod, 'terminal_ux_guest_script'), 'Installed terminal UX probe is missing')
+        compile(self.mod.terminal_ux_guest_script(), '<guest-terminal-ux>', 'exec')