11 Commits
28 changed files with 3299 additions and 1 deletions
+29
View File
@@ -24,3 +24,32 @@ a refresh-rate setting does not prove delivered FPS.
- [Hardware target](docs/hardware/asus-fx517zc.md)
Build outputs, VM disks, runtime state and credentials stay outside Git.
## Run the prototype
```sh
python3 -m unittest discover -s tests -v
nix-shell dev/tools.nix
bin/hydra add guest --iso /absolute/path/installer.iso
bin/hydra start guest
bin/hydra view guest
# From a dedicated graphical login; --nested uses the current graphical display.
bin/hydra session --nested
```
After installation, shut the guest down and run `bin/hydra media guest --detach`
to boot its existing disk without installation media. `add` also supports disk-only
profiles, `--width`, `--height`, and `--refresh-millihz`. Distribution preferences
in `/etc/hydra/defaults.toml` affect only new profiles; explicit CLI options win.
`nix-shell dev/tools.nix --run 'python3 dev/smoke.py'` checks two disposable KVM
counter guests and private headless viewers, including manual USB selection,
reserved switching under inhibition, recovery and reconnect. See
[verification commands and physical checks still pending](docs/verification.md).
An empty session opens local recovery. Enter `:shell`, create and start guests
with `hydra add` / `hydra start`, then leave the shell and enter `:refresh`.
Refresh rebuilds workspaces and reconnects viewers for the current profiles;
running VMs keep their processes and disks. Use it after adding or removing
profiles. New profiles save an individual MAC address; legacy profiles obtain
a stable address from their owner and disk path, including on the shared bridge.
Executable
+9
View File
@@ -0,0 +1,9 @@
#!/usr/bin/env python3
"""Run HydraOS from this checkout without installing a host package."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from hydra.cli import main
raise SystemExit(main())
+8
View File
@@ -0,0 +1,8 @@
#!/bin/sh
set -eu
source_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
build_dir=$(mktemp -d /tmp/hydra-input-build-XXXXXX)
wayland-scanner client-header "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-client-protocol.h"
wayland-scanner private-code "$source_dir/virtual-keyboard-unstable-v1.xml" "$build_dir/virtual-keyboard-protocol.c"
cc -Wall -Wextra -Werror "$source_dir/input-probe.c" "$build_dir/virtual-keyboard-protocol.c" -I"$build_dir" $(pkg-config --cflags --libs wayland-client xkbcommon) -o "$build_dir/input-probe"
printf '%s\n' "$build_dir/input-probe"
+100
View File
@@ -0,0 +1,100 @@
#!/usr/bin/env python3
"""Compile actual QEMU EDID code and decode the guest-visible response.
Catches loss of the configured refresh rate, a frontend overriding it, and
monitor ranges excluding the generated preferred mode. No VM is touched.
"""
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import tarfile
source_root = Path(sys.argv[1])
# Accept the pinned QEMU source archive directly. Read only the four files
# needed here, avoiding unrelated upstream absolute symlinks in rom sources.
if source_root.is_file():
extracted = tempfile.TemporaryDirectory(prefix='hydra-edid-source-')
with tarfile.open(source_root) as archive:
members = {name: next(member for member in archive.getmembers()
if member.name.endswith('/' + name)) for name in (
'hw/display/edid-generate.c', 'include/hw/display/edid.h',
'include/hw/virtio/virtio-gpu.h', 'hw/display/virtio-gpu-base.c')}
source_root = Path(extracted.name)
for name, member in members.items():
destination = source_root / name
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_bytes(archive.extractfile(member).read())
subprocess.run(['patch', '--batch', '-p1', '-i',
str(Path(__file__).resolve().parents[1] / 'patches/qemu-virtio-refresh-rate.patch')],
cwd=source_root, check=True)
edid = (source_root / 'hw/display/edid-generate.c').read_text()
edid = '\n'.join(line for line in edid.splitlines() if not line.startswith('#include'))
header = (source_root / 'include/hw/display/edid.h').read_text()
header = header[header.index('typedef struct qemu_edid_info'):header.index('void qemu_edid_region_io')]
gpu = (source_root / 'hw/display/virtio-gpu-base.c').read_text()
start = gpu.index('void\nvirtio_gpu_base_generate_edid')
gpu = gpu[start:gpu.index('static void virtio_gpu_invalidate_display', start)]
stubs = r'''
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#define ARRAY_SIZE(x) (sizeof(x)/sizeof((x)[0]))
#define MAX(a,b) ((a) > (b) ? (a) : (b))
#define MIN(a,b) ((a) < (b) ? (a) : (b))
#define DIV_ROUND_UP(n,d) (((n) + (d) - 1) / (d))
#define cpu_to_le32(x) (x)
static void stw_le_p(void *p,uint16_t v) { uint8_t *b=p; b[0]=v; b[1]=v>>8; }
static void stw_be_p(void *p,uint16_t v) { uint8_t *b=p; b[1]=v; b[0]=v>>8; }
static void stl_le_p(void *p,uint32_t v) { uint8_t *b=p; for(int i=0;i<4;i++) b[i]=v>>(8*i); }
typedef struct VirtIOGPUOutput { const char *name; } VirtIOGPUOutput;
typedef struct VirtIOGPUOutputList { struct VirtIOGPUOutputList *next; VirtIOGPUOutput *value; } VirtIOGPUOutputList;
struct virtio_gpu_requested_state { uint16_t width_mm,height_mm; uint32_t width,height,refresh_rate; int x,y; };
typedef struct VirtIOGPUBase { struct { uint32_t refresh_rate; VirtIOGPUOutputList *outputs; } conf; struct virtio_gpu_requested_state req_state[1]; } VirtIOGPUBase;
struct virtio_gpu_resp_edid { uint32_t size; uint8_t edid[1024]; };
'''
main = r'''
int main(int argc,char **argv) {
VirtIOGPUBase gpu = {0}; struct virtio_gpu_resp_edid out = {0};
gpu.conf.refresh_rate=atoi(argv[1]); gpu.req_state[0].refresh_rate=atoi(argv[2]);
gpu.req_state[0].width=atoi(argv[3]); gpu.req_state[0].height=atoi(argv[4]);
virtio_gpu_base_generate_edid(&gpu,0,&out);
fwrite(out.edid,1,qemu_edid_size(out.edid),stdout); return 0;
}
'''
compiler = os.environ.get('CC', 'cc')
with tempfile.TemporaryDirectory(prefix='qemu-edid-probe-') as temporary:
directory = Path(temporary)
(directory / 'probe.c').write_text(stubs + header + edid + gpu + main)
subprocess.run([compiler, '-O2', str(directory / 'probe.c'), '-o', str(directory / 'probe')], check=True)
for configured, frontend, width, height, expected in [
(0, 0, 1920, 1080, 75),
(0, 60000, 1920, 1080, 60),
(144000, 0, 1920, 1080, 144),
(144000, 60000, 1920, 1080, 144),
(144000, 0, 1600, 900, 144),
(0, 144000, 1920, 1080, 144),
(50000, 0, 1920, 1080, 50),
(255000, 0, 1280, 800, 255),
]:
data = subprocess.check_output([str(directory / 'probe'), *map(str, (configured, frontend, width, height))])
assert all(sum(data[offset:offset + 128]) % 256 == 0 for offset in range(0, len(data), 128)), 'invalid EDID checksum'
timing = data[54:72]
clock = int.from_bytes(timing[:2], 'little') * 10000
horizontal = timing[2] + ((timing[4] >> 4) << 8)
hblank = timing[3] + ((timing[4] & 15) << 8)
vertical = timing[5] + ((timing[7] >> 4) << 8)
vblank = timing[6] + ((timing[7] & 15) << 8)
actual = clock / (horizontal + hblank) / (vertical + vblank)
assert (horizontal, vertical) == (width, height)
assert abs(actual - expected) < 0.01, f'configured={configured}, frontend={frontend}: wanted {expected} Hz, got {actual}'
ranges = next(data[offset:offset + 18] for offset in (54, 72, 90, 108) if data[offset:offset + 5] == b'\0\0\0\xfd\0')
assert ranges[5] <= actual <= ranges[6], f'vertical range {ranges[5:7]} excludes {actual}'
horizontal_khz = clock / (horizontal + hblank) / 1000
assert ranges[7] <= horizontal_khz <= ranges[8], f'horizontal range {ranges[7:9]} excludes {horizontal_khz}'
if configured == 0 and expected <= 75:
assert ranges[5:9] == bytes((50, 125, 30, 160)), 'legacy monitor range changed'
print(f'PASS configured={configured}, frontend={frontend}: {width}x{height}@{actual:.6f}, ranges={list(ranges[5:9])}')
+78
View File
@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""Check real NIC identities on a bridge isolated in user/net/mount namespaces."""
import argparse
import json
from pathlib import Path
import subprocess
import sys
import tempfile
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hydra.profiles import Profile
from hydra.runtime import Runtime
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--namespace-root', type=Path, help=argparse.SUPPRESS)
args = parser.parse_args()
if args.namespace_root is None:
root = Path(tempfile.mkdtemp(prefix='hydra-network-', dir='/tmp'))
print('Evidence: ' + str(root), flush=True)
(root / 'parent-namespaces.json').write_text(json.dumps({name: str(Path('/proc/self/ns', name).readlink())
for name in ('user', 'net', 'mnt')}))
# All bridge and /etc changes below occur after this explicit namespace
# boundary. The inherited host bridge configuration is never modified.
subprocess.run(['unshare', '-Urnm', '--', str(Path(sys.executable).resolve()),
str(Path(__file__).resolve()), '--namespace-root', str(root)], check=True)
return
root = args.namespace_root
parent_namespaces = json.loads((root / 'parent-namespaces.json').read_text())
namespaces = {name: str(Path('/proc/self/ns', name).readlink()) for name in ('user', 'net', 'mnt')}
assert all(namespaces[name] != parent_namespaces[name] for name in namespaces), 'private namespaces required'
private_etc = root / 'etc'
(private_etc / 'qemu').mkdir(parents=True)
(private_etc / 'qemu/bridge.conf').write_text('allow hydrabr0\n')
subprocess.run(['mount', '--bind', str(private_etc), '/etc'], check=True)
subprocess.run(['ip', 'link', 'add', 'hydrabr0', 'type', 'bridge'], check=True)
subprocess.run(['ip', 'link', 'set', 'hydrabr0', 'up'], check=True)
manager = Runtime(root / 'config', root / 'state', root / 'run')
result = {'evidence': str(root), 'namespace': 'private user/net/mount',
'parent_namespaces': parent_namespaces, 'probe_namespaces': namespaces, 'names': {}}
try:
for name in ('alpha', 'beta'):
manager.add(Profile(name, memory=128, cpus=1, disk_size='8M',
graphics='software', network='bridge'))
manager.start(name)
def actual_mac(name):
with manager._connect(manager.paths(name)) as qmp:
network = qmp.command('human-monitor-command', {'command-line': 'info network'})
mac = manager.profile(name).mac
assert 'macaddr=' + mac in network, network
assert int(mac.split(':')[0], 16) & 3 == 2
return mac, network
for name in ('alpha', 'beta'):
mac, network = actual_mac(name)
result['names'][name] = {'mac': mac, 'network': network}
assert result['names']['alpha']['mac'] != result['names']['beta']['mac']
links = json.loads(subprocess.check_output(['ip', '-j', 'link', 'show', 'master', 'hydrabr0'], text=True))
assert len(links) == 2, links
result['bridge_ports'] = [link['ifname'] for link in links]
manager.stop('alpha', force=True)
manager.start('alpha')
assert actual_mac('alpha')[0] == result['names']['alpha']['mac']
result['restart_preserved_mac'] = True
result['dhcp_and_guest_network_drivers_pending'] = True
result['passed'] = True
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
print(json.dumps(result, indent=2))
finally:
for profile in manager.profiles():
if manager.status(profile.name)['status'] == 'running':
manager.stop(profile.name, force=True)
if __name__ == '__main__':
main()
+25
View File
@@ -0,0 +1,25 @@
.code16
.global _start
_start:
cli
xor %ax,%ax
mov %ax,%ds
mov %ax,%ss
mov $0x7c00,%sp
movl $0x48594452,0x508
movl $0,0x500
movw $0,0x504
loop:
incl 0x500
in $0x64,%al
test $1,%al
jz loop
in $0x60,%al
mov 0x504,%bx
mov %al,0x600(%bx)
inc %bx
and $255,%bx
mov %bx,0x504
jmp loop
.org 510
.word 0xaa55
+63
View File
@@ -0,0 +1,63 @@
/* Private-display integration helper; stdin uses Linux evdev key codes.
* Compile via dev/build-input-probe.sh inside the temporary tools shell. */
#define _GNU_SOURCE
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <unistd.h>
#include <wayland-client.h>
#include <xkbcommon/xkbcommon.h>
#include "virtual-keyboard-client-protocol.h"
static struct wl_seat *seat;
static struct zwp_virtual_keyboard_manager_v1 *manager;
static void global(void *data, struct wl_registry *registry, uint32_t id,
const char *interface, uint32_t version) {
(void)data; (void)version;
if (!strcmp(interface,"wl_seat"))
seat = wl_registry_bind(registry,id,&wl_seat_interface,1);
else if (!strcmp(interface,"zwp_virtual_keyboard_manager_v1"))
manager = wl_registry_bind(registry,id,&zwp_virtual_keyboard_manager_v1_interface,1);
}
static void removed(void *data,struct wl_registry *registry,uint32_t id) {
(void)data; (void)registry; (void)id;
}
int main(void) {
/* The caller supplies the separately created scratch Wayland runtime. */
assert(getenv("WAYLAND_DISPLAY") && getenv("XDG_RUNTIME_DIR"));
struct wl_display *display = wl_display_connect(NULL); assert(display);
struct wl_registry *registry = wl_display_get_registry(display);
struct wl_registry_listener listener={global,removed};
wl_registry_add_listener(registry,&listener,NULL); wl_display_roundtrip(display);
assert(seat && manager);
struct zwp_virtual_keyboard_v1 *keyboard = zwp_virtual_keyboard_manager_v1_create_virtual_keyboard(manager,seat);
struct xkb_context *context=xkb_context_new(XKB_CONTEXT_NO_FLAGS);
struct xkb_rule_names names={.rules="evdev",.model="pc105",.layout="us"};
struct xkb_keymap *map=xkb_keymap_new_from_names(context,&names,XKB_KEYMAP_COMPILE_NO_FLAGS); assert(map);
struct xkb_state *state=xkb_state_new(map); assert(state);
char *text=xkb_keymap_get_as_string(map,XKB_KEYMAP_FORMAT_TEXT_V1);
size_t size=strlen(text)+1;
int fd=memfd_create("hydra-private-keymap",0); assert(fd>=0);
assert(write(fd,text,size)==(ssize_t)size);
zwp_virtual_keyboard_v1_keymap(keyboard,WL_KEYBOARD_KEYMAP_FORMAT_XKB_V1,fd,size);
wl_display_roundtrip(display); close(fd); free(text);
char command[16]; unsigned key; unsigned tick=0;
while (scanf("%15s %u",command,&key)==2) {
if (!strcmp(command,"wait")) { usleep(key*1000); continue; }
unsigned down=!strcmp(command,"down");
assert(down || !strcmp(command,"up"));
xkb_state_update_key(state,key+8,down?XKB_KEY_DOWN:XKB_KEY_UP);
zwp_virtual_keyboard_v1_key(keyboard,++tick,key,down);
zwp_virtual_keyboard_v1_modifiers(keyboard,
xkb_state_serialize_mods(state,XKB_STATE_MODS_DEPRESSED),
xkb_state_serialize_mods(state,XKB_STATE_MODS_LATCHED),
xkb_state_serialize_mods(state,XKB_STATE_MODS_LOCKED),
xkb_state_serialize_layout(state,XKB_STATE_LAYOUT_EFFECTIVE));
assert(wl_display_roundtrip(display)>=0);
}
zwp_virtual_keyboard_v1_destroy(keyboard); wl_display_roundtrip(display);
xkb_state_unref(state); xkb_keymap_unref(map); xkb_context_unref(context);
wl_display_disconnect(display);
return 0;
}
+196
View File
@@ -0,0 +1,196 @@
#!/usr/bin/env python3
"""Exercise the product's empty-session/recovery/refresh loop privately."""
import argparse
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import time
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hydra.runtime import Runtime
from smoke import (wait_for, counter, focus_events, fresh_focus_transition,
measure_hidden_interval, classify_compositor_errors)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
parser.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
parser.add_argument('--keyboard', required=True, help='compiled dev/input-probe.c')
args = parser.parse_args()
for name in ('labwc', 'viewer', 'keyboard'):
executable = shutil.which(getattr(args, name))
if not executable:
parser.error('missing executable: ' + getattr(args, name))
setattr(args, name, executable)
if not os.access('/dev/kvm', os.R_OK | os.W_OK):
parser.error('KVM access required')
source = Path(__file__).resolve().parents[1]
root = Path(tempfile.mkdtemp(prefix='hydra-session-refresh-', dir='/tmp'))
print('Evidence: ' + str(root), flush=True)
xdg = root / 'xdg'; xdg.mkdir(mode=0o700)
manager = Runtime(root / 'config', root / 'state', root / 'run')
cli = [sys.executable, str(source / 'bin/hydra'), '--config', str(manager.config),
'--state', str(manager.state), '--runtime', str(manager.runtime)]
subprocess.run(['as', '--32', str(source / 'dev/guest-probe.S'), '-o', str(root / 'boot.o')], check=True)
subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(root / 'boot.o'), str(root / 'boot.bin')], check=True)
(root / 'boot.raw').write_bytes((root / 'boot.bin').read_bytes() + bytes(8 * 1024 * 1024 - 512))
def wrapper(name, code):
path = root / name
path.write_text('#!' + sys.executable + '\n' + code)
path.chmod(0o700)
return str(path)
common = 'import json, os, pathlib, subprocess, sys, time\nroot = pathlib.Path(' + repr(str(root)) + ')\n'
labwc = wrapper('labwc-private', common + '''
if '--version' in sys.argv:
os.execv(@LABWC@, [@LABWC@] + sys.argv[1:])
config = pathlib.Path(sys.argv[sys.argv.index('-C') + 1])
number = len(list(root.glob('snapshot-*.json')))
data = json.loads((config.parent / 'session.json').read_text())
data.update(pid=os.getpid(), config_directory=str(config))
(root / ('snapshot-%d.xml' % number)).write_text((config / 'rc.xml').read_text())
(root / ('snapshot-%d.json' % number)).write_text(json.dumps(data))
# Test-only wrapper replaces physical backends before the real compositor runs.
os.environ.update(WLR_BACKENDS='headless', WLR_RENDERER='pixman',
WLR_XWAYLAND='/bin/false', LABWC_UPDATE_ACTIVATION_ENV='0',
XDG_DATA_DIRS='/nonexistent')
os.execv(@LABWC@, [@LABWC@, '-d'] + sys.argv[1:])
'''.replace('@LABWC@', repr(args.labwc)))
viewer = wrapper('viewer-private', common + '''
if '--help' in sys.argv:
os.execv(VIEWER, [VIEWER] + sys.argv[1:])
name = sys.argv[sys.argv.index('--title') + 1].split(': ', 1)[1]
(root / (name + '-viewer.json')).write_text(json.dumps({
'argv': sys.argv[1:], 'config': os.environ['XDG_CONFIG_HOME'],
'gdk_backend': os.environ.get('GDK_BACKEND'), 'wayland_display': os.environ['WAYLAND_DISPLAY']}))
log = os.open(root / (name + '.log'), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
os.dup2(log, 1); os.dup2(log, 2)
os.environ['WAYLAND_DEBUG'] = '1'
os.execv(VIEWER, [VIEWER] + sys.argv[1:])
'''.replace('VIEWER', repr(args.viewer)))
terminal = wrapper('terminal-private', common + '''
manifest = pathlib.Path(sys.argv[-1])
socket = pathlib.Path(os.environ['XDG_RUNTIME_DIR']) / os.environ['WAYLAND_DISPLAY']
(root / 'recovery-ready').write_text(str(manifest))
while not (root / 'recovery-input').exists():
if not socket.exists():
sys.exit(0)
time.sleep(.05)
if not socket.exists():
sys.exit(0)
commands = (root / 'recovery-input').read_text()
# The actual product recovery helper executes :shell and :refresh.
with (root / 'recovery.log').open('a') as log:
sys.exit(subprocess.run(sys.argv[2:], input=commands, text=True, stdout=log, stderr=log).returncode)
''')
shell = wrapper('scratch-shell', common + 'sys.path.insert(0, ' + repr(str(source)) + ')\n' + '''
from hydra.runtime import Runtime
from smoke import counter, wait_for
cli = CLI
manager = Runtime(root / 'config', root / 'state', root / 'run')
for name in ('alpha', 'beta'):
subprocess.run(cli + ['add', name, '--memory', '128', '--cpus', '1', '--disk-size', '8M',
'--graphics', 'software', '--network', 'user'], check=True)
paths = manager.paths(name); manager.prepare(paths)
subprocess.run([os.environ['HYDRA_QEMU_IMG'], 'convert', '-f', 'raw', '-O', 'qcow2',
str(root / 'boot.raw'), str(paths.disk)], check=True)
subprocess.run(cli + ['start', name], check=True)
time.sleep(.3)
identities = {name: {'pid': manager.paths(name).pid.read_text().strip(),
'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
for name in ('alpha', 'beta')}
(root / 'before.json').write_text(json.dumps(identities))
'''.replace('CLI', repr(cli)).replace('from smoke import', 'from dev.smoke import'))
env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), HYDRA_LABWC=labwc,
HYDRA_VIEWER=viewer, HYDRA_TERMINAL=terminal, HYDRA_BRIGHTNESS=shutil.which('true'),
SHELL=shell, GSETTINGS_BACKEND='memory')
for key in ('WAYLAND_DISPLAY', 'DISPLAY', 'LABWC_PID'):
env.pop(key, None)
log = (root / 'session.log').open('w')
launcher, injector = None, None
result = {'evidence': str(root), 'physical_input_pending': True,
'terminal': 'test-only command feeder; real recovery helper and scratch :shell',
'backend': 'test-only wrapper around actual product launch_session and actual labwc'}
try:
launcher = subprocess.Popen(cli + ['session'], env=env, stdout=log, stderr=log)
wait_for(lambda: (root / 'recovery-ready').exists(), 'empty session did not open recovery')
first = json.loads((root / 'snapshot-0.json').read_text())
assert first['names'] == [], 'session was not initially empty'
(root / 'recovery-input').write_text(':shell\n:refresh\n')
wait_for(lambda: (root / 'snapshot-1.json').exists(), 'product loop did not restart after :refresh', timeout=15)
second = json.loads((root / 'snapshot-1.json').read_text())
assert second['names'] == ['alpha', 'beta'], second
assert first['pid'] != second['pid'], 'compositor was not refreshed'
logs = {name: root / (name + '.log') for name in ('alpha', 'beta')}
for name, path in logs.items():
wait_for(lambda: path.exists() and 'set_title("HydraOS: ' + name + '")' in path.read_text(),
name + ' product viewer did not connect')
viewers = {name: json.loads((root / (name + '-viewer.json')).read_text()) for name in logs}
assert len({data['config'] for data in viewers.values()}) == 2, 'viewer settings were shared'
for data in viewers.values():
assert data['gdk_backend'] == 'wayland'
assert data['config'].startswith(str(manager.runtime) + '/viewer-')
assert '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0' in data['argv']
assert '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0' in data['argv']
alog, blog = logs['alpha'], logs['beta']
display = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
keyboard_env = dict(env, WAYLAND_DISPLAY=display)
injector = subprocess.Popen([args.keyboard], env=keyboard_env, stdin=subprocess.PIPE, text=True)
time.sleep(.2)
wait_for(lambda: focus_events(alog) and focus_events(alog)[-1] == 'enter', 'alpha did not focus')
def switch(code, source_log, target_log):
before_source, before_target = focus_events(source_log), focus_events(target_log)
injector.stdin.write(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29\nwait 100\n')
injector.stdin.flush()
wait_for(lambda: fresh_focus_transition(before_source, before_target,
focus_events(source_log), focus_events(target_log)), 'fresh reserved focus transition failed')
switch(106, alog, blog)
intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
switch(105, blog, alog)
intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
before = json.loads((root / 'before.json').read_text())
after = {name: {'pid': manager.paths(name).pid.read_text().strip(),
'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
for name in ('alpha', 'beta')}
for name in before:
assert after[name]['pid'] == before[name]['pid'] and after[name]['uuid'] == before[name]['uuid']
assert after[name]['counter'] != before[name]['counter']
# Only the PID recorded by our own wrapper, verified against its private
# config path, may be signalled. Never inherit a host compositor PID.
command = Path('/proc') / str(second['pid']) / 'cmdline'
assert second['config_directory'].encode() in command.read_bytes()
os.kill(second['pid'], signal.SIGTERM)
assert launcher.wait(timeout=10) == 0, 'product session exit failed'
for name in before:
assert manager.status(name)['status'] == 'running'
assert manager.paths(name).pid.read_text().strip() == before[name]['pid']
result.update(initial_names=first['names'], refreshed_names=second['names'],
compositor_pids=[first['pid'], second['pid']], before=before, after=after,
viewers=viewers,
hidden_intervals=intervals, survived_session_exit=True,
expected_xwayland_diagnostics=classify_compositor_errors((root / 'session.log').read_text()),
passed=True)
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
print(json.dumps(result, indent=2))
finally:
if injector and injector.poll() is None:
injector.terminate(); injector.wait(timeout=5)
if launcher and launcher.poll() is None:
launcher.send_signal(signal.SIGINT); launcher.wait(timeout=12)
for profile in manager.profiles():
if manager.status(profile.name)['status'] == 'running':
manager.stop(profile.name, force=True)
log.close()
if __name__ == '__main__':
main()
+273
View File
@@ -0,0 +1,273 @@
#!/usr/bin/env python3
"""Two disposable KVM guests on a private headless display; never host input."""
import argparse
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import time
from xml.etree import ElementTree as ET
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hydra.profiles import Profile
from hydra.runtime import Runtime
from hydra.session import build_labwc_config, build_viewer_argv, _viewer_environment
def wait_for(predicate, description, timeout=8):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return
time.sleep(.05)
raise AssertionError(description)
def counter(manager, name):
with manager._connect(manager.paths(name)) as qmp:
reply = qmp.command('human-monitor-command', {'command-line': 'xp /3wx 0x500'})
values = [int(word, 16) for word in re.findall(r'0x([0-9a-fA-F]+)', reply)]
assert len(values) == 3 and values[2] == 0x48594452, 'scratch guest did not boot'
return values[0]
def focus_events(log):
"""Only keyboard focus, excluding pointer-enter and old window titles."""
return re.findall(r'wl_keyboard#\d+\.(enter|leave)\(', log.read_text())
def fresh_focus_transition(source_before, target_before, source_after, target_after):
return (len(source_after) > len(source_before) and source_after[-1] == 'leave'
and len(target_after) > len(target_before) and target_after[-1] == 'enter')
def measure_hidden_interval(read_counter, hidden_log, visible_log, duration=.5):
hidden_focus, visible_focus = focus_events(hidden_log), focus_events(visible_log)
assert hidden_focus and hidden_focus[-1] == 'leave', 'sampled guest is not hidden'
assert visible_focus and visible_focus[-1] == 'enter', 'other guest is not focused'
def unchanged():
assert (focus_events(hidden_log) == hidden_focus and
focus_events(visible_log) == visible_focus), 'focus changed during hidden interval'
start = time.monotonic()
before = read_counter()
unchanged()
while time.monotonic() - start < duration:
time.sleep(min(.05, duration))
unchanged()
after = read_counter()
unchanged()
assert after != before, 'hidden guest stopped working during continuously hidden interval'
return {'before': before, 'after': after,
'elapsed_seconds': time.monotonic() - start, 'focus_unchanged': True}
def classify_compositor_errors(log):
expected = []
for line in log.splitlines():
if '[ERROR]' not in line:
continue
if re.search(r'\[xwayland/sockets\.c:\d+\] Failed to bind socket '
r'@?/tmp/\.X11-unix/X\d+: Address already in use$', line):
expected.append(line)
else:
raise AssertionError('Unexpected compositor error: ' + line)
return expected
def main():
cli = argparse.ArgumentParser(description=__doc__)
cli.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
cli.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
cli.add_argument('--keyboard', help='compiled dev/input-probe.c; built automatically when omitted')
args = cli.parse_args()
for name in ('viewer', 'labwc'):
value = shutil.which(getattr(args, name))
if not value:
cli.error(f'missing {name}: {getattr(args, name)}')
setattr(args, name, value)
source = Path(__file__).resolve().parent
if not args.keyboard:
args.keyboard = subprocess.check_output([str(source / 'build-input-probe.sh')], text=True).strip()
if not os.access('/dev/kvm', os.R_OK | os.W_OK):
cli.error('KVM access required for this integration check')
root = Path(tempfile.mkdtemp(prefix='hydra-smoke-', dir='/tmp'))
print(f'Evidence: {root}', flush=True)
xdg = root / 'xdg'
xdg.mkdir(mode=0o700)
config = root / 'labwc'
config.mkdir()
recovery = root / 'recovery'
unreserved = root / 'unreserved'
names = ['alpha', 'beta']
rc = ET.fromstring(build_labwc_config(names, [sys.executable, '-c',
'from pathlib import Path; Path(' + repr(str(recovery)) + ').touch()'], initial='alpha'))
# Test-only controls prove native labwc inhibition is active; production
# config has no ToggleKeybinds or ordinary host shortcut.
keyboard = rc.find('keyboard')
binding = ET.SubElement(keyboard, 'keybind', key='C-A-S-i')
ET.SubElement(binding, 'action', name='ToggleKeybinds')
binding = ET.SubElement(keyboard, 'keybind', key='F12')
ET.SubElement(binding, 'action', name='Execute', command=shutil.which('touch') + ' ' + str(unreserved))
ET.ElementTree(rc).write(config / 'rc.xml')
(config / 'autostart').write_text('#!/bin/sh\n')
(config / 'environment').touch()
env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), WLR_BACKENDS='headless',
WLR_RENDERER='pixman', WLR_XWAYLAND='/bin/false', XDG_DATA_DIRS='/nonexistent',
GSETTINGS_BACKEND='memory', LABWC_UPDATE_ACTIVATION_ENV='0')
for key in ('DISPLAY', 'WAYLAND_DISPLAY'):
env.pop(key, None)
manager = Runtime(root / 'config', root / 'state', root / 'run')
boot_object, boot = root / 'boot.o', root / 'boot.bin'
subprocess.run(['as', '--32', str(source / 'guest-probe.S'), '-o', str(boot_object)], check=True)
subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(boot_object), str(boot)], check=True)
raw = root / 'boot.raw'
raw.write_bytes(boot.read_bytes() + bytes(8 * 1024 * 1024 - 512))
media = root / 'installation.iso'
media.touch()
for name in names:
manager.add(Profile(name, media if name == 'alpha' else None, memory=128, cpus=1, disk_size='8M', graphics='software',
width=1920, height=1080, refresh_millihz=144000))
paths = manager.paths(name)
manager.prepare(paths)
subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'convert', '-f', 'raw',
'-O', 'qcow2', str(raw), str(paths.disk)], check=True)
processes, streams = [], []
result = {'evidence': str(root), 'physical_checks_pending': True}
try:
cl = (root / 'labwc.log').open('w'); streams.append(cl)
compositor = subprocess.Popen([args.labwc, '-d', '-C', str(config)], env=env, stdout=cl, stderr=cl)
processes.append(compositor)
wait_for(lambda: any(not p.name.endswith('.lock') for p in xdg.glob('wayland-*')), 'private display did not start')
env['WAYLAND_DISPLAY'] = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
injector = subprocess.Popen([args.keyboard], env=env, stdin=subprocess.PIPE, text=True)
processes.append(injector)
def keys(commands):
injector.stdin.write(commands + '\nwait 100\n'); injector.stdin.flush()
time.sleep(.15)
assert injector.poll() is None, 'private input helper exited'
def chord(code):
keys(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29')
def switch(key, source_log, target_log):
source_before, target_before = focus_events(source_log), focus_events(target_log)
chord(key)
wait_for(lambda: fresh_focus_transition(source_before, target_before,
focus_events(source_log), focus_events(target_log)),
'reserved switch did not produce fresh keyboard leave/enter')
return {'from_events': len(focus_events(source_log)),
'to_events': len(focus_events(target_log)), 'keycode': key}
def attach(name, number):
vl = (root / f'{name}-{number}.log').open('w'); streams.append(vl)
previous = dict(os.environ)
try:
os.environ.update(env)
viewer_env = _viewer_environment(root / f'{name}-{number}-settings')
finally:
os.environ.clear(); os.environ.update(previous)
viewer_env['WAYLAND_DEBUG'] = '1'
proc = subprocess.Popen(build_viewer_argv(manager.paths(name), args.viewer),
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')
return proc, log
for name in names:
manager.start(name)
media.unlink() # Open QEMU media disappears; status/control/view still work.
time.sleep(1)
before = {name: counter(manager, name) for name in names}
with manager._connect(manager.paths('alpha')) as qmp:
children = qmp.command('qom-list', {'path': '/machine/peripheral-anon'})
video = next(child['name'] for child in children if child['type'] == 'child<virtio-vga>')
mode = {prop: qmp.command('qom-get', {'path': '/machine/peripheral-anon/' + video, 'property': prop})
for prop in ('xres', 'yres', 'refresh_rate')}
assert mode == {'xres': 1920, 'yres': 1080, 'refresh_rate': 144000}, mode
usb_before = qmp.command('human-monitor-command', {'command-line': 'info usb'})
identities = {name: manager.paths(name).pid.read_text().strip() for name in names}
alpha, alog = attach('alpha', 1)
beta, blog = attach('beta', 1)
wait_for(lambda: '.enter(' in alog.read_text(), 'alpha did not receive focus')
# F12 must work before inhibition; then become suppressed in alpha.
keys('down 88\nup 88')
wait_for(unreserved.exists, 'ordinary test binding failed before inhibition')
unreserved.unlink()
chord(23) # I: test-only ToggleKeybinds.
keys('down 88\nup 88')
assert not unreserved.exists(), 'keybinding inhibition did not suppress ordinary F12'
switch(106, alog, blog) # Right: beta focused, alpha continuously hidden.
hidden_intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
switch(105, blog, alog) # Left: alpha focused, beta continuously hidden.
hidden_intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
chord(1) # Escape: production recovery binding still works.
wait_for(recovery.exists, 'reserved recovery failed with inhibition')
dialog = 'set_title("Select USB devices for redirection")'
assert dialog not in alog.read_text(), 'USB selector opened automatically'
# Hold modifiers through opening, then release inside the modal dialog.
keys('down 29\ndown 56\ndown 42\ndown 22')
wait_for(lambda: dialog in alog.read_text(), 'manual USB selector did not open', timeout=3)
keys('up 22\nup 42\nup 56\nup 29')
keys('down 1\nup 1') # Close selector.
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'
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'
alpha.terminate(); alpha.wait(timeout=5)
assert manager.status('alpha')['status'] == 'running'
alpha, attached = attach('alpha', 2)
after = {name: counter(manager, name) for name in names}
assert all(after[name] != before[name] for name in names), 'background guest stopped working'
# Quit only beta; alpha's counter/lifecycle remains independent.
manager.stop('beta', force=True)
assert manager.status('alpha')['status'] == 'running'
assert counter(manager, 'alpha') != after['alpha']
compositor.terminate(); compositor.wait(timeout=5)
assert manager.status('alpha')['status'] == 'running'
manager.stop('alpha', force=True)
manager.media('alpha', None)
assert manager.profile('alpha').iso is None
manager.start('alpha')
time.sleep(1)
disk_boot_counter = counter(manager, 'alpha')
assert disk_boot_counter > 0
expected_compositor_noise = classify_compositor_errors((root / 'labwc.log').read_text())
result.update(hidden_guest_intervals=hidden_intervals, post_usb_switches=post_usb_switches,
expected_compositor_noise=expected_compositor_noise,
manual_usb_dialog=True, standard_us_keymap=True,
native_labwc_inhibition=True, reserved_switch_and_recovery=True,
qemu_survives_viewer_and_compositor=True, independent_lifecycle=True,
guest_counters_before=before, guest_counters_after=after,
guest_pids=identities, explicit_reattach=True, requested_qemu_mode=mode,
usb_devices_unchanged_without_selection=True,
missing_iso_does_not_block_views_or_control=True,
restart_without_original_media=True, disk_boot_counter=disk_boot_counter,
viewer_version=subprocess.check_output([args.viewer, '--version'], text=True).strip(),
guest_input_and_physical_modifiers='not proven by this boot counter workload')
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
print(json.dumps(result, indent=2))
finally:
for proc in reversed(processes):
if proc.poll() is None:
proc.terminate(); proc.wait(timeout=5)
for stream in streams:
stream.close()
for name in names:
if manager.status(name)['status'] != 'stopped':
manager.stop(name, force=True)
if __name__ == '__main__':
main()
+28
View File
@@ -0,0 +1,28 @@
# Temporary tools only: nix-shell dev/tools.nix. No host/profile installation.
let
pkgs = import (builtins.fetchTarball {
url = "https://github.com/NixOS/nixpkgs/archive/7c8764b7c7b09b34f632464276218ef9090eaa11.tar.gz";
sha256 = "sha256-MWhktfFtFE8Jl7bf9Pkp7TR5MSdYCjCswSo0Gkb4ywc=";
}) {};
graphics = import (builtins.fetchTarball {
url = "https://github.com/NixOS/nixpkgs/archive/7fd4004d0511991fff3a0524856a9be09e28c5f0.tar.gz";
sha256 = "sha256-i5qk1kLiit98rRbh1AXmWBNT9QXC/y4l5Wk/u5EdpDQ=";
}) {};
qemu = (pkgs.qemu.override { hostCpuOnly = true; }).overrideAttrs (old: {
patches = (old.patches or []) ++ [ ../patches/qemu-virtio-refresh-rate.patch ];
});
viewer = pkgs.virt-viewer.overrideAttrs (old: {
src = pkgs.fetchurl {
url = "https://releases.pagure.org/virt-viewer/virt-viewer-11.0.tar.xz";
hash = "sha256-pD+iMlxMHHelyMmAZaww7wURohrJjlkPIjQIabrZq9A=";
};
patches = (old.patches or []) ++ [ ../patches/virt-viewer-hydra-session.patch ];
doCheck = true;
});
in pkgs.mkShell {
nativeBuildInputs = [ pkgs.stdenv.cc pkgs.pkg-config pkgs.wayland-scanner ];
buildInputs = [ pkgs.wayland.dev pkgs.libxkbcommon.dev ];
packages = [ pkgs.python3 qemu viewer graphics.labwc pkgs.foot
pkgs.brightnessctl pkgs.wayland-utils pkgs.wlr-randr ];
passthru = { inherit viewer qemu; labwc = graphics.labwc; };
}
+113
View File
@@ -0,0 +1,113 @@
<?xml version="1.0" encoding="UTF-8"?>
<protocol name="virtual_keyboard_unstable_v1">
<copyright>
Copyright © 2008-2011 Kristian Høgsberg
Copyright © 2010-2013 Intel Corporation
Copyright © 2012-2013 Collabora, Ltd.
Copyright © 2018 Purism SPC
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice (including the next
paragraph) shall be included in all copies or substantial portions of the
Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
</copyright>
<interface name="zwp_virtual_keyboard_v1" version="1">
<description summary="virtual keyboard">
The virtual keyboard provides an application with requests which emulate
the behaviour of a physical keyboard.
This interface can be used by clients on its own to provide raw input
events, or it can accompany the input method protocol.
</description>
<request name="keymap">
<description summary="keyboard mapping">
Provide a file descriptor to the compositor which can be
memory-mapped to provide a keyboard mapping description.
Format carries a value from the keymap_format enumeration.
</description>
<arg name="format" type="uint" summary="keymap format"/>
<arg name="fd" type="fd" summary="keymap file descriptor"/>
<arg name="size" type="uint" summary="keymap size, in bytes"/>
</request>
<enum name="error">
<entry name="no_keymap" value="0" summary="No keymap was set"/>
</enum>
<request name="key">
<description summary="key event">
A key was pressed or released.
The time argument is a timestamp with millisecond granularity, with an
undefined base. All requests regarding a single object must share the
same clock.
Keymap must be set before issuing this request.
State carries a value from the key_state enumeration.
</description>
<arg name="time" type="uint" summary="timestamp with millisecond granularity"/>
<arg name="key" type="uint" summary="key that produced the event"/>
<arg name="state" type="uint" summary="physical state of the key"/>
</request>
<request name="modifiers">
<description summary="modifier and group state">
Notifies the compositor that the modifier and/or group state has
changed, and it should update state.
The client should use wl_keyboard.modifiers event to synchronize its
internal state with seat state.
Keymap must be set before issuing this request.
</description>
<arg name="mods_depressed" type="uint" summary="depressed modifiers"/>
<arg name="mods_latched" type="uint" summary="latched modifiers"/>
<arg name="mods_locked" type="uint" summary="locked modifiers"/>
<arg name="group" type="uint" summary="keyboard layout"/>
</request>
<request name="destroy" type="destructor" since="1">
<description summary="destroy the virtual keyboard keyboard object"/>
</request>
</interface>
<interface name="zwp_virtual_keyboard_manager_v1" version="1">
<description summary="virtual keyboard manager">
A virtual keyboard manager allows an application to provide keyboard
input events as if they came from a physical keyboard.
</description>
<enum name="error">
<entry name="unauthorized" value="0" summary="client not authorized to use the interface"/>
</enum>
<request name="create_virtual_keyboard">
<description summary="Create a new virtual keyboard">
Creates a new virtual keyboard associated to a seat.
If the compositor enables a keyboard to perform arbitrary actions, it
should present an error when an untrusted client requests a new
keyboard.
</description>
<arg name="seat" type="object" interface="wl_seat"/>
<arg name="id" type="new_id" interface="zwp_virtual_keyboard_v1"/>
</request>
</interface>
</protocol>
@@ -72,13 +72,17 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
### Task 3: Integration and recovery
**Files:** Create `dev/smoke.py`, `docs/verification.md`; update README with verified commands.
**Files:** Create `dev/smoke.py`, `docs/verification.md`; update README with verified commands. Amend runtime/profile/CLI/session tests and dependency patches for integration defects and the installed-guest workflow below.
**Interfaces:** Consumes Tasks1/2, produces disposable-VM integration evidence.
- [ ] Run two disposable KVM instances/viewers, never existing lab disks; verify private QMP and independent lifecycle.
- [ ] Verify viewer/session exit preserves QEMU and reattach works; switching neither restarts nor reconnects guests.
- [ ] Check input only in disposable session, without injecting into current host or user's guests; add regression tests for discovered defects.
- [ ] Resolve Task 2's observed USB shortcut dispatch failure; demonstrate manual selector opening without automatic USB redirection on the private display. Check reserved switching/recovery actions with a standard keymap and keyboard inhibition.
- [ ] Complete installed-guest boot workflow: optional installation media, explicit attach/detach while stopped, disk boot without the original ISO, and status/stop usable even if previously configured ISO disappears. Preserve exclusive disk creation and verified QMP control.
- [ ] Add validated display dimensions/optional refresh and distribution defaults under `/etc/hydra/defaults.toml`. Defaults affect newly created profiles only; explicit CLI values win. Developer default remains user networking and stock-QEMU-compatible refresh; installed Hydra uses hydrabr0 and ASUS 1920x1080/144000 mHz preference.
- [ ] Port the existing tested `qemu-virtio-refresh-rate.patch` from the old lab as a tracked dependency patch; validate the generated EDID and requested mode. Distinguish advertised refresh from achieved frame rate and physical display verification.
- [ ] Record held-key, physical brightness and 144Hz checks needing dedicated TTY. Nested tests cannot prove them.
- [ ] Run full suite and independent review; fix material findings with reproducing tests.
- [ ] Commit evidence and push reviewed prototype branch to supplied repository.
+178
View File
@@ -0,0 +1,178 @@
# Graphics and recovery verification
Run from the checkout. Tools are temporary Nix development dependencies; no
persistent host installation is required. The two-guest smoke test needs access
to `/dev/kvm` and creates its own headless Wayland display, VM disks and sockets
under a fresh `/tmp/hydra-smoke-*` directory. It never uses the inherited display,
active TTY, existing guests or existing disks. Evidence remains in that directory;
the script stops only its own verified QEMU instances on exit.
```sh
python3 -m unittest discover -s tests -v
nix-build dev/tools.nix -A viewer --no-out-link
nix-build dev/tools.nix -A qemu --no-out-link
nix-shell dev/tools.nix --run 'python3 dev/smoke.py'
nix-shell dev/tools.nix --run 'python3 dev/session-refresh-smoke.py --keyboard "$(dev/build-input-probe.sh)"'
nix-shell dev/tools.nix --run 'python3 dev/check-network.py'
```
The smoke test builds a private input helper with an ordinary evdev/pc105 US
keyboard map. It boots two tiny counter workloads from separate scratch qcow2
disks, connects two fullscreen viewers, and checks reserved switching/recovery,
manual USB selection, explicit reconnect and independent QMP lifecycles.
After each confirmed workspace switch, the hidden guest is sampled across a
half-second interval with no focus changes, switches or reconnects; the check is
then repeated with the other guest hidden. `result.json` records both intervals.
Guest memory counters must advance and QEMU PIDs remain unchanged through
switching. Viewer protocol logs must show no reconnection during switching.
A test-only `ToggleKeybinds` action and F12 marker establish native labwc
keybinding inhibition before checking the production reserved bindings. Those
extra controls are absent from the production session configuration. See the
[labwc configuration manual](https://labwc.github.io/labwc-config.5.html) for
`overrideInhibition` and the [actions manual](https://labwc.github.io/labwc-actions.5.html)
for native `ToggleKeybinds` behavior.
USB acceptance checks the actual selector window title, holds the modifiers
through opening, releases them inside the modal dialog, then asserts fresh
keyboard leave/enter transitions for both Right and Left switches.
QEMU's USB device list must stay unchanged without a selection. Both viewer
USB filters deny automatic hotplug and connection redirection. This check does
not attach a physical USB device or prove guest driver behavior.
The viewer adaptation handles only the configured manual USB accelerator before
forwarding an event to the guest. Upstream keyboard grabs remove the GTK
application accelerator group; without this adaptation, the configured USB
shortcut reaches the viewer but does not open its selector.
The pinned labwc 0.20.2 runtime has no documented runtime XWayland-disable
option. This native-Wayland probe sets `WLR_XWAYLAND=/bin/false` so no X server
is launched, and explicitly disables activation-environment updates. It can
still log an occupied X socket while wlroots reserves a display number. Only
`[xwayland/sockets.c] Failed to bind socket ... Address already in use` is
classified as expected noise in `result.json`; any other compositor `[ERROR]`
fails the probe. Full logs remain available. This classification does not
suppress viewer errors or compiler/test diagnostics. The viewer build's upstream
Meson boolean-check and compiler-warning-option guidance are build noise; the four-test
result, compiler errors and process exit status remain the build acceptance gate.
## Fresh-session recovery and profile changes
A session with no profiles opens recovery with first-use guidance. Enter `:shell`
to run `hydra add NAME --iso PATH` and `hydra start NAME`; leave the shell and enter
`:refresh`. After adding or removing profiles, refresh reloads the profile list,
rebuilds workspace rules, restarts only the session's own compositor, and attaches
new isolated viewers to running guests. QEMU processes and workloads survive.
The recovery prompt can then reconnect the new names; reserved Left/Right chords
switch between their workspaces. `:quit` closes the recovery prompt.
`dev/session-refresh-smoke.py` exercises the actual product launch loop, recovery
helper, `:shell` and `:refresh`, starting with an empty profile directory. A
test-only labwc wrapper selects private headless backends; a terminal command
feeder drives recovery and a scratch shell adds and starts two counter guests.
Actual labwc and native Wayland viewers then demonstrate fresh keyboard focus
transitions in both directions, hidden workload continuity, unchanged QEMU
PIDs/UUIDs through refresh, and guest survival after normal compositor exit.
This checks the recovery command flow, not physical terminal rendering/usability.
It leaves snapshots, recovery output, protocol logs and `result.json` in
`/tmp/hydra-session-refresh-*` and stops only its verified scratch QEMU processes.
## Bridge identities
New profiles persist a locally administered unicast `mac` in their TOML file.
It remains stable across stop/start, media changes and runtime socket-directory
changes. The generated address incorporates the VM disk path and user identity,
so separate profiles/users avoid QEMU's identical default MAC on a shared bridge.
Existing profiles without `mac` use the same deterministic calculation without
rewriting their file. Relocating a legacy disk or changing its owner changes
that fallback; explicitly persisting its current address retains the identity.
Explicit TOML addresses must also be locally administered unicast, and users
copying profiles must assign distinct addresses to concurrently used copies.
`dev/check-network.py` requires unprivileged user/network/mount namespaces. It
asserts all three namespace identities changed before creating a private
`hydrabr0` and mounting a private QEMU bridge-helper ACL over `/etc` within that
namespace. Two actual scratch QEMUs must report different configured NIC MACs,
both bridge TAP ports must exist, and restart must preserve the first guest's
address. Evidence remains under `/tmp/hydra-network-*`; host bridges, host `/etc`
and existing guests are unchanged. This does not prove guest network drivers,
DHCP leases or external connectivity; those need the installed-guest harness.
## Installed guest workflow
`--iso` is optional. Attach or detach media only while the VM is stopped:
```sh
bin/hydra add guest --iso /absolute/path/installer.iso
bin/hydra start guest
bin/hydra view guest
# Shut down from inside the guest, then check it is stopped.
bin/hydra status guest
bin/hydra media guest --detach
bin/hydra start guest
```
For an existing stopped guest, `bin/hydra media guest --iso PATH` attaches
installation/rescue media. A missing configured ISO prevents a new start with
an actionable error; it does not prevent listing profiles, status, stopping a
running guest, reconnecting a viewer or detaching the missing media. Media
updates share the VM lifecycle lock and preserve its qcow2 disk. Abrupt stopping
still requires `bin/hydra stop guest --force` and verified QMP identity.
## Display preferences and EDID
New developer profiles use user networking, 1280×800 and `refresh_millihz=0`.
Zero omits the patched QEMU refresh property and remains compatible with stock
QEMU. Hydra's installed defaults can use this flat `/etc/hydra/defaults.toml`:
```toml
network = "bridge"
bridge = "hydrabr0"
width = 1920
height = 1080
refresh_millihz = 144000
```
Only `add` reads distribution defaults. Explicit `--network`, `--bridge`,
`--width`, `--height` and `--refresh-millihz` options win; existing profiles do
not change. `--defaults PATH` selects another preferences file for development.
Missing defaults use developer preferences; unknown or invalid fields fail
before creating a profile. Dimensions are integers from 1 to 16384, refresh is
zero for automatic refresh or an integer from 50000 to 255000 mHz, matching
the vertical range descriptor limits. This bound does not establish every
resolution/refresh combination; pixel-clock and horizontal timing limits are
mode-dependent and the verified target remains 1920×1080/144000 mHz. Nonzero
refresh requires the tracked patched QEMU provided by `dev/tools.nix` and distribution packaging.
The patch keeps the requested refresh through frontend resizing and extends
EDID monitor ranges to include the generated preferred mode. To verify the
actual patched QEMU generator with the pinned source archive:
```sh
qemu_source=$(nix-build dev/tools.nix -A qemu.src --no-out-link)
nix-shell dev/tools.nix --run "python3 dev/check-edid.py '$qemu_source'"
```
The probe applies the tracked patch to four relevant source files in a temporary
directory, compiles the actual EDID and virtio GPU generator functions with a
small dependency harness, and decodes the result. It checks 60/75 Hz fallback,
144 Hz configuration, a conflicting 60 Hz frontend request, resized dimensions,
checksums and monitor ranges. It also checks the supported vertical-refresh
boundaries: 1920×1080/50000 mHz and 1280×800/255000 mHz. Range endpoints
use floor/ceiling of the encoded timing after pixel-clock rounding, so the
49.998895 Hz lower-boundary timing is included by a 49 Hz minimum. Normal
automatic 60/75 Hz cases retain the legacy 50–125 Hz range. At 1920×1080/144000 mHz the preferred timing is
143.997231 Hz after EDID clock rounding. This is an advertised guest mode,
not measured frame delivery or proof of a 144 Hz physical output.
## Still requiring a dedicated physical session
- Physical reserved chords, brightness/backlight permissions and touchpad input.
- Physical held keys/modifiers across guest switching, grabs and modal dialogs;
guest key-up delivery, pointer grab release and absence of stuck modifiers.
- Physical USB selection/redirection and guest drivers; no automatic redirection.
- Toolbar behavior at physical screen edges and actual recovery shell usability.
- ASUS output selection at 1920×1080/144 Hz, guest-reported display mode, delivered
frame rate, hidden guest graphics continuity, tearing and switching latency.
Headless checks establish CPU workload continuity, protocol focus transitions
and the exercised compositor actions. They cannot prove physical input or scanout.
+1
View File
@@ -0,0 +1 @@
"""HydraOS independent VM management."""
+92
View File
@@ -0,0 +1,92 @@
"""Portable source-checkout command line interface."""
import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from .profiles import Profile, load_defaults
from .runtime import Runtime, RuntimeError
from .session import launch_session, session_helper, view
def parser() -> argparse.ArgumentParser:
cli = argparse.ArgumentParser(prog='hydra', description='Independent HydraOS VM manager')
cli.add_argument('--config', type=Path, default=Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'hydra')
cli.add_argument('--state', type=Path, default=Path(os.environ.get('XDG_STATE_HOME', Path.home() / '.local/state')) / 'hydra')
cli.add_argument('--runtime', type=Path, default=Path(os.environ.get('XDG_RUNTIME_DIR', f'/tmp/hydra-{os.getuid()}')) / 'hydra')
cli.add_argument('--defaults', type=Path, default=Path('/etc/hydra/defaults.toml'),
help='Distribution preferences for newly created profiles')
commands = cli.add_subparsers(dest='command', required=True)
add = commands.add_parser('add', help='Create a VM profile without starting it')
add.add_argument('name')
add.add_argument('--iso', type=Path)
add.add_argument('--memory', type=int, default=4096)
add.add_argument('--cpus', type=int, default=2)
add.add_argument('--disk-size', default='40G')
add.add_argument('--graphics', choices=('auto', 'software'), default='auto')
add.add_argument('--network', choices=('user', 'bridge'))
add.add_argument('--bridge')
add.add_argument('--width', type=int)
add.add_argument('--height', type=int)
add.add_argument('--refresh-millihz', type=int)
media = commands.add_parser('media', help='Attach or detach installation media while stopped')
media.add_argument('name')
selection = media.add_mutually_exclusive_group(required=True)
selection.add_argument('--iso', type=Path)
selection.add_argument('--detach', action='store_true')
commands.add_parser('list', help='List configured VMs')
status = commands.add_parser('status', help='Query VM state through private QMP')
status.add_argument('name', nargs='?')
for command in ('start', 'shutdown', 'stop'):
cmd = commands.add_parser(command)
cmd.add_argument('name')
if command == 'stop':
cmd.add_argument('--force', action='store_true', required=True)
viewer = commands.add_parser('view', help='Connect a viewer without changing VM power')
viewer.add_argument('name')
viewer.add_argument('--windowed', action='store_true')
session = commands.add_parser('session', help='Start the minimal HydraOS labwc session')
session.add_argument('--nested', action='store_true')
helper = commands.add_parser('_session-helper', help=argparse.SUPPRESS)
helper.add_argument('operation')
helper.add_argument('manifest', type=Path)
return cli
def main(argv: list[str] | None = None) -> int:
args = parser().parse_args(argv)
runtime = Runtime(args.config, args.state, args.runtime)
try:
if args.command == 'view':
return view(runtime, args.name, windowed=args.windowed)
elif args.command == 'session':
return launch_session(runtime, nested=args.nested)
elif args.command == '_session-helper':
return session_helper(args.operation, args.manifest)
elif args.command == 'add':
preferences = load_defaults(args.defaults)
preferences.update({key: getattr(args, key) for key in
('network', 'bridge', 'width', 'height', 'refresh_millihz')
if getattr(args, key) is not None})
profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size,
args.graphics, **preferences)
runtime.add(profile)
result = {'name': profile.name, 'status': 'added'}
elif args.command == 'media':
result = runtime.media(args.name, None if args.detach else args.iso)
elif args.command == 'list':
result = [{'name': profile.name, 'iso': str(profile.iso) if profile.iso else None} for profile in runtime.profiles()]
elif args.command == 'status':
result = runtime.status(args.name) if args.name else [runtime.status(p.name) for p in runtime.profiles()]
elif args.command == 'stop':
result = runtime.stop(args.name, force=args.force)
else:
result = getattr(runtime, args.command)(args.name)
print(json.dumps(result, indent=2))
return 0
except (RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc:
print(f'hydra: {exc}', file=sys.stderr)
return 1
+104
View File
@@ -0,0 +1,104 @@
"""Validated, independently editable VM profiles."""
import json
import re
import tomllib
from dataclasses import dataclass
from pathlib import Path
def validate_name(name: str) -> str:
if not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', name):
raise ValueError('VM name must match [a-z][a-z0-9-]{0,31}')
return name
@dataclass(frozen=True)
class Profile:
name: str
iso: Path | None = None
memory: int = 4096
cpus: int = 2
disk_size: str = '40G'
graphics: str = 'auto'
network: str = 'user'
bridge: str = 'hydrabr0'
width: int = 1280
height: int = 800
refresh_millihz: int = 0
mac: str | None = None
def __post_init__(self):
validate_name(self.name)
if self.iso is not None:
if not isinstance(self.iso, (str, Path)) or not str(self.iso):
raise ValueError('ISO must be a nonempty path or omitted')
object.__setattr__(self, 'iso', Path(self.iso).expanduser().resolve())
if self.mac is not None:
if (not isinstance(self.mac, str) or
not re.fullmatch(r'[0-9a-fA-F]{2}(:[0-9a-fA-F]{2}){5}', self.mac) or
int(self.mac[:2], 16) & 3 != 2):
raise ValueError('mac must be a locally administered unicast Ethernet address')
object.__setattr__(self, 'mac', self.mac.lower())
for key, upper in (('memory', 262144), ('cpus', 256)):
value = getattr(self, key)
if type(value) is not int or not 1 <= value <= upper:
raise ValueError(f'{key} must be an integer between 1 and {upper}')
for key in ('width', 'height'):
if type(getattr(self, key)) is not int or not 1 <= getattr(self, key) <= 16384:
raise ValueError(f'{key} must be an integer between 1 and 16384')
if type(self.refresh_millihz) is not int or not (self.refresh_millihz == 0 or
50000 <= self.refresh_millihz <= 255000):
raise ValueError('refresh_millihz must be 0 (automatic) or an integer between 50000 and 255000')
if not isinstance(self.disk_size, str) or not re.fullmatch(r'[1-9][0-9]{0,5}[MGT]', self.disk_size):
raise ValueError('disk_size must be a positive size with M, G or T suffix')
if self.graphics not in ('auto', 'software'):
raise ValueError('graphics must be auto or software')
if self.network not in ('user', 'bridge'):
raise ValueError('network must be user or bridge')
if not isinstance(self.bridge, str) or not re.fullmatch(r'[a-zA-Z0-9_-]{1,15}', self.bridge):
raise ValueError('bridge must be a valid Linux interface name')
def load_profile(path: Path) -> Profile:
path = Path(path)
with path.open('rb') as stream:
data = tomllib.load(stream)
unknown = data.keys() - Profile.__dataclass_fields__.keys()
if unknown:
raise ValueError(f'Unknown profile fields: {", ".join(sorted(unknown))}')
name = data.pop('name', path.stem)
if name != path.stem:
raise ValueError('Profile name must match its filename')
try:
return Profile(name=name, **data)
except TypeError as exc:
raise ValueError(f'Invalid profile: {exc}') from exc
def save_profile(path: Path, profile: Profile) -> None:
path = Path(path)
if path.stem != profile.name:
raise ValueError('Profile name must match its filename')
lines = []
for field in Profile.__dataclass_fields__:
value = getattr(profile, field)
if value is None:
continue
lines.append(f'{field} = {value if type(value) is int else json.dumps(str(value), ensure_ascii=False)}\n')
with path.open('x', encoding='utf-8') as stream:
stream.write(''.join(lines))
def load_defaults(path: Path) -> dict:
"""Distribution preferences apply only when creating a new profile."""
try:
with Path(path).open('rb') as stream:
data = tomllib.load(stream)
except FileNotFoundError:
return {}
unknown = data.keys() - {'network', 'bridge', 'width', 'height', 'refresh_millihz'}
if unknown:
raise ValueError(f'Unknown defaults fields: {", ".join(sorted(unknown))}')
Profile('defaults-check', **data)
return data
+91
View File
@@ -0,0 +1,91 @@
"""Bounded JSON QMP over private Unix sockets, with reply ID matching."""
import json
import socket
import time
from pathlib import Path
class QMPError(RuntimeError):
pass
class QMPClient:
def __init__(self, path: Path, timeout: float = 2.0):
if timeout <= 0:
raise ValueError('QMP timeout must be positive')
self.timeout = timeout
self.socket = socket.socket(socket.AF_UNIX)
self.buffer = bytearray()
self.counter = 0
deadline = time.monotonic() + timeout
try:
self.socket.settimeout(timeout)
self.socket.connect(str(path))
greeting = self._receive(deadline)
if not isinstance(greeting.get('QMP'), dict):
raise QMPError('Invalid QMP greeting')
self._command('qmp_capabilities', None, deadline)
except (OSError, QMPError) as exc:
self.close()
if isinstance(exc, QMPError):
raise
raise QMPError(f'QMP connection failed: {exc}') from exc
def _receive(self, deadline: float) -> dict:
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise QMPError('QMP deadline exceeded')
if b'\n' in self.buffer:
line, _, rest = self.buffer.partition(b'\n')
self.buffer = bytearray(rest)
try:
message = json.loads(line)
except (ValueError, UnicodeDecodeError) as exc:
raise QMPError('Malformed QMP JSON') from exc
if not isinstance(message, dict):
raise QMPError('QMP response must be an object')
return message
self.socket.settimeout(remaining)
try:
chunk = self.socket.recv(65536)
except OSError as exc:
raise QMPError(f'QMP receive failed: {exc}') from exc
if not chunk:
raise QMPError('QMP connection closed')
self.buffer.extend(chunk)
if len(self.buffer) > 1048576:
raise QMPError('QMP message exceeds 1 MiB')
def _command(self, name: str, arguments: dict | None, deadline: float) -> dict:
self.counter += 1
request = {'execute': name, 'id': self.counter}
if arguments is not None:
request['arguments'] = arguments
try:
self.socket.settimeout(max(0.001, deadline - time.monotonic()))
self.socket.sendall(json.dumps(request).encode() + b'\r\n')
except OSError as exc:
raise QMPError(f'QMP send failed: {exc}') from exc
while True:
response = self._receive(deadline)
if response.get('id') != self.counter:
continue
if 'error' in response:
raise QMPError(f'QMP {name}: {response["error"]}')
if 'return' not in response:
raise QMPError('QMP reply has no return value')
return response['return']
def command(self, name: str, arguments: dict | None = None) -> dict:
return self._command(name, arguments, time.monotonic() + self.timeout)
def close(self):
self.socket.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
+342
View File
@@ -0,0 +1,342 @@
"""Independent QEMU lifecycle; all destructive control requires QMP identity."""
import contextlib
import errno
import fcntl
import hashlib
import json
import os
import socket
import stat
import subprocess
import tempfile
import time
import uuid
from dataclasses import dataclass, replace
from pathlib import Path
from .profiles import Profile, load_profile, save_profile, validate_name
from .qmp import QMPClient, QMPError
class RuntimeError(Exception):
"""A VM lifecycle operation could not be safely completed."""
@dataclass(frozen=True)
class VMPaths:
name: str
disk: Path
log: Path
lock: Path
qmp: Path
spice: Path
pid: Path
metadata: Path
@property
def uuid(self) -> str:
return str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.disk)))
def _qemu_value(value: Path | str) -> str:
# QEMU's key/value parser uses doubled commas for literal commas.
return str(value).replace(',', ',,')
def _intel_render_node() -> Path | None:
for node in sorted(Path('/dev/dri').glob('renderD*')):
vendor = Path('/sys/class/drm') / node.name / 'device/vendor'
try:
if vendor.read_text().strip() == '0x8086' and os.access(node, os.R_OK | os.W_OK):
return node
except OSError:
continue
return None
def _stable_mac(paths: VMPaths) -> str:
# Disk/state identity survives restarts and runtime socket-directory changes.
# UID distinguishes owners even when a shared bridge sees the same names.
digest = bytearray(hashlib.sha256(f'{os.getuid()}:{paths.disk}'.encode()).digest()[:6])
digest[0] = (digest[0] & 0xfc) | 0x02
return ':'.join(f'{octet:02x}' for octet in digest)
def build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]:
accel = 'kvm' if os.access('/dev/kvm', os.R_OK | os.W_OK) else 'tcg'
argv = [os.environ.get('HYDRA_QEMU', 'qemu-system-x86_64'), '-name', profile.name,
'-uuid', paths.uuid, '-machine', f'q35,accel={accel}',
'-m', str(profile.memory), '-smp', str(profile.cpus),
'-daemonize', '-pidfile', str(paths.pid), '-nodefaults',
'-chardev', f'socket,id=qmp,path={_qemu_value(paths.qmp)},server=on,wait=off',
'-mon', 'chardev=qmp,mode=control']
if accel == 'kvm':
argv += ['-cpu', 'host']
blockdevs = [
{'driver': 'file', 'filename': str(paths.disk), 'node-name': 'disk-file'},
{'driver': 'qcow2', 'file': 'disk-file', 'node-name': 'disk'},
]
if profile.iso is not None:
blockdevs += [
{'driver': 'file', 'filename': str(profile.iso), 'node-name': 'iso-file', 'read-only': True},
{'driver': 'raw', 'file': 'iso-file', 'node-name': 'iso', 'read-only': True},
]
for options in blockdevs:
argv += ['-blockdev', json.dumps(options)]
argv += ['-device', 'virtio-blk-pci,drive=disk']
if profile.iso is not None:
argv += ['-device', 'ide-cd,drive=iso']
argv += ['-boot', 'order=dc' if profile.iso is not None else 'order=c', '-device', f'virtio-net-pci,netdev=net,mac={profile.mac or _stable_mac(paths)}',
'-netdev', 'user,id=net' if profile.network == 'user'
else f'bridge,id=net,br={profile.bridge}',
'-device', 'qemu-xhci,id=usb', '-device', 'usb-tablet', '-device', 'usb-kbd']
argv += ['-audiodev', 'spice,id=audio', '-device', 'ich9-intel-hda',
'-device', 'hda-duplex,audiodev=audio']
for channel in range(3):
argv += ['-chardev', f'spicevmc,id=usbredir{channel},name=usbredir',
'-device', f'usb-redir,chardev=usbredir{channel}']
spice = (f'unix=on,addr={_qemu_value(paths.spice)},disable-ticketing=on,'
'disable-copy-paste=on,disable-agent-file-xfer=on')
mode = f',xres={profile.width},yres={profile.height}'
if profile.refresh_millihz:
mode += f',refresh_rate={profile.refresh_millihz}'
render = _intel_render_node() if profile.graphics == 'auto' else None
if render:
spice += f',gl=on,rendernode={_qemu_value(render)}'
argv += ['-display', f'egl-headless,rendernode={_qemu_value(render)}', '-device', 'virtio-vga-gl' + mode]
else:
argv += ['-display', 'none', '-device', 'virtio-vga' + mode]
return argv + ['-spice', spice]
def _private_dir(path: Path) -> None:
if path.is_symlink():
raise RuntimeError(f'Private directory must not be a symlink: {path}')
path.mkdir(parents=True, mode=0o700, exist_ok=True)
info = path.stat()
if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
raise RuntimeError(f'Private directory is not owned by this user: {path}')
path.chmod(0o700)
class Runtime:
def __init__(self, config: Path, state: Path, runtime: Path, boot_timeout: float = 10):
# Keep final components lexical so symlink checks still see them.
self.config = Path(config).expanduser().absolute()
self.state = Path(state).expanduser().absolute()
self.runtime = Path(runtime).expanduser().absolute()
self.boot_timeout = boot_timeout
def paths(self, name: str) -> VMPaths:
validate_name(name)
state = self.state / name
run = self.runtime / name
paths = VMPaths(name, state / 'disk.qcow2', state / 'qemu.log', state / 'lock',
run / 'qmp.sock', run / 'spice.sock', run / 'qemu.pid', run / 'identity.json')
if any(len(os.fsencode(path)) >= 108 for path in (paths.qmp, paths.spice)):
raise RuntimeError('Unix socket path is too long; choose a shorter --runtime path')
return paths
def prepare(self, paths: VMPaths) -> None:
for directory in (self.config, self.state, self.runtime, paths.disk.parent, paths.qmp.parent):
_private_dir(directory)
def add(self, profile: Profile) -> None:
self._validate_media(profile.iso)
_private_dir(self.config)
if profile.mac is None:
profile = replace(profile, mac=_stable_mac(self.paths(profile.name)))
save_profile(self.config / f'{profile.name}.toml', profile)
@staticmethod
def _validate_media(iso: Path | None) -> None:
if iso is not None and not iso.is_file():
raise RuntimeError(f'ISO does not exist or is not a file: {iso}; attach valid media or detach it while stopped')
def media(self, name: str, iso: Path | None) -> dict:
paths = self.paths(name)
with self._locked(paths):
current = self.status(name)
if current['status'] != 'stopped':
raise RuntimeError(f'Cannot change media for {name}: {current["status"]}')
profile = replace(self.profile(name), iso=iso)
self._validate_media(profile.iso)
destination = self.config / f'{name}.toml'
if destination.is_symlink():
raise RuntimeError('Refusing to replace a symlink profile')
with tempfile.TemporaryDirectory(prefix=f'.{name}-', dir=self.config) as tmp:
temp = Path(tmp) / f'{name}.toml'
save_profile(temp, profile)
temp.chmod(0o600)
os.replace(temp, destination)
return {'name': name, 'iso': str(profile.iso) if profile.iso else None, 'status': 'media-updated'}
def profiles(self) -> list[Profile]:
if not self.config.exists():
return []
return [load_profile(path) for path in sorted(self.config.glob('*.toml'))]
def profile(self, name: str) -> Profile:
validate_name(name)
return load_profile(self.config / f'{name}.toml')
@contextlib.contextmanager
def _locked(self, paths: VMPaths):
self.prepare(paths)
fd = os.open(paths.lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
try:
fcntl.flock(fd, fcntl.LOCK_EX)
yield
finally:
os.close(fd)
def _owned_process_alive(self, paths: VMPaths) -> bool:
# PID is only a diagnostic hint. Never send a signal to it.
try:
pid = int(paths.pid.read_text())
args = Path(f'/proc/{pid}/cmdline').read_bytes().split(b'\0')
wanted_uuid = paths.uuid.encode()
wanted_name = paths.name.encode()
return (b'-uuid' in args and args[args.index(b'-uuid') + 1] == wanted_uuid
and b'-name' in args and args[args.index(b'-name') + 1] == wanted_name)
except (OSError, ValueError, IndexError):
return False
def _connect(self, paths: VMPaths) -> QMPClient:
try:
record = json.loads(paths.metadata.read_text())
if not isinstance(record, dict) or record.get('name') != paths.name or record.get('uuid') != paths.uuid:
raise RuntimeError('VM identity record does not match selected VM')
client = QMPClient(paths.qmp, timeout=min(2, self.boot_timeout))
try:
name_result = client.command('query-name')
uuid_result = client.command('query-uuid')
if not isinstance(name_result, dict) or not isinstance(uuid_result, dict):
raise RuntimeError('Malformed QMP identity reply')
name = name_result.get('name')
actual_uuid = uuid_result.get('UUID')
if name != paths.name or actual_uuid != record['uuid']:
raise RuntimeError('QMP VM identity mismatch; refusing control')
except Exception:
client.close()
raise
return client
except (OSError, ValueError, QMPError) as exc:
raise RuntimeError(f'Cannot verify VM identity: {exc}') from exc
def _listening(self, path: Path) -> bool:
try:
mode = path.lstat().st_mode
except FileNotFoundError:
return False
if not stat.S_ISSOCK(mode):
raise RuntimeError(f'Refusing non-socket runtime path: {path}')
with socket.socket(socket.AF_UNIX) as probe:
probe.settimeout(min(1, self.boot_timeout))
try:
probe.connect(str(path))
return True
except OSError as exc:
if exc.errno in (errno.ENOENT, errno.ECONNREFUSED):
return False
raise RuntimeError(f'Cannot establish socket ownership: {exc}') from exc
def status(self, name: str) -> dict:
paths = self.paths(name)
try:
if not self._listening(paths.qmp):
if self._owned_process_alive(paths):
return {'name': name, 'status': 'unknown', 'error': 'VM process exists without working QMP'}
return {'name': name, 'status': 'stopped'}
with self._connect(paths) as client:
status = client.command('query-status')
if not isinstance(status, dict) or not isinstance(status.get('status'), str):
raise RuntimeError('Malformed QMP status reply')
return {'name': name, **status}
except (RuntimeError, QMPError) as exc:
return {'name': name, 'status': 'unknown', 'error': str(exc)}
def _create_disk(self, profile: Profile, paths: VMPaths) -> None:
if paths.disk.exists() or paths.disk.is_symlink():
if paths.disk.is_symlink() or not paths.disk.is_file():
raise RuntimeError(f'Refusing non-regular disk: {paths.disk}')
return
fd, temporary = tempfile.mkstemp(prefix='.disk-', suffix='.qcow2', dir=paths.disk.parent)
os.close(fd)
temp = Path(temporary)
temp.unlink() # qemu-img creates a new file; final publication never replaces a disk.
try:
subprocess.run([os.environ.get('HYDRA_QEMU_IMG', 'qemu-img'), 'create', '-f', 'qcow2',
str(temp), profile.disk_size], check=True, capture_output=True,
text=True, timeout=30)
temp.chmod(0o600)
os.link(temp, paths.disk)
except (OSError, subprocess.SubprocessError) as exc:
raise RuntimeError(f'Disk creation failed: {exc}') from exc
finally:
temp.unlink(missing_ok=True)
def start(self, name: str) -> dict:
paths = self.paths(name)
with self._locked(paths):
profile = self.profile(name)
current = self.status(name)
if current['status'] != 'stopped':
raise RuntimeError(f'Cannot start {name}: {current["status"]}')
self._validate_media(profile.iso)
for path in (paths.qmp, paths.spice):
if self._listening(path):
raise RuntimeError(f'Runtime socket is already in use: {path}')
path.unlink(missing_ok=True)
self._create_disk(profile, paths)
fd, temporary = tempfile.mkstemp(prefix='.identity-', dir=paths.metadata.parent)
try:
with os.fdopen(fd, 'w') as record:
record.write(json.dumps({'name': name, 'uuid': paths.uuid}) + '\n')
os.replace(temporary, paths.metadata)
finally:
Path(temporary).unlink(missing_ok=True)
paths.pid.unlink(missing_ok=True)
try:
log_fd = os.open(paths.log, os.O_CREAT | os.O_APPEND | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
with os.fdopen(log_fd, 'ab') as log:
subprocess.run(build_qemu_argv(profile, paths), stdin=subprocess.DEVNULL,
stdout=log, stderr=log, start_new_session=True,
check=True, timeout=self.boot_timeout,
umask=0o077)
deadline = time.monotonic() + self.boot_timeout
while time.monotonic() < deadline:
current = self.status(name)
if current['status'] not in ('stopped', 'unknown'):
return current
time.sleep(0.05)
raise RuntimeError(f'QEMU did not become ready; inspect {paths.log}')
except (OSError, subprocess.SubprocessError) as exc:
raise RuntimeError(f'QEMU launch failed; inspect {paths.log}: {exc}') from exc
def shutdown(self, name: str) -> dict:
paths = self.paths(name)
try:
with self._locked(paths), self._connect(paths) as client:
client.command('system_powerdown')
except QMPError as exc:
raise RuntimeError(f'Guest shutdown request failed: {exc}') from exc
return {'name': name, 'status': 'shutdown-requested'}
def stop(self, name: str, force: bool = False) -> dict:
if not force:
raise RuntimeError('Abrupt stop requires --force; use shutdown for guest ACPI shutdown')
paths = self.paths(name)
with self._locked(paths):
try:
with self._connect(paths) as client:
client.command('quit')
except QMPError as exc:
raise RuntimeError(f'QEMU quit request failed: {exc}') from exc
deadline = time.monotonic() + self.boot_timeout
while time.monotonic() < deadline:
if self.status(name)['status'] == 'stopped':
return {'name': name, 'status': 'stopped'}
time.sleep(0.05)
raise RuntimeError('QEMU accepted quit but has not stopped within the deadline')
+294
View File
@@ -0,0 +1,294 @@
"""Minimal labwc session; viewer and compositor lifetimes never control QEMU."""
import fcntl
import json
import os
import re
import shlex
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from urllib.parse import quote
from xml.etree import ElementTree as ET
from .profiles import validate_name
from .runtime import Runtime, VMPaths, _private_dir
class SessionError(ValueError):
"""An actionable viewer/session error safe to present at the CLI."""
def viewer_title(name: str) -> str:
return 'HydraOS: ' + validate_name(name)
def build_viewer_argv(paths: VMPaths, viewer: str, *, windowed=False) -> list[str]:
argv = [viewer, '--title', viewer_title(paths.name), '--hotkeys',
'release-cursor=ctrl+alt+shift+r,usb-device-select=ctrl+alt+shift+u',
'--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0',
'--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0']
if not windowed:
argv += ['--full-screen', '--hide-toolbar']
# QEMU denies clipboard/file transfer; no shared-dir/CD flag is added here.
return argv + ['spice+unix://' + quote(str(paths.spice), safe='/')]
def build_labwc_config(names: list[str], recovery_argv: list[str],
brightness='brightnessctl', *, initial: str | None = None) -> str:
names = [validate_name(name) for name in names]
if len(set(names)) != len(names):
raise SessionError('Duplicate VM workspace name')
if initial is not None and initial not in names:
raise SessionError('Initial VM workspace must be configured')
root = ET.Element('labwc_config')
core = ET.SubElement(root, 'core')
ET.SubElement(core, 'primarySelection').text = 'no'
desktops = ET.SubElement(root, 'desktops', number=str(max(1, len(names))))
if names:
labels = ET.SubElement(desktops, 'names')
for name in names:
ET.SubElement(labels, 'name').text = name
if initial is not None:
ET.SubElement(desktops, 'initial').text = str(names.index(initial) + 1)
ET.SubElement(desktops, 'popupTime').text = '0'
theme = ET.SubElement(root, 'theme')
ET.SubElement(theme, 'keepBorder').text = 'no'
keyboard = ET.SubElement(root, 'keyboard')
for key, direction in (('C-A-S-Left', 'left'), ('C-A-S-Right', 'right')):
binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes')
ET.SubElement(binding, 'action', name='GoToDesktop', to=direction, wrap='yes')
commands = [('C-A-S-Escape', recovery_argv),
('XF86MonBrightnessUp', [brightness, '-c', 'backlight', 'set', '+5%']),
('XF86MonBrightnessDown', [brightness, '-c', 'backlight', 'set', '5%-'])]
for key, argv in commands:
binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes')
ET.SubElement(binding, 'action', name='Execute', command=shlex.join(argv))
# An empty mouse section restores labwc defaults. A binding with no action
# suppresses that fallback without consuming guest window pointer events.
context = ET.SubElement(ET.SubElement(root, 'mouse'), 'context', name='Root')
ET.SubElement(context, 'mousebind', button='Left', action='Press')
device = ET.SubElement(ET.SubElement(root, 'libinput'), 'device', category='touchpad')
ET.SubElement(device, 'tap').text = 'yes'
rules = ET.SubElement(root, 'windowRules')
for index, name in enumerate(names, 1):
rule = ET.SubElement(rules, 'windowRule', title=viewer_title(name),
serverDecoration='no')
ET.SubElement(rule, 'action', name='SendToDesktop', to=str(index), follow='no')
ET.SubElement(rule, 'action', name='SetDecorations', decorations='none')
ET.indent(root)
return ET.tostring(root, encoding='unicode', xml_declaration=True) + '\n'
def _tool(variable: str, default: str) -> str:
value = os.environ.get(variable, default)
executable = shutil.which(value)
if not executable:
raise SessionError(f'Missing {default} tool ({variable}); enter the temporary dev/tools.nix environment')
return executable
def _display() -> None:
if not os.environ.get('WAYLAND_DISPLAY') and not os.environ.get('DISPLAY'):
raise SessionError('No graphical display; start a session or use a graphical host terminal')
def _viewer() -> str:
viewer = _tool('HYDRA_VIEWER', 'remote-viewer')
result = subprocess.run([viewer, '--help'], capture_output=True, text=True, timeout=10)
if result.returncode or '--hide-toolbar' not in result.stdout:
raise SessionError('Viewer lacks HydraOS toolbar/USB adaptation; use dev/tools.nix')
return viewer
def _socket(paths: VMPaths) -> None:
try:
mode = paths.spice.lstat().st_mode
except FileNotFoundError as exc:
raise SessionError(f'{paths.name}: no SPICE socket; run hydra start {paths.name}, then retry') from exc
if not stat.S_ISSOCK(mode):
raise SessionError(f'{paths.name}: SPICE path is not a Unix socket')
def _viewer_environment(directory: Path) -> dict[str, str]:
config = directory / 'config'
settings = config / 'virt-viewer'
settings.mkdir(parents=True, mode=0o700)
(settings / 'settings').write_text('[virt-viewer]\nshare-clipboard=false\n', encoding='utf-8')
env = dict(os.environ, XDG_CONFIG_HOME=str(config), GSETTINGS_BACKEND='memory')
# A dedicated backend prevents accidental XWayland input behavior in HydraOS.
if env.get('WAYLAND_DISPLAY'):
env['GDK_BACKEND'] = 'wayland'
return env
def view(runtime: Runtime, name: str, *, windowed=False) -> int:
_display()
viewer = _viewer()
runtime.profile(name)
paths = runtime.paths(name)
_socket(paths)
_private_dir(runtime.runtime)
with tempfile.TemporaryDirectory(prefix='viewer-', dir=runtime.runtime) as tmp:
return subprocess.run(build_viewer_argv(paths, viewer, windowed=windowed),
env=_viewer_environment(Path(tmp))).returncode
def compositor_environment(environ: dict[str, str], *, nested=False) -> dict[str, str]:
env = dict(environ)
if nested:
if env.get('WAYLAND_DISPLAY'):
env['WLR_BACKENDS'] = 'wayland'
elif env.get('DISPLAY'):
env['WLR_BACKENDS'] = 'x11'
else:
raise SessionError('Nested session requires an existing graphical display')
else:
env.pop('WAYLAND_DISPLAY', None)
env.pop('DISPLAY', None)
env['WLR_BACKENDS'] = 'drm,libinput'
env['XDG_CURRENT_DESKTOP'] = 'HydraOS'
return env
def _module_argv(command: str, manifest: Path) -> list[str]:
# Absolute source launcher also works when labwc changes the working directory.
return [sys.executable, str(Path(__file__).resolve().parents[1] / 'bin/hydra'),
'_session-helper', command, str(manifest)]
def _recovery_argv(manifest: Path, terminal: str) -> list[str]:
return [terminal, '--title=HydraOS-recovery'] + _module_argv('recover', manifest)
def launch_session(runtime: Runtime, *, nested=False) -> int:
env = compositor_environment(dict(os.environ), nested=nested)
labwc = _tool('HYDRA_LABWC', 'labwc')
terminal = _tool('HYDRA_TERMINAL', 'foot')
brightness = _tool('HYDRA_BRIGHTNESS', 'brightnessctl')
version = subprocess.run([labwc, '--version'], capture_output=True, text=True, timeout=10)
match = re.search(r'(\d+)\.(\d+)\.(\d+)', version.stdout)
if version.returncode or not match or tuple(map(int, match.groups())) < (0, 20, 0):
raise SessionError('HydraOS requires labwc >= 0.20.0 for reserved host input bindings')
_private_dir(runtime.runtime)
while True:
profiles = runtime.profiles()
initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None)
with tempfile.TemporaryDirectory(prefix='session-', dir=runtime.runtime) as tmp:
directory = Path(tmp)
config = directory / 'labwc'
config.mkdir(mode=0o700)
manifest = directory / 'session.json'
manifest.write_text(json.dumps({'config': str(runtime.config), 'state': str(runtime.state),
'runtime': str(runtime.runtime), 'names': [p.name for p in profiles],
'terminal': terminal, 'directory': str(directory)}), encoding='utf-8')
(config / 'rc.xml').write_text(build_labwc_config([p.name for p in profiles],
_recovery_argv(manifest, terminal), brightness,
initial=initial), encoding='utf-8')
(config / 'autostart').write_text('#!/bin/sh\n' + shlex.join(_module_argv('clients', manifest)) + ' &\n', encoding='utf-8')
(config / 'environment').touch()
env['PYTHONUNBUFFERED'] = '1'
compositor = subprocess.Popen([labwc, '-C', str(config)], env=env)
refresh = False
try:
while True:
if (directory / 'refresh.request').exists():
refresh = True
break
if compositor.poll() is not None:
# A requesting recovery terminal can exit before polling.
refresh = (directory / 'refresh.request').exists()
break
time.sleep(.1)
finally:
if compositor.poll() is None:
compositor.terminate()
compositor.wait(timeout=10)
if not refresh:
return compositor.returncode
# Only our compositor was stopped. Its viewers disconnect; detached
# QEMU instances retain their identity and running workloads.
def session_helper(command: str, manifest: Path) -> int:
data = json.loads(manifest.read_text(encoding='utf-8'))
if (not isinstance(data, dict) or
any(not isinstance(data.get(key), str) or not Path(data[key]).is_absolute()
for key in ('config', 'state', 'runtime', 'terminal')) or
not isinstance(data.get('names'), list)):
raise SessionError('Invalid session manifest')
for name in data['names']:
validate_name(name)
if command not in ('clients', 'recover', 'refresh') and (
not command.startswith('reconnect:') or command.split(':', 1)[1] not in data['names']):
raise SessionError('Invalid session helper operation')
runtime = Runtime(Path(data['config']), Path(data['state']), Path(data['runtime']))
if command == 'refresh':
fd = os.open(manifest.parent / 'refresh.request', os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
os.close(fd)
print('Refreshing workspaces and viewers; VMs keep running.')
return 0
if command == 'recover':
print('HydraOS local recovery. VMs keep running independently.\n'
'Ctrl+Alt+Shift+Left/Right switches workspaces; Ctrl+Alt+Shift+R releases input; Ctrl+Alt+Shift+U opens USB.\n'
'Enter a VM name to reconnect, :shell for a host shell, :refresh to rebuild workspaces/viewers, or :quit.')
if not data['names']:
print('No VM profiles yet. Use :shell to run hydra add NAME --iso PATH and hydra start NAME.\n'
'After adding or removing profiles, leave the shell and enter :refresh; running VMs are preserved.')
while True:
try:
print(json.dumps([runtime.status(name) for name in data['names']], indent=2))
choice = input('hydra recovery> ').strip()
except (EOFError, KeyboardInterrupt):
return 0
if choice == ':quit':
return 0
if choice == ':refresh':
return session_helper('refresh', manifest)
if choice == ':shell':
subprocess.run([os.environ.get('SHELL', '/bin/sh')])
elif choice in data['names']:
subprocess.Popen(_module_argv('reconnect:' + choice, manifest), start_new_session=True)
else:
print('Enter a configured VM name, :shell, :refresh, or :quit.')
if command.startswith('reconnect:'):
name = command.split(':', 1)[1]
# One viewer per VM inside this session; independent host view commands
# retain their separate lifecycle and never share VM manager locks.
with (manifest.parent / (name + '.viewer.lock')).open('a') as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print(f'{name}: viewer is already connected; close recovery or switch workspace')
return 0
return view(runtime, name)
processes = []
recovery = None
def open_recovery():
nonlocal recovery
if recovery is None or recovery.poll() is not None:
recovery = subprocess.Popen(_recovery_argv(manifest, data['terminal']))
for name in data['names']:
try:
if runtime.status(name)['status'] == 'running':
processes.append(subprocess.Popen(_module_argv('reconnect:' + name, manifest)))
except (OSError, ValueError) as exc:
print(f'hydra: {name}: {exc}', file=sys.stderr)
if not processes:
open_recovery()
# No signals to QEMU or viewers: compositor disconnect tears down clients,
# while the VM manager's detached QEMU processes retain their lifetime.
while processes or (recovery is not None and recovery.poll() is None):
for process in processes[:]:
code = process.poll()
if code is not None:
processes.remove(process)
print(f'hydra: viewer exited ({code}); reconnect from local recovery', file=sys.stderr)
open_recovery()
time.sleep(0.2)
return 0
+74
View File
@@ -0,0 +1,74 @@
--- a/include/hw/virtio/virtio-gpu.h
+++ b/include/hw/virtio/virtio-gpu.h
@@ -128,6 +128,7 @@
uint32_t flags;
uint32_t xres;
uint32_t yres;
+ uint32_t refresh_rate;
uint64_t hostmem;
VirtIOGPUOutputList *outputs;
};
@@ -173,7 +174,8 @@
DEFINE_PROP_BIT("edid", _state, _conf.flags, \
VIRTIO_GPU_FLAG_EDID_ENABLED, true), \
DEFINE_PROP_UINT32("xres", _state, _conf.xres, 1280), \
- DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800)
+ DEFINE_PROP_UINT32("yres", _state, _conf.yres, 800), \
+ DEFINE_PROP_UINT32("refresh_rate", _state, _conf.refresh_rate, 0)
typedef struct VGPUDMABuf {
QemuDmaBuf *buf;
--- a/hw/display/virtio-gpu-base.c
+++ b/hw/display/virtio-gpu-base.c
@@ -64,7 +64,9 @@
.height_mm = g->req_state[scanout].height_mm,
.prefx = g->req_state[scanout].width,
.prefy = g->req_state[scanout].height,
- .refresh_rate = g->req_state[scanout].refresh_rate,
+ /* An explicit monitor refresh rate, in mHz, survives UI resizes. */
+ .refresh_rate = g->conf.refresh_rate ? g->conf.refresh_rate :
+ g->req_state[scanout].refresh_rate,
};
for (output_idx = 0, node = g->conf.outputs;
--- a/hw/display/edid-generate.c
+++ b/hw/display/edid-generate.c
@@ -220,17 +220,20 @@
desc[5 + len] = '\n';
}
-static void edid_desc_ranges(uint8_t *desc)
+static void edid_desc_ranges(uint8_t *desc, uint64_t clock,
+ uint32_t htotal, uint32_t vtotal)
{
edid_desc_type(desc, 0xfd);
- /* vertical (50 -> 125 Hz) */
- desc[5] = 50;
- desc[6] = 125;
-
- /* horizontal (30 -> 160 kHz) */
- desc[7] = 30;
- desc[8] = 160;
+ /* Enclose the encoded timing after its 10 kHz pixel-clock rounding,
+ * while retaining legacy ranges whenever they already contain it. */
+ uint64_t vperiod = (uint64_t)htotal * vtotal;
+ uint64_t hperiod = (uint64_t)htotal * 1000;
+ desc[5] = MIN(50, clock / vperiod);
+ desc[6] = MIN(255, MAX(125, DIV_ROUND_UP(clock, vperiod)));
+
+ desc[7] = MIN(30, clock / hperiod);
+ desc[8] = MIN(255, MAX(160, DIV_ROUND_UP(clock, hperiod)));
/* max dot clock (2550 MHz) */
desc[9] = 2550 / 10;
@@ -513,7 +516,8 @@
* so dta descriptor offsets don't move any more.
*/
- edid_desc_ranges(desc);
+ edid_desc_ranges(desc, (uint64_t)timings.clock * 10000,
+ info->prefx + timings.xblank, info->prefy + timings.yblank);
desc = edid_desc_next(edid, dta, desc);
if (desc && info->name) {
+147
View File
@@ -0,0 +1,147 @@
--- a/src/virt-viewer-app.c
+++ b/src/virt-viewer-app.c
@@ -134,6 +134,7 @@
gboolean authretry;
gboolean started;
gboolean fullscreen;
+ gboolean hide_toolbar;
gboolean attach;
gboolean shared;
gboolean quitting;
@@ -952,6 +953,13 @@
gchar *subtitle = NULL;
const gchar *title = virt_viewer_app_get_title(app);
+ /* Hidden fullscreen windows use the requested title as stable identity. */
+ if (title != NULL && virt_viewer_app_get_hide_toolbar(app) &&
+ virt_viewer_app_get_fullscreen(app)) {
+ g_object_set(window, "subtitle", title, NULL);
+ return;
+ }
+
if (title != NULL) {
VirtViewerDisplay *display = virt_viewer_window_get_display(window);
gchar *d = strstr(title, "%d");
@@ -2134,6 +2142,7 @@
static gboolean opt_verbose = FALSE;
static gboolean opt_debug = FALSE;
static gboolean opt_fullscreen = FALSE;
+static gboolean opt_hide_toolbar = FALSE;
static gboolean opt_kiosk = FALSE;
static gboolean opt_kiosk_quit = FALSE;
static gchar *opt_cursor = NULL;
@@ -2437,6 +2446,7 @@
{ "smartcard-remove", "app.smartcard-remove", {"<Shift>F9", NULL, NULL} },
{ "secure-attention", "win.secure-attention", {"<Ctrl><Alt>End", NULL, NULL} },
{ "usb-device-reset", "win.usb-device-reset", {"<Ctrl><Shift>r", NULL, NULL} },
+ { "usb-device-select", "win.usb-device-select", {NULL, NULL, NULL} },
};
static gchar **hotkey_names;
@@ -2484,6 +2494,7 @@
priv->resource = virt_viewer_get_resource();
virt_viewer_app_set_debug(opt_debug);
+ priv->hide_toolbar = opt_hide_toolbar;
virt_viewer_app_set_fullscreen(self, opt_fullscreen);
virt_viewer_app_set_keymap(self, opt_keymap);
@@ -3098,6 +3109,14 @@
}
gboolean
+virt_viewer_app_get_hide_toolbar(VirtViewerApp *self)
+{
+ VirtViewerAppPrivate *priv = virt_viewer_app_get_instance_private(self);
+
+ return priv->hide_toolbar;
+}
+
+gboolean
virt_viewer_app_get_fullscreen(VirtViewerApp *self)
{
g_return_val_if_fail(VIRT_VIEWER_IS_APP(self), FALSE);
@@ -3557,6 +3576,8 @@
N_("Zoom level of window, in percentage"), "ZOOM" },
{ "full-screen", 'f', 0, G_OPTION_ARG_NONE, &opt_fullscreen,
N_("Open in full screen mode (adjusts guest resolution to fit the client)"), NULL },
+ { "hide-toolbar", '\0', 0, G_OPTION_ARG_NONE, &opt_hide_toolbar,
+ N_("Hide the toolbar in full screen mode"), NULL },
{ "hotkeys", 'H', 0, G_OPTION_ARG_STRING, &opt_hotkeys,
N_("Customise hotkeys"), NULL },
{ "auto-resize", '\0', 0, G_OPTION_ARG_STRING, &opt_resize,
--- a/src/virt-viewer-app.h
+++ b/src/virt-viewer-app.h
@@ -100,6 +100,7 @@
GList* virt_viewer_app_get_windows(VirtViewerApp *self);
VirtViewerSession* virt_viewer_app_get_session(VirtViewerApp *self);
gboolean virt_viewer_app_get_fullscreen(VirtViewerApp *app);
+gboolean virt_viewer_app_get_hide_toolbar(VirtViewerApp *app);
void virt_viewer_app_clear_hotkeys(VirtViewerApp *app);
GList* virt_viewer_app_get_initial_displays(VirtViewerApp* self);
gint virt_viewer_app_get_initial_monitor_for_display(VirtViewerApp* self, gint display);
--- a/src/virt-viewer-window.c
+++ b/src/virt-viewer-window.c
@@ -740,9 +740,10 @@
return;
}
- if (!self->kiosk) {
- virt_viewer_timed_revealer_force_reveal(self->revealer, TRUE);
- }
+ gboolean hide_toolbar = virt_viewer_app_get_hide_toolbar(self->app);
+ gtk_widget_set_visible(GTK_WIDGET(self->revealer), !hide_toolbar);
+ virt_viewer_timed_revealer_force_reveal(self->revealer,
+ !self->kiosk && !hide_toolbar);
if (self->display) {
virt_viewer_display_set_monitor(self->display, monitor);
@@ -1354,6 +1355,13 @@
title = g_strdup(g_get_application_name());
}
+ /* Keep exact workspace identity across pointer/keyboard grab changes. */
+ if (self->app && virt_viewer_app_get_hide_toolbar(self->app) &&
+ virt_viewer_app_get_fullscreen(self->app) && self->subtitle) {
+ g_free(title);
+ title = g_strdup(self->subtitle);
+ }
+
gtk_window_set_title(GTK_WINDOW(self->window), title);
if (self->subtitle) {
gtk_header_bar_set_title(GTK_HEADER_BAR(header), self->subtitle);
@@ -1467,6 +1475,34 @@
display = self->display;
event = (GdkEventKey *)ev;
+ /* Keyboard grabs remove GTK's application accelerator group so normal
+ * shortcuts reach the guest. Keep only the explicitly configured manual
+ * USB selector available before forwarding this event to the display. */
+ if (!self->kiosk) {
+ gchar **accels = gtk_application_get_accels_for_action(
+ GTK_APPLICATION(self->app), "win.usb-device-select");
+ gboolean matched = FALSE;
+ for (guint i = 0; accels[i]; i++) {
+ guint key;
+ GdkModifierType mods;
+ gtk_accelerator_parse(accels[i], &key, &mods);
+ if (key && gdk_keyval_to_lower(event->keyval) == key &&
+ (event->state & gtk_accelerator_get_default_mod_mask()) == mods) {
+ matched = TRUE;
+ break;
+ }
+ }
+ g_strfreev(accels);
+ if (matched) {
+ GAction *action = g_action_map_lookup_action(
+ G_ACTION_MAP(self->window), "usb-device-select");
+ if (action && g_action_get_enabled(action)) {
+ g_action_activate(action, NULL);
+ return TRUE;
+ }
+ }
+ }
+
gtk_widget_grab_focus(GTK_WIDGET(display));
// Look through keymaps - if set for mappings and intercept
+4
View File
@@ -0,0 +1,4 @@
#!/bin/sh
# Portable source launcher. Desktop/session packaging may call this directly.
HYDRA_CHECKOUT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) || exit 1
exec "$HYDRA_CHECKOUT/bin/hydra" session "$@"
+66
View File
@@ -0,0 +1,66 @@
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from hydra.profiles import load_profile
class CLITests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hc-')
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.defaults = self.root / 'defaults.toml'
self.base = [sys.executable, 'bin/hydra', '--config', str(self.root / 'config'),
'--state', str(self.root / 'state'), '--runtime', str(self.root / 'run'),
'--defaults', str(self.defaults)]
def run_cli(self, *args):
return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5)
def test_defaults_affect_new_profiles_only_and_explicit_options_win(self):
self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n')
added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0')
self.assertEqual(added.returncode, 0, added.stderr)
first = load_profile(self.root / 'config/first.toml')
self.assertEqual((first.network, first.width, first.height, first.refresh_millihz), ('user', 1600, 1080, 0))
added = self.run_cli('add', 'second')
self.assertEqual(added.returncode, 0, added.stderr)
second = load_profile(self.root / 'config/second.toml')
self.assertEqual((second.network, second.width, second.height, second.refresh_millihz), ('bridge', 1920, 1080, 144000))
self.defaults.write_text('width=1024\nnetwork="user"\n')
self.assertEqual(load_profile(self.root / 'config/second.toml'), second)
self.defaults.write_text('width="bad"\n')
listed = self.run_cli('list')
self.assertEqual(listed.returncode, 0, listed.stderr)
self.assertEqual(len(json.loads(listed.stdout)), 2)
self.assertEqual(self.run_cli('add', 'third').returncode, 1)
self.assertFalse((self.root / 'config/third.toml').exists())
def test_missing_defaults_use_developer_preferences_and_media_cli(self):
added = self.run_cli('add', 'guest')
self.assertEqual(added.returncode, 0, added.stderr)
profile = load_profile(self.root / 'config/guest.toml')
self.assertEqual((profile.network, profile.width, profile.height, profile.refresh_millihz), ('user', 1280, 800, 0))
iso = self.root / 'installer.iso'
iso.touch()
attached = self.run_cli('media', 'guest', '--iso', str(iso))
self.assertEqual(attached.returncode, 0, attached.stderr)
iso.unlink()
listed = self.run_cli('list')
self.assertEqual(listed.returncode, 0, listed.stderr)
self.assertEqual(self.run_cli('status', 'guest').returncode, 0)
detached = self.run_cli('media', 'guest', '--detach')
self.assertEqual(detached.returncode, 0, detached.stderr)
self.assertIsNone(load_profile(self.root / 'config/guest.toml').iso)
def test_unknown_or_invalid_distribution_defaults_are_rejected(self):
for content in ('iso="/private.iso"\n', 'network="public"\n', 'refresh_millihz=true\n', '[display]\nwidth=1920\n'):
self.defaults.write_text(content)
result = self.run_cli('add', 'guest')
self.assertEqual(result.returncode, 1, result.stderr)
self.assertNotIn('Traceback', result.stderr)
self.assertFalse((self.root / 'config/guest.toml').exists())
+95
View File
@@ -0,0 +1,95 @@
import tempfile
import unittest
from pathlib import Path
from hydra.profiles import Profile, load_profile, save_profile
class ProfilesTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hydra-test-')
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.iso = self.root / 'installer with spaces,comma.iso'
self.iso.touch()
def test_roundtrip_paths_and_defaults(self):
profile = Profile('guest', self.iso)
save_profile(self.root / 'guest.toml', profile)
loaded = load_profile(self.root / 'guest.toml')
self.assertEqual(loaded.iso, self.iso)
self.assertEqual((loaded.memory, loaded.cpus, loaded.disk_size), (4096, 2, '40G'))
self.assertEqual((loaded.graphics, loaded.network, loaded.bridge), ('auto', 'user', 'hydrabr0'))
def test_invalid_names_resources_and_modes(self):
for name in ('', '../guest', 'Guest', '-guest', 'a' * 33, 'a\n'):
with self.subTest(name=name), self.assertRaises(ValueError):
Profile(name, self.iso)
for field, values in {'memory': (0, -1, True, '4096'), 'cpus': (0, -1, True, 257),
'disk_size': ('0G', '-1G', '40G,evil', '40'),
'graphics': ('gl',), 'network': ('public',),
'bridge': ('../bad', 'a,evil')}.items():
for value in values:
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
Profile('guest', self.iso, **{field: value})
def test_invalid_media_path_types(self):
for iso in (True, 42, ''):
with self.subTest(iso=iso), self.assertRaises(ValueError):
Profile('guest', iso)
def test_duplicate_profile_preserves_existing(self):
dest = self.root / 'guest.toml'
save_profile(dest, Profile('guest', self.iso))
original = dest.read_bytes()
with self.assertRaises(FileExistsError):
save_profile(dest, Profile('guest', self.iso, memory=8192))
self.assertEqual(dest.read_bytes(), original)
def test_disk_only_profile_roundtrips_without_media(self):
dest = self.root / 'guest.toml'
save_profile(dest, Profile('guest'))
self.assertIsNone(load_profile(dest).iso)
self.assertNotIn('iso =', dest.read_text())
def test_missing_media_does_not_block_loading_profile(self):
dest = self.root / 'guest.toml'
save_profile(dest, Profile('guest', self.iso))
self.iso.unlink()
self.assertEqual(load_profile(dest).iso, self.iso)
def test_display_bounds_and_types(self):
for field, values in {'width': (0, True, 16385), 'height': (0, '1080', 16385),
'refresh_millihz': (-1, True, 1000001)}.items():
for value in values:
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
Profile('guest', self.iso, **{field: value})
def test_explicit_mac_roundtrip_and_validation(self):
dest = self.root / 'guest.toml'
save_profile(dest, Profile('guest', mac='02:ab:cd:ef:01:23'))
self.assertEqual(load_profile(dest).mac, '02:ab:cd:ef:01:23')
for value in ('ff:00:00:00:00:01', '00:00:00:00:00:01',
'02:00:00:00:00:zz', '02:00:00:00:00', True):
with self.subTest(mac=value), self.assertRaises(ValueError):
Profile('guest', mac=value)
def test_refresh_vertical_descriptor_boundaries(self):
for value in (0, 50000, 255000):
path = self.root / 'guest.toml'
save_profile(path, Profile('guest', refresh_millihz=value))
self.assertEqual(load_profile(path).refresh_millihz, value)
path.unlink()
for value in (1, 30000, 49999, 255001, 1000000):
with self.subTest(refresh=value), self.assertRaises(ValueError):
Profile('guest', refresh_millihz=value)
def test_profile_name_must_match_filename(self):
path = self.root / 'guest.toml'
path.write_text(f'name = "other"\niso = "{self.iso}"\n')
with self.assertRaises(ValueError):
load_profile(path)
if __name__ == '__main__':
unittest.main()
+141
View File
@@ -0,0 +1,141 @@
import json
import socket
import tempfile
import threading
import time
import unittest
from pathlib import Path
from hydra.qmp import QMPClient, QMPError
class QMPServer:
def __init__(self, path, handler):
self.socket = socket.socket(socket.AF_UNIX)
self.socket.bind(str(path))
self.socket.listen(4)
self.socket.settimeout(0.1)
self.handler = handler
self.stop = threading.Event()
self.errors = []
self.thread = threading.Thread(target=self.run, daemon=True)
self.thread.start()
def run(self):
while not self.stop.is_set():
try:
conn, _ = self.socket.accept()
except socket.timeout:
continue
except OSError:
break
try:
with conn:
conn.settimeout(1)
self.handler(conn)
except (BrokenPipeError, ConnectionResetError, socket.timeout):
pass
except Exception as exc:
self.errors.append(exc)
def close(self):
self.stop.set()
self.socket.close()
self.thread.join(2)
def send(conn, message):
conn.sendall(json.dumps(message).encode() + b'\r\n')
def greeting(conn):
send(conn, {'QMP': {'version': {'qemu': {'major': 10, 'minor': 0, 'micro': 0},
'package': ''}, 'capabilities': []}})
class QMPTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hq-')
self.addCleanup(self.tmp.cleanup)
self.path = Path(self.tmp.name) / 'qmp'
def server(self, handler):
srv = QMPServer(self.path, handler)
self.addCleanup(srv.close)
return srv
def test_handshake_ids_events_and_return(self):
commands = []
def handler(conn):
greeting(conn)
stream = conn.makefile('rb')
for index in range(2):
request = json.loads(stream.readline())
commands.append(request)
send(conn, {'event': 'STOP', 'data': {}})
send(conn, {'return': {'ignored': True}, 'id': 'unrelated'})
send(conn, {'return': {} if index == 0 else {'status': 'running'}, 'id': request['id']})
self.server(handler)
with QMPClient(self.path, timeout=0.5) as client:
self.assertEqual(client.command('query-status'), {'status': 'running'})
self.assertEqual([cmd['execute'] for cmd in commands], ['qmp_capabilities', 'query-status'])
self.assertNotEqual(commands[0]['id'], commands[1]['id'])
def test_command_error(self):
def handler(conn):
greeting(conn)
stream = conn.makefile('rb')
for index in range(2):
request = json.loads(stream.readline())
if index == 0:
send(conn, {'return': {}, 'id': request['id']})
else:
send(conn, {'error': {'class': 'CommandNotFound', 'desc': 'unsupported'}, 'id': request['id']})
self.server(handler)
with QMPClient(self.path, timeout=0.5) as client, self.assertRaises(QMPError):
client.command('missing')
def test_greeting_timeout_is_bounded(self):
self.server(lambda conn: time.sleep(0.3))
started = time.monotonic()
with self.assertRaises(QMPError):
QMPClient(self.path, timeout=0.08)
self.assertLess(time.monotonic() - started, 0.25)
def test_event_stream_cannot_extend_command_deadline(self):
def handler(conn):
greeting(conn)
request = json.loads(conn.makefile('rb').readline())
while True:
send(conn, {'event': 'STOP', 'data': {}})
time.sleep(0.01)
self.server(handler)
started = time.monotonic()
with self.assertRaises(QMPError):
QMPClient(self.path, timeout=0.08)
self.assertLess(time.monotonic() - started, 0.25)
def test_malformed_greeting_rejected(self):
self.server(lambda conn: send(conn, {'return': {}}))
with self.assertRaises(QMPError):
QMPClient(self.path)
def test_command_timeout_after_successful_handshake(self):
def handler(conn):
greeting(conn)
with conn.makefile('rb') as stream:
request = json.loads(stream.readline())
send(conn, {'return': {}, 'id': request['id']})
stream.readline()
time.sleep(0.3)
self.server(handler)
with QMPClient(self.path, timeout=0.08) as client:
started = time.monotonic()
with self.assertRaises(QMPError):
client.command('query-status')
self.assertLess(time.monotonic() - started, 0.25)
if __name__ == '__main__':
unittest.main()
+394
View File
@@ -0,0 +1,394 @@
import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch
from hydra.profiles import Profile, save_profile
from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv
from test_qmp import QMPServer, greeting, send
# External protocol fixtures isolate lifecycle tests from user VMs and real disks.
FAKE_QEMU = r'''#!/usr/bin/env python3
import json, os, socket, sys
args = sys.argv[1:]
def val(flag): return args[args.index(flag)+1]
if os.environ.get('HYDRA_TEST_FAIL'):
print('intentional boot failure', file=sys.stderr)
sys.exit(1)
path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',')
pid = os.fork()
if pid:
sys.exit(0)
os.setsid()
with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid()))
s = socket.socket(socket.AF_UNIX)
s.bind(path)
s.listen(8)
alive = True
while alive:
c, _ = s.accept()
try:
with c:
c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n')
with c.makefile('rb') as stream:
for line in stream:
cmd = json.loads(line)
name = cmd['execute']
result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')},
'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {})
c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n')
if name == 'quit':
alive = False
break
except (BrokenPipeError, ConnectionResetError):
pass
s.close()
os.unlink(path)
'''
FAKE_IMG = r'''#!/usr/bin/env python3
import sys
with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture')
'''
class RuntimeTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hr-')
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma',
self.root / 'run,comma', boot_timeout=1)
self.iso = self.root / 'installer space,comma.iso'
self.iso.touch()
self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M')
self.runtime.add(self.profile)
self.qemu = self.root / 'fake qemu'
self.qemu.write_text(FAKE_QEMU)
self.qemu.chmod(0o700)
self.img = self.root / 'fake img'
self.img.write_text(FAKE_IMG)
self.img.chmod(0o700)
env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)})
env.start()
self.addCleanup(env.stop)
self.addCleanup(self.cleanup_guest)
def cleanup_guest(self):
try:
if self.runtime.status('guest')['status'] == 'running':
self.runtime.stop('guest', force=True)
except VMError:
pass
def test_mac_persisted_distinct_and_stable_across_restarts(self):
self.runtime.add(Profile('other', network='bridge'))
other = self.runtime.profile('other')
guest = self.runtime.profile('guest')
self.assertIsNotNone(guest.mac)
self.assertNotEqual(guest.mac, other.mac)
octet = int(guest.mac.split(':')[0], 16)
self.assertEqual(octet & 3, 2) # locally administered, unicast
def nic(profile, runtime):
return next(arg for arg in build_qemu_argv(profile, runtime.paths(profile.name))
if arg.startswith('virtio-net-pci,'))
expected = nic(guest, self.runtime)
self.runtime.start('guest')
self.runtime.stop('guest', force=True)
self.runtime.start('guest')
self.runtime.stop('guest', force=True)
self.assertEqual(nic(self.runtime.profile('guest'), self.runtime), expected)
changed_runtime = Runtime(self.runtime.config, self.runtime.state, self.root / 'different-run')
self.assertEqual(nic(changed_runtime.profile('guest'), changed_runtime), expected)
self.runtime.media('guest', None)
self.assertEqual(self.runtime.profile('guest').mac, guest.mac)
def test_legacy_profiles_gain_stable_owner_specific_argv_mac(self):
file = self.runtime.config / 'legacy.toml'
save_profile(file, Profile('legacy', network='bridge'))
original = file.read_bytes()
legacy = self.runtime.profile('legacy')
self.assertIsNone(legacy.mac)
paths = self.runtime.paths('legacy')
def mac():
return next(arg for arg in build_qemu_argv(legacy, paths) if arg.startswith('virtio-net-pci,'))
with patch('os.getuid', return_value=1000):
first = mac()
self.assertEqual(mac(), first)
self.assertNotEqual(first, next(arg for arg in build_qemu_argv(
legacy, self.runtime.paths('other')) if arg.startswith('virtio-net-pci,')))
with patch('os.getuid', return_value=1001):
self.assertNotEqual(mac(), first)
self.assertEqual(file.read_bytes(), original)
def test_disk_boot_omits_media_and_requests_display(self):
profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
graphics='software')
argv = build_qemu_argv(profile, self.runtime.paths('guest'))
self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
self.assertNotIn('ide-cd,drive=iso', argv)
self.assertIn('order=c', argv)
devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
self.assertEqual(len(devices), 2)
stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
self.assertFalse(any('refresh_rate=' in arg for arg in stock))
def test_missing_iso_is_error_at_start_without_creating_disk(self):
self.iso.unlink()
with self.assertRaisesRegex(VMError, 'ISO'):
self.runtime.start('guest')
self.assertFalse(self.runtime.paths('guest').disk.exists())
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
def test_media_changes_only_while_stopped_and_preserves_disk(self):
paths = self.runtime.paths('guest')
self.runtime.start('guest')
disk = paths.disk.read_bytes()
config = (self.runtime.config / 'guest.toml').read_bytes()
for iso in (None, self.iso):
with self.assertRaises(VMError):
self.runtime.media('guest', iso)
self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
self.iso.unlink()
self.assertEqual(self.runtime.profile('guest').iso, self.iso)
self.runtime.stop('guest', force=True)
self.runtime.media('guest', None)
self.assertIsNone(self.runtime.profile('guest').iso)
self.runtime.start('guest')
self.runtime.stop('guest', force=True)
self.assertEqual(paths.disk.read_bytes(), disk)
replacement = self.root / 'replacement.iso'
replacement.touch()
self.runtime.media('guest', replacement)
self.assertEqual(self.runtime.profile('guest').iso, replacement)
with self.assertRaisesRegex(VMError, 'ISO'):
self.runtime.media('guest', self.iso)
self.assertEqual(self.runtime.profile('guest').iso, replacement)
def test_argv_is_private_and_comma_safe(self):
paths = self.runtime.paths('guest')
argv = build_qemu_argv(self.profile, paths)
self.assertEqual(argv[0], str(self.qemu))
blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs])
self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs])
spice = argv[argv.index('-spice')+1]
self.assertIn('unix=on', spice)
self.assertIn('disable-copy-paste=on', spice)
self.assertIn('disable-agent-file-xfer=on', spice)
self.assertIn('run,,comma', spice)
self.assertIn('virtio-blk-pci,drive=disk', argv)
self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
self.assertNotIn('-vnc', argv)
self.assertNotIn('-virtfs', argv)
def test_socket_removed_during_stat_is_safely_not_listening(self):
paths = self.runtime.paths('guest')
paths.qmp.parent.mkdir(parents=True)
with socket.socket(socket.AF_UNIX) as fixture:
fixture.bind(str(paths.qmp))
original_lstat = Path.lstat
def disappearing_socket(path):
if path == paths.qmp:
path.unlink(missing_ok=True)
return original_lstat(path)
with patch.object(Path, 'lstat', disappearing_socket):
self.assertFalse(self.runtime._listening(paths.qmp))
def test_argv_provides_audio_and_manual_usb_channels(self):
argv = build_qemu_argv(self.profile, self.runtime.paths('guest'))
self.assertIn('spice,id=audio', argv)
self.assertIn('hda-duplex,audiodev=audio', argv)
for channel in range(3):
self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv)
self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv)
self.assertFalse(any('usb-host' in argument for argument in argv))
def test_metadata_symlink_does_not_overwrite_user_data(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
valuable = self.root / 'valuable data'
valuable.write_text('preserve me')
paths.metadata.symlink_to(valuable)
self.runtime.start('guest')
self.assertEqual(valuable.read_text(), 'preserve me')
self.assertFalse(paths.metadata.is_symlink())
def test_disk_symlink_never_changes_its_target(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
valuable = self.root / 'valuable disk'
valuable.write_text('preserve me')
paths.disk.symlink_to(valuable)
with self.assertRaises(VMError):
self.runtime.start('guest')
self.assertEqual(valuable.read_text(), 'preserve me')
def test_malformed_identity_record_is_safe_unknown(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.metadata.write_text('[]')
server = QMPServer(paths.qmp, greeting)
self.addCleanup(server.close)
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
commands = []
def handler(conn):
greeting(conn)
for line in conn.makefile('rb'):
req = json.loads(line)
commands.append(req['execute'])
result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
send(conn, {'return': result, 'id': req['id']})
server = QMPServer(paths.qmp, handler)
self.addCleanup(server.close)
with self.assertRaises(VMError):
self.runtime.stop('guest', force=True)
self.assertNotIn('quit', commands)
def test_reused_pid_without_qmp_never_controls_a_process(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.pid.write_text(str(os.getpid()))
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
with patch('os.kill', side_effect=AssertionError('must not signal any PID')):
with self.assertRaises(VMError):
self.runtime.stop('guest', force=True)
def test_malformed_query_result_is_safe_unknown(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
def handler(conn):
greeting(conn)
for line in conn.makefile('rb'):
req = json.loads(line)
send(conn, {'return': [], 'id': req['id']})
server = QMPServer(paths.qmp, handler)
self.addCleanup(server.close)
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
def test_cli_qmp_failure_reports_error_without_traceback(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
def handler(conn):
greeting(conn)
for line in conn.makefile('rb'):
req = json.loads(line)
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
if req['execute'] == 'system_powerdown':
send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']})
else:
send(conn, {'return': result, 'id': req['id']})
server = QMPServer(paths.qmp, handler)
self.addCleanup(server.close)
run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config),
'--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime),
'shutdown', 'guest'], capture_output=True, text=True, timeout=4)
self.assertEqual(run.returncode, 1)
self.assertIn('cannot power down', run.stderr)
self.assertNotIn('Traceback', run.stderr)
def test_disk_and_profile_survive_shutdown_and_force(self):
paths = self.runtime.paths('guest')
paths.disk.parent.mkdir(parents=True, exist_ok=True)
paths.disk.write_bytes(b'existing valuable disk')
self.runtime.start('guest')
self.assertEqual(self.runtime.status('guest')['status'], 'running')
self.runtime.shutdown('guest')
self.assertEqual(self.runtime.status('guest')['status'], 'running')
with self.assertRaises(VMError):
self.runtime.stop('guest', force=False)
self.runtime.stop('guest', force=True)
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk')
def test_simultaneous_start_only_launches_one_vm(self):
barrier = threading.Barrier(2)
results = []
def start():
barrier.wait()
try:
results.append(self.runtime.start('guest')['status'])
except VMError as exc:
results.append(str(exc))
threads = [threading.Thread(target=start) for _ in range(2)]
for thread in threads:
thread.start()
for thread in threads:
thread.join(4)
self.assertFalse(thread.is_alive())
self.assertEqual(results.count('running'), 1)
self.assertEqual(len(results), 2)
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
def test_stale_socket_is_replaced_only_when_no_listener(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
stale = socket.socket(socket.AF_UNIX)
stale.bind(str(paths.qmp))
stale.close()
self.runtime.start('guest')
self.assertEqual(self.runtime.status('guest')['status'], 'running')
def test_failed_boot_is_reported_and_disk_preserved(self):
with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError):
self.runtime.start('guest')
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text())
def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self):
paths = self.runtime.paths('guest')
self.runtime.prepare(paths)
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
paths.pid.write_text(str(os.getpid()))
commands = []
def handler(conn):
greeting(conn)
for line in conn.makefile('rb'):
req = json.loads(line)
commands.append(req['execute'])
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {})
send(conn, {'return': result, 'id': req['id']})
server = QMPServer(paths.qmp, handler)
self.addCleanup(server.close)
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True),
lambda: self.runtime.start('guest')):
with self.assertRaises(VMError):
control()
self.assertNotIn('quit', commands)
self.assertNotIn('system_powerdown', commands)
self.assertFalse(paths.disk.exists())
def test_private_runtime_and_socket_length(self):
self.runtime.prepare(self.runtime.paths('guest'))
self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700)
with self.assertRaises(VMError):
Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest')
def test_cli_lifecycle_and_global_paths(self):
base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state),
'--runtime', str(self.runtime.runtime)]
for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'),
(['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]:
run = subprocess.run(base + args, text=True, capture_output=True, timeout=4)
self.assertEqual(run.returncode, 0, run.stderr)
self.assertIn(expected, run.stdout)
if __name__ == '__main__':
unittest.main()
+301
View File
@@ -0,0 +1,301 @@
import contextlib
import fcntl
import io
import json
import os
import shlex
import socket
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from xml.etree import ElementTree as ET
from hydra.profiles import Profile
from hydra.runtime import Runtime
from hydra import session
from hydra.cli import main, parser
class SessionTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hs-')
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.iso = self.root / 'scratch.iso'
self.iso.touch()
self.manager = Runtime(self.root / 'config', self.root / 'state', self.root / 'run')
self.manager.add(Profile('alpha', self.iso))
self.manager.add(Profile('beta', self.iso))
def test_config_has_exact_titles_workspaces_and_explicit_fullscreen(self):
tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot', 'a & b', "x'y"]))
self.assertEqual([n.text for n in tree.findall('./desktops/names/name')], ['alpha', 'beta'])
self.assertEqual(tree.findtext('./desktops/popupTime'), '0')
rules = tree.findall('./windowRules/windowRule')
self.assertEqual([r.get('title') for r in rules], ['HydraOS: alpha', 'HydraOS: beta'])
for index, rule in enumerate(rules, 1):
self.assertEqual(rule.get('serverDecoration'), 'no')
self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('to'), str(index))
self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('follow'), 'no')
# Fullscreen is requested by the viewer, never toggled by a rule.
self.assertIsNone(rule.find("./action[@name='ToggleFullscreen']"))
recovery = tree.find("./keyboard/keybind[@key='C-A-S-Escape']/action")
self.assertEqual(shlex.split(recovery.get('command')), ['foot', 'a & b', "x'y"])
def test_only_reserved_host_keys_and_inert_mouse_binding(self):
tree = ET.fromstring(session.build_labwc_config(['alpha'], ['foot']))
self.assertIsNone(tree.find('./keyboard/default'))
self.assertIsNone(tree.find('./mouse/default'))
bindings = tree.findall('./keyboard/keybind')
self.assertEqual({b.get('key') for b in bindings}, {'C-A-S-Left', 'C-A-S-Right', 'C-A-S-Escape', 'XF86MonBrightnessUp', 'XF86MonBrightnessDown'})
self.assertTrue(all(b.get('overrideInhibition') == 'yes' for b in bindings))
self.assertEqual(len(tree.findall('./mouse/context/mousebind')), 1)
self.assertEqual(tree.findall('./mouse/context/mousebind/action'), [])
self.assertEqual(tree.findtext('./libinput/device/tap'), 'yes')
def test_empty_session_has_one_workspace_without_empty_names(self):
tree = ET.fromstring(session.build_labwc_config([], ['foot']))
self.assertEqual(tree.find('./desktops').get('number'), '1')
self.assertIsNone(tree.find('./desktops/names'))
def test_viewer_argv_is_private_safe_and_manual_usb_only(self):
paths = self.manager.paths('alpha')
argv = session.build_viewer_argv(paths, '/tool path/viewer')
self.assertEqual(argv[0], '/tool path/viewer')
self.assertIn('--full-screen', argv)
self.assertIn('--hide-toolbar', argv)
self.assertFalse(any(arg.startswith('--spice-shared-dir') for arg in argv))
self.assertFalse(any(arg.startswith('--spice-share-cd') for arg in argv))
self.assertIn('--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0', argv)
self.assertIn('--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0', argv)
self.assertIn('usb-device-select=ctrl+alt+shift+u', argv[argv.index('--hotkeys') + 1])
self.assertEqual(argv[-1], 'spice+unix://' + str(paths.spice))
windowed = session.build_viewer_argv(paths, 'remote-viewer', windowed=True)
self.assertNotIn('--full-screen', windowed)
self.assertNotIn('--hide-toolbar', windowed)
with self.assertRaises(ValueError):
session.viewer_title('bad*')
def test_missing_display_and_viewer_are_recoverable_cli_errors(self):
base = ['--config', str(self.manager.config), '--state', str(self.manager.state), '--runtime', str(self.manager.runtime)]
with patch.dict(os.environ, {}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error:
self.assertEqual(main(base + ['view', 'alpha']), 1)
self.assertIn('display', error.getvalue().lower())
with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': '/missing/remote-viewer'}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error:
self.assertEqual(main(base + ['view', 'alpha']), 1)
self.assertIn('viewer', error.getvalue().lower())
def test_windowed_viewer_exit_with_missing_iso_leaves_vm_control_untouched(self):
self.iso.unlink()
viewer = self.root / 'fake-viewer'
viewer.write_text('#!/bin/sh\ncase "$1" in --help) echo --hide-toolbar; exit 0;; esac\nexit 7\n')
viewer.chmod(0o700)
self.manager.paths('alpha').spice.parent.mkdir(parents=True, exist_ok=True)
sock = socket.socket(socket.AF_UNIX)
self.addCleanup(sock.close)
sock.bind(str(self.manager.paths('alpha').spice))
with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer)}), patch.object(self.manager, 'stop', side_effect=AssertionError('must not stop')):
self.assertEqual(session.view(self.manager, 'alpha', windowed=True), 7)
self.assertTrue(self.manager.paths('alpha').spice.exists())
def test_session_parser_and_default_physical_backend(self):
self.assertTrue(parser().parse_args(['session', '--nested']).nested)
physical = session.compositor_environment({'WAYLAND_DISPLAY': 'active', 'DISPLAY': ':0', 'WLR_BACKENDS': 'wayland'}, nested=False)
self.assertNotIn('WAYLAND_DISPLAY', physical)
self.assertNotIn('DISPLAY', physical)
self.assertEqual(physical['WLR_BACKENDS'], 'drm,libinput')
nested = session.compositor_environment({'WAYLAND_DISPLAY': 'active'}, nested=True)
self.assertEqual(nested['WLR_BACKENDS'], 'wayland')
with self.assertRaises(session.SessionError):
session.compositor_environment({}, nested=True)
def test_uri_escapes_percent_query_and_fragment(self):
from dataclasses import replace
paths = replace(self.manager.paths('alpha'), spice=self.root / 'a %?#.sock')
self.assertTrue(session.build_viewer_argv(paths, 'viewer')[-1].endswith('/a%20%25%3F%23.sock'))
def test_workspace_names_cannot_be_interpreted_as_relative_actions(self):
tree = ET.fromstring(session.build_labwc_config(['left', 'right', 'current'], ['foot']))
self.assertEqual([a.get('to') for a in tree.findall("./windowRules/windowRule/action[@name='SendToDesktop']")], ['1', '2', '3'])
def test_launch_writes_only_selected_private_runtime_and_serializes_paths(self):
capture = self.root / 'capture.json'
labwc = self.root / 'fake-labwc'
labwc.write_text('#!' + os.sys.executable + '\n' + '''import json, os, sys
from pathlib import Path
if '--version' in sys.argv:
print('labwc 0.20.2')
else:
config = Path(sys.argv[sys.argv.index('-C') + 1])
manifest = json.loads((config.parent / 'session.json').read_text())
Path(os.environ['CAPTURE']).write_text(json.dumps({'manifest': manifest, 'config': str(config), 'xml': (config / 'rc.xml').read_text(), 'env': dict(os.environ), 'autostart': (config / 'autostart').read_text()}))
''')
labwc.chmod(0o700)
old_home = self.root / 'untouched-home'
old_home.mkdir()
marker = old_home / 'keep'
marker.write_text('valuable')
with patch.dict(os.environ, {'HYDRA_LABWC': str(labwc), 'HYDRA_TERMINAL': str(labwc), 'HYDRA_BRIGHTNESS': str(labwc), 'CAPTURE': str(capture), 'HOME': str(old_home), 'DISPLAY': ':fake', 'WAYLAND_DISPLAY': 'active'}):
self.assertEqual(session.launch_session(self.manager), 0)
data = json.loads(capture.read_text())
self.assertTrue(Path(data['config']).is_relative_to(self.manager.runtime))
self.assertFalse(Path(data['config']).exists())
self.assertEqual(data['manifest']['config'], str(self.manager.config))
self.assertNotIn('DISPLAY', data['env'])
self.assertNotIn('WAYLAND_DISPLAY', data['env'])
self.assertEqual(list(old_home.iterdir()), [marker])
self.assertEqual(marker.read_text(), 'valuable')
def test_missing_socket_and_stock_viewer_do_not_create_disk(self):
viewer = self.root / 'viewer'
viewer.write_text('#!/bin/sh\necho --hide-toolbar\n')
viewer.chmod(0o700)
with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'fake', 'HYDRA_VIEWER': str(viewer)}):
with self.assertRaisesRegex(session.SessionError, 'no SPICE socket'):
session.view(self.manager, 'alpha')
viewer.write_text('#!/bin/sh\necho stock-viewer\n')
with self.assertRaisesRegex(session.SessionError, 'adaptation'):
session.view(self.manager, 'alpha')
self.assertFalse(self.manager.paths('alpha').disk.exists())
def test_viewer_settings_are_isolated_and_disable_clipboard(self):
env = session._viewer_environment(self.root / 'viewer-state')
self.assertEqual(Path(env['XDG_CONFIG_HOME']).joinpath('virt-viewer/settings').read_text(), '[virt-viewer]\nshare-clipboard=false\n')
self.assertEqual(env['GSETTINGS_BACKEND'], 'memory')
def test_initial_workspace_can_select_a_running_vm_after_stopped_profiles(self):
tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot'], initial='beta'))
self.assertEqual(tree.findtext('./desktops/initial'), '2')
with self.assertRaises(session.SessionError):
session.build_labwc_config(['alpha'], ['foot'], initial='missing')
def test_bad_helper_manifest_is_a_cli_error(self):
manifest = self.root / 'bad-session.json'
manifest.write_text('[]')
with contextlib.redirect_stderr(io.StringIO()) as error:
self.assertEqual(main(['_session-helper', 'recover', str(manifest)]), 1)
self.assertIn('manifest', error.getvalue().lower())
def test_empty_launch_refresh_rebuilds_profile_snapshot(self):
manager = Runtime(self.root / 'empty-config', self.root / 'empty-state', self.root / 'empty-run')
capture = self.root / 'refresh-capture.json'
compositor = self.root / 'refresh-labwc'
compositor.write_text('#!' + os.sys.executable + '\n' + r'''import json, os, subprocess, sys
from pathlib import Path
if '--version' in sys.argv:
print('labwc 0.20.2')
sys.exit(0)
config = Path(sys.argv[sys.argv.index('-C') + 1])
manifest = config.parent / 'session.json'
data = json.loads(manifest.read_text())
capture = Path(os.environ['CAPTURE'])
seen = json.loads(capture.read_text()) if capture.exists() else []
seen.append({'names': data['names'], 'xml': (config / 'rc.xml').read_text()})
capture.write_text(json.dumps(seen))
if len(seen) == 1:
launcher = os.environ['HYDRA_SOURCE_LAUNCHER']
for name in ('first', 'second'):
subprocess.run([sys.executable, launcher, '--config', data['config'], '--state', data['state'],
'--runtime', data['runtime'], 'add', name], check=True)
subprocess.run([sys.executable, launcher, '_session-helper', 'recover', str(manifest)],
input=':refresh\n', text=True, check=True)
sys.exit(67)
sys.exit(0)
''')
compositor.chmod(0o700)
with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture),
'HYDRA_SOURCE_LAUNCHER': str(Path('bin/hydra').resolve())}), contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(session.launch_session(manager), 0)
seen = json.loads(capture.read_text())
self.assertEqual([entry['names'] for entry in seen], [[], ['first', 'second']])
tree = ET.fromstring(seen[1]['xml'])
self.assertEqual(tree.find('desktops').get('number'), '2')
self.assertEqual([rule.get('title') for rule in tree.findall('./windowRules/windowRule')],
['HydraOS: first', 'HydraOS: second'])
self.assertFalse(manager.paths('first').disk.exists())
def test_no_running_vms_opens_recovery_without_start_or_stop(self):
terminal = self.root / 'terminal'
capture = self.root / 'recovery-argv.json'
terminal.write_text('#!' + os.sys.executable + '\nimport json, os, sys\nfrom pathlib import Path\nPath(os.environ["CAPTURE"]).write_text(json.dumps(sys.argv[1:]))\n')
terminal.chmod(0o700)
manifest = self.root / 'session.json'
manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha', 'beta'], 'terminal': str(terminal), 'directory': str(self.root)}))
with patch.dict(os.environ, {'CAPTURE': str(capture)}), patch.object(Runtime, 'start', side_effect=AssertionError('must not start')), patch.object(Runtime, 'stop', side_effect=AssertionError('must not stop')):
self.assertEqual(session.session_helper('clients', manifest), 0)
self.assertEqual(json.loads(capture.read_text())[0], '--title=HydraOS-recovery')
self.assertEqual(self.manager.status('alpha')['status'], 'stopped')
self.assertFalse(self.manager.paths('alpha').disk.exists())
def test_reconnect_does_not_duplicate_a_session_viewer(self):
manifest = self.root / 'session.json'
manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha'], 'terminal': '/bin/sh'}))
with (self.root / 'alpha.viewer.lock').open('a') as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
with patch.object(session, 'view', side_effect=AssertionError('must not duplicate')), contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(session.session_helper('reconnect:alpha', manifest), 0)
def _recovery_fixture(self):
for name in ('s', 'q'):
self.manager.add(Profile(name, self.iso))
paths = self.manager.paths(name)
paths.spice.parent.mkdir(parents=True)
fixture = socket.socket(socket.AF_UNIX)
self.addCleanup(fixture.close)
fixture.bind(str(paths.spice))
viewer = self.root / 'viewer'
viewer.write_text('#!' + os.sys.executable + '\n' + '''import os, sys
from pathlib import Path
from urllib.parse import unquote, urlsplit
if '--help' in sys.argv:
print('--hide-toolbar')
else:
name = Path(unquote(urlsplit(sys.argv[-1]).path)).parent.name
Path(os.environ['RECOVERY_CAPTURE'], name + '.connected').write_text(name)
''')
viewer.chmod(0o700)
shell = self.root / 'shell'
shell.write_text('#!' + os.sys.executable + '\nimport os\nfrom pathlib import Path\nPath(os.environ["RECOVERY_CAPTURE"], "shell.opened").touch()\n')
shell.chmod(0o700)
manifest = self.root / 'session.json'
manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['s', 'q'], 'terminal': '/bin/sh'}))
return manifest, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer), 'RECOVERY_CAPTURE': str(self.root), 'SHELL': str(shell)}
def _check_recovery_vm_name(self, name):
manifest, env = self._recovery_fixture()
processes = []
original_popen = subprocess.Popen
def launch(*args, **kwargs):
process = original_popen(*args, **kwargs)
processes.append(process)
return process
with patch.dict(os.environ, env), patch('builtins.input', side_effect=[name, ':quit', EOFError()]), patch.object(subprocess, 'Popen', launch), contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(session.session_helper('recover', manifest), 0)
for process in processes:
self.assertEqual(process.wait(timeout=5), 0)
connected = self.root / (name + '.connected')
self.assertTrue(connected.exists(), f'VM {name} was not reconnected')
self.assertEqual(connected.read_text(), name)
self.assertFalse((self.root / 'shell.opened').exists())
self.assertFalse(self.manager.paths(name).disk.exists())
def test_recovery_reconnects_vm_named_s(self):
self._check_recovery_vm_name('s')
def test_recovery_reconnects_vm_named_q(self):
self._check_recovery_vm_name('q')
def test_recovery_colon_commands_open_shell_and_close_terminal(self):
manifest, env = self._recovery_fixture()
with patch.dict(os.environ, env), patch('builtins.input', side_effect=[':shell', ':quit', EOFError()]), contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(session.session_helper('recover', manifest), 0)
self.assertTrue((self.root / 'shell.opened').exists())
self.assertFalse((self.root / 's.connected').exists())
self.assertFalse((self.root / 'q.connected').exists())
if __name__ == '__main__':
unittest.main()
+48
View File
@@ -0,0 +1,48 @@
"""Evidence gates used by the real private-display integration probe."""
import tempfile
import unittest
from pathlib import Path
from dev import smoke
class SmokeEvidenceTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='hydra-smoke-test-')
self.addCleanup(self.tmp.cleanup)
root = Path(self.tmp.name)
self.hidden = root / 'hidden.log'
self.visible = root / 'visible.log'
self.hidden.write_text('wl_keyboard#21.enter(1)\nwl_keyboard#21.leave(2)\n')
self.visible.write_text('wl_keyboard#21.enter(3)\n')
def test_focus_parser_ignores_pointer_events(self):
self.hidden.write_text(self.hidden.read_text() + 'wl_pointer#22.enter(4)\n')
self.assertEqual(smoke.focus_events(self.hidden), ['enter', 'leave'])
def test_old_focus_events_cannot_prove_a_new_switch(self):
self.assertFalse(smoke.fresh_focus_transition(['enter', 'leave'], ['enter'],
['enter', 'leave'], ['enter']))
self.assertTrue(smoke.fresh_focus_transition(['enter'], ['leave'],
['enter', 'leave'], ['leave', 'enter']))
def test_hidden_guest_must_advance_during_hidden_interval(self):
with self.assertRaisesRegex(AssertionError, 'hidden guest stopped'):
smoke.measure_hidden_interval(lambda: 10, self.hidden, self.visible, duration=.01)
readings = iter((10, 20))
result = smoke.measure_hidden_interval(lambda: next(readings), self.hidden, self.visible, duration=.01)
self.assertEqual((result['before'], result['after']), (10, 20))
self.assertGreaterEqual(result['elapsed_seconds'], .01)
def test_focus_change_during_sampling_invalidates_hidden_evidence(self):
def reading():
self.hidden.write_text(self.hidden.read_text() + 'wl_keyboard#21.enter(4)\n')
return 10
with self.assertRaisesRegex(AssertionError, 'focus changed'):
smoke.measure_hidden_interval(reading, self.hidden, self.visible, duration=.01)
def test_compositor_classifier_does_not_hide_other_errors(self):
occupied = '[ERROR] [xwayland/sockets.c:47] Failed to bind socket @/tmp/.X11-unix/X2: Address already in use'
self.assertEqual(smoke.classify_compositor_errors(occupied), [occupied])
with self.assertRaisesRegex(AssertionError, 'Unexpected compositor'):
smoke.classify_compositor_errors(occupied + '\n[ERROR] [config.c:1] invalid configuration')