diff --git a/dev/iso-smoke.py b/dev/iso-smoke.py index aafbd08..a2e6763 100644 --- a/dev/iso-smoke.py +++ b/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') diff --git a/dev/smoke.py b/dev/smoke.py index e92c9dc..c238b28 100644 --- a/dev/smoke.py +++ b/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 blog.read_text().count('set_title("HydraOS: beta")') == 1, 'switch reconnected beta' + assert alog.read_text().count('set_title("Hydra OS: alpha")') == 1, 'switch reconnected alpha' + 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' diff --git a/tests/test_iso_smoke.py b/tests/test_iso_smoke.py index 6bf9993..a5c4e53 100644 --- a/tests/test_iso_smoke.py +++ b/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(), '', 'exec')