Emil_Shanaty/Hydra-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: Emil_Shanaty/Hydra-OS:feat/image
Branches Tags
Emil_Shanaty/Hydra-OS:main Emil_Shanaty/Hydra-OS:feat/image Emil_Shanaty/Hydra-OS:feat/validation Emil_Shanaty/Hydra-OS:feat/prototype
Emil_Shanaty/Hydra-OS:0.2.0
..
pull from: Emil_Shanaty/Hydra-OS:feat/prototype
Branches Tags
Emil_Shanaty/Hydra-OS:main Emil_Shanaty/Hydra-OS:feat/image Emil_Shanaty/Hydra-OS:feat/validation Emil_Shanaty/Hydra-OS:feat/prototype
Emil_Shanaty/Hydra-OS:0.2.0
11 Commits
feat/image .. feat/prototype
Author SHA1 Message Date
Codex ea43a67a97 fix: preserve guest network identities and refresh session profiles 2026-10-13 05:12:04 +03:00
Codex e9906a793b fix: enclose rounded preferred timings in EDID ranges 2026-10-13 04:44:26 +03:00
Codex adfce75973 fix: verify work during continuously hidden guest intervals 2026-10-13 04:41:07 +03:00
Codex 7353c4ec92 test: verify independent guests and recovery on private displays 2026-10-13 04:35:14 +03:00
Codex 623ecdef85 fix: keep manual USB selection available during keyboard grabs 2026-10-13 04:29:04 +03:00
Codex 6f7344fca5 feat: support installed guests and distribution display defaults 2026-10-13 04:26:49 +03:00
Codex 0e86fffd54 fix: avoid VM name collisions in recovery commands 2026-10-13 04:22:29 +03:00
Codex e6cf7b7252 docs: specify installed guest and graphics integration checks 2026-10-13 04:21:12 +03:00
Codex 09ed8f314f feat: add fullscreen HydraOS session 2026-10-13 04:18:34 +03:00
Codex a6367e55ae fix: tolerate QMP socket removal during status probes 2026-10-13 04:11:42 +03:00
Codex d7d01a4520 feat: add independent VM manager 2026-10-13 04:04:30 +03:00
61 changed files with 45 additions and 12177 deletions
Expand all files Collapse all files
README.md
+3 -61
View File
@@ -6,12 +6,9 @@ desktops may use the same independent VM manager and viewers.
## Status
The first installable UEFI preview passed a complete disposable QEMU/KVM run:
installation, boot without the ISO, password login, persistent data, networking,
nested KVM and VM survival across user logout. See the
[build and acceptance record](docs/releases/2026-10-13/acceptance.json).
The first physical target is ASUS TUF Dash F15 FX517ZC; laptop hardware checks
remain pending. Existing NixOS and VM disks were not installation or test targets.
Under active development, not yet an installation-ready distribution. The target
is an installable UEFI ISO, first validated on ASUS TUF Dash F15 FX517ZC. Existing
NixOS and VM disks are not installation or test targets.
1. Independent QEMU/KVM lifecycle with private QMP control.
2. Minimal labwc session and fullscreen SPICE viewers.
@@ -28,61 +25,6 @@ a refresh-rate setting does not prove delivered FPS.
Build outputs, VM disks, runtime state and credentials stay outside Git.
## First use
Follow the [installation guide](docs/install.md) to install from the live ISO.
The initial empty graphical session opens a recovery terminal. Enter `:shell`
for an ordinary host shell; `sudo nmtui` configures host network connections.
Guest operating systems are supplied by you:
```sh
hydra add work --iso /absolute/path/ubuntu.iso --memory 8192 --cpus 4
hydra start work
hydra add experiments --iso /absolute/path/gentoo.iso
hydra start experiments
exit
```
Back at `hydra recovery>`, enter `:refresh` to show the newly created guests.
Guests remain running while you switch, reconnect viewers or leave the host
graphical session. Shut a guest down from inside its OS, or request an orderly
shutdown with `hydra shutdown NAME`. Check `hydra status NAME` before detaching
its installer with `hydra media NAME --detach` and starting it again.
| Shortcut | Action |
| --- | --- |
| Ctrl+Alt+Shift+Left / Right | Switch guest workspaces |
| Ctrl+Alt+Shift+Escape | Open local recovery |
| Ctrl+Alt+Shift+R | Release viewer input capture |
| Ctrl+Alt+Shift+U | Choose a USB device to redirect |
| Brightness keys | Adjust host backlight |
New guests share `hydrabr0` (`10.203.0.0/24`) and have distinct persistent MAC
addresses. Services can communicate between guests without per-port forwarding;
access from your LAN requires a router route through the HydraOS host. Each
guest controls its own services and firewall. Host Internet access is separate
from this local guest network.
Profiles live in `~/.config/hydra`, disks in `~/.local/state/hydra`; system defaults
for new profiles are in `/etc/hydra/defaults.toml`. Use `hydra --help` for overrides.
Other host desktops can use `hydra start NAME` and `hydra view NAME --windowed`
without entering the minimal labwc session. Ordinary Gentoo administration and
Portage remain available through the recovery shell.
## Build the ISO
Follow the [isolated Gentoo build instructions](image/README.md). Outputs are
`build/artifacts/hydra-os.iso` and `hydra-os.iso.sha256`; binary images stay out
of Git. [Recorded source inputs and package versions](docs/releases/2026-10-13/rootfs-manifest.json)
identify the tested image. The builder uses Gentoo systemd for login/device
management, labwc for the minimal graphical session, and QEMU/KVM for guests.
Portage and the tracked QEMU/viewer patches remain available for ordinary updates.
Resume a build from the same source checkout and work directory used to create
its builder. Moving from a worktree to another checkout requires a fresh
`--work` directory under the project, or using the original worktree; provenance
checks deliberately reject an existing container mounted to a different source.
## Run the prototype
```sh
bin/hydra-install
-11
View File
@@ -1,11 +0,0 @@
#!/usr/bin/env python3
"""Source checkout entry point; installed copies use the installed Python package."""
from pathlib import Path
import sys
checkout = Path(__file__).resolve().parent.parent
if (checkout / 'hydra/installer.py').is_file():
sys.path.insert(0, str(checkout))
from hydra.installer import main
raise SystemExit(main())
dev/iso-smoke.py
-911
View File
@@ -1,911 +0,0 @@
#!/usr/bin/env python3
"""Boot/install/reboot acceptance on a fresh, isolated UEFI/KVM VM only.
No existing disk option, host sudo, SSH, host TTY, or GUI is used. Artifacts and
logs remain in a new private scratch directory even after failure.
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import inspect
import json
import os
from pathlib import Path
import re
import secrets
import select
import shlex
import shutil
import socket
import stat
import subprocess
import sys
import tempfile
import time
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hydra.qmp import QMPClient, QMPError
class SmokeError(RuntimeError):
pass
def lifecycle_socket(path, *, connect_socket=True):
path = Path(path)
info = path.lstat()
if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.getuid():
raise SmokeError('Lifecycle socket is not owned by the installed user')
identity = [info.st_dev, info.st_ino, info.st_uid, info.st_mode]
if connect_socket:
with socket.socket(socket.AF_UNIX) as connection:
connection.settimeout(2)
connection.connect(str(path))
after = path.lstat()
if identity != [after.st_dev, after.st_ino, after.st_uid, after.st_mode]:
raise SmokeError('Lifecycle socket changed while connecting')
return identity
def lifecycle_snapshot(context):
"""Prove live identity and owned runtime endpoints, never trust PID alone."""
try:
if context['uid'] != os.getuid():
raise SmokeError('Lifecycle probe is not the installed normal user')
runtime = Path(context['runtime'])
run = runtime / context['name']
directories = {}
for directory in (runtime, run):
info = directory.lstat()
if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid()
or stat.S_IMODE(info.st_mode) != 0o700):
raise SmokeError('Lifecycle runtime must remain private and owned')
directories[str(directory)] = [info.st_dev, info.st_ino, info.st_mode]
metadata = json.loads((run / 'identity.json').read_text())
if metadata != {'name': context['name'], 'uuid': context['uuid']}:
raise SmokeError('Lifecycle metadata is stale or foreign')
pid = int((run / 'qemu.pid').read_text())
process = Path('/proc') / str(pid)
if process.stat().st_uid != os.getuid():
raise SmokeError('Lifecycle process belongs to another user')
fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
if fields[0] == 'Z':
raise SmokeError('Lifecycle process has exited')
argv = (process / 'cmdline').read_bytes().split(b'\0')
for option, expected in ((b'-name', context['name']), (b'-uuid', context['uuid'])):
if argv[argv.index(option) + 1] != expected.encode():
raise SmokeError('Lifecycle PID is not the owned VM identity')
endpoints = {'qmp': lifecycle_socket(run / 'qmp.sock', connect_socket=False)}
with QMPClient(run / 'qmp.sock', timeout=2) as client:
if (client.command('query-name') != {'name': context['name']}
or client.command('query-uuid') != {'UUID': context['uuid']}):
raise SmokeError('Lifecycle QMP identity is foreign')
if client.command('query-status').get('status') != 'running':
raise SmokeError('Lifecycle VM is not running')
if not client.command('query-kvm').get('enabled'):
raise SmokeError('Lifecycle VM did not initialise nested KVM')
endpoints['spice'] = lifecycle_socket(run / 'spice.sock')
audio = Path(context['user_runtime'])
endpoints['pipewire'] = lifecycle_socket(audio / 'pipewire-0')
endpoints['pulse'] = lifecycle_socket(audio / 'pulse/native')
services = {}
for service in context.get('audio_services', []):
subprocess.run(['systemctl', '--user', 'is-active', service], check=True,
capture_output=True, timeout=5)
services[service] = subprocess.run(['systemctl', '--user', 'show', service,
'-p', 'InvocationID', '-p', 'ExecMainPID'], check=True,
capture_output=True, text=True, timeout=5).stdout
return {'name': context['name'], 'uuid': context['uuid'], 'pid': pid,
'process_start': fields[19], 'directories': directories,
'sockets': endpoints, 'audio_services': services}
except (OSError, ValueError, KeyError, IndexError, QMPError, subprocess.SubprocessError) as exc:
raise SmokeError('Lifecycle runtime or process was lost: ' + str(exc)) from exc
def verify_lifecycle_survival(before, after):
if before != after:
raise SmokeError('VM identity, private runtime, or audio endpoints changed across logout')
def login_session_records(uid):
"""Read logind's public properties, avoiding display-table row semantics."""
def query(*args):
return subprocess.run(['loginctl', *args], check=True, capture_output=True,
text=True, timeout=5).stdout
try:
ids = query('show-user', str(uid), '-p', 'Sessions', '--value').split()
records = []
for ident in ids:
output = query('show-session', ident, '--all', '--no-pager',
'-p', 'Id', '-p', 'User', '-p', 'State', '-p', 'Class',
'-p', 'Leader', '-p', 'TTY')
record = dict(line.split('=', 1) for line in output.splitlines())
if not {'Id', 'User', 'State', 'Class', 'Leader', 'TTY'} <= record.keys():
raise SmokeError('Incomplete logind session identity')
record['User'], record['Leader'] = int(record['User']), int(record['Leader'])
if record['Id'] != ident or record['User'] != uid:
raise SmokeError('Logind returned a foreign session identity')
records.append(record)
return records
except (OSError, ValueError, subprocess.SubprocessError) as exc:
raise SmokeError('Cannot establish guest login session state: ' + str(exc)) from exc
def is_open_login(record):
# systemd v258: closing session scopes can remain while user processes
# survive. Manager-class records describe bookkeeping, not a PAM login.
users = {'user', 'user-early', 'user-incomplete', 'user-light', 'user-early-light'}
bookkeeping = {'manager', 'manager-early', 'background', 'background-light',
'greeter', 'lock-screen', 'none'}
if record['State'] not in {'opening', 'online', 'active', 'closing'}:
raise SmokeError('Unknown logind session state')
if record['Class'] not in users | bookkeeping:
raise SmokeError('Unknown logind session class')
return record['Class'] in users and record['State'] != 'closing'
def require_initial_login(uid, ident):
opened = [record for record in login_session_records(uid) if is_open_login(record)]
if (len(opened) != 1 or opened[0]['Id'] != ident or opened[0]['TTY'] != 'ttyS0'
or opened[0]['Leader'] <= 0 or opened[0]['State'] not in {'online', 'active'}):
raise SmokeError('Lifecycle test requires exactly its actual serial PAM login')
return opened[0]
def no_open_logins(context):
original = context['login_session']
records = login_session_records(context['uid'])
for record in records:
if record['Id'] == original['Id']:
if (record['User'] != original['User'] or record['TTY'] != original['TTY']
or record['Class'] != original['Class']
or record['Leader'] not in {0, original['Leader']}):
raise SmokeError('Original PAM session identity was replaced')
if record['State'] != 'closing':
return False
if is_open_login(record):
return False
return True
def capture_logout_snapshot(context):
if not no_open_logins(context):
raise SmokeError('An open login overlaps the logout sample')
during = lifecycle_snapshot(context)
if not no_open_logins(context):
raise SmokeError('A login opened while the logout sample was running')
verify_lifecycle_survival(context['before'], during)
return during
LIFECYCLE_BODY = r'''
def run(*args):
return subprocess.run(args, check=True, capture_output=True, text=True, timeout=40).stdout
def cli(context, *args):
return json.loads(run('hydra', '--config', context['config'], '--state', context['state'],
'--runtime', context['runtime'], *args))
def write_report(path, data):
temporary = path.with_suffix('.tmp')
temporary.write_text(json.dumps(data, indent=2) + '\n')
temporary.chmod(0o600)
temporary.replace(path)
mode, base_arg = sys.argv[1:]
base = Path(base_arg)
context_path = base / 'context.json'
observation_path = base / 'logout-observation.json'
if mode == 'start':
if not Path('/dev/kvm').exists():
print('SKIP: installed guest nested KVM unavailable')
raise SystemExit(0)
if os.getuid() == 0 or not os.access('/dev/kvm', os.R_OK | os.W_OK):
raise SmokeError('Lifecycle test needs installed-user nested KVM access')
user_runtime = Path(os.environ['XDG_RUNTIME_DIR'])
if user_runtime != Path('/run/user') / str(os.getuid()):
raise SmokeError('Lifecycle test needs the real login runtime directory')
if run('loginctl', 'show-user', str(os.getuid()), '-p', 'Linger', '--value').strip() != 'no':
raise SmokeError('Lingering would mask the installed logout policy')
login_session = require_initial_login(os.getuid(), os.environ['XDG_SESSION_ID'])
name = 'logout-' + base.name.rsplit('-', 1)[1]
state = base / 'state'
runtime = user_runtime / base.name
context = {'uid': os.getuid(), 'name': name, 'config': str(base / 'config'),
'state': str(state), 'runtime': str(runtime), 'user_runtime': str(user_runtime),
'uuid': str(uuid.uuid5(uuid.NAMESPACE_URL, str(state / name / 'disk.qcow2'))),
'login_session': login_session,
'audio_services': ['pipewire.service', 'pipewire-pulse.service', 'wireplumber.service']}
run('systemctl', '--user', 'start', *context['audio_services'])
cli(context, 'add', name, '--memory', '256', '--cpus', '1', '--disk-size', '64M',
'--graphics', 'software', '--network', 'bridge', '--bridge', 'hydrabr0')
if cli(context, 'start', name).get('status') != 'running':
raise SmokeError('Real installed hydra CLI failed to start lifecycle VM')
try:
disk = Path(context['state']) / name / 'disk.qcow2'
info = disk.lstat()
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid():
raise SmokeError('Lifecycle VM disk is not a new owned regular qcow2')
context['before'] = lifecycle_snapshot(context)
except Exception:
# CLI control is scoped to the newly created private identity.
cli(context, 'stop', name, '--force')
raise
write_report(context_path, context)
with open(base / 'monitor.log', 'ab', buffering=0) as log:
subprocess.Popen([sys.executable, __file__, 'monitor', str(base)],
stdin=subprocess.DEVNULL, stdout=log, stderr=log,
start_new_session=True)
print('READY: normal-user Hydra VM and audio sockets recorded before real logout')
elif mode == 'monitor':
context = json.loads(context_path.read_text())
try:
deadline = time.monotonic() + 35
while not no_open_logins(context):
if time.monotonic() >= deadline:
raise SmokeError('Serial session did not actually log out')
time.sleep(.2)
# Exceed systemd's usual 10-second user-manager stop delay. No helper
# login or lingering is allowed to keep the tested user session alive.
time.sleep(15)
during = capture_logout_snapshot(context)
write_report(observation_path, {'status': 'passed', 'no_open_logins': True,
'survival_seconds': 15, 'during_logout': during})
except Exception as exc:
write_report(observation_path, {'status': 'failed', 'error': str(exc)})
raise
elif mode == 'finish':
context = json.loads(context_path.read_text())
try:
if os.environ.get('XDG_RUNTIME_DIR') != context['user_runtime']:
raise SmokeError('Relogin replaced the user runtime path')
deadline = time.monotonic() + 10
while not observation_path.is_file():
if time.monotonic() >= deadline:
raise SmokeError('Logout monitor died or did not record survival')
time.sleep(.2)
observation = json.loads(observation_path.read_text())
if observation.get('status') != 'passed' or observation.get('no_open_logins') is not True:
raise SmokeError('No-open-login observation failed: ' + json.dumps(observation))
after = lifecycle_snapshot(context)
verify_lifecycle_survival(context['before'], after)
if cli(context, 'status', context['name']).get('status') != 'running':
raise SmokeError('Real installed hydra CLI could not reconnect after login')
print(json.dumps({'status': 'passed', 'real_logout': True, 'no_open_logins': True,
'survival_seconds': observation['survival_seconds'], 'same_vm': after,
'audio_scope': 'Socket and service continuity only; signal/playback untested'}))
finally:
# Real product CLI verifies metadata and live QMP name/UUID before quit.
# Never signal a saved PID or control a VM outside these private paths.
cleanup_status = cli(context, 'status', context['name']).get('status')
if cleanup_status == 'running':
run_path = Path(context['runtime']) / context['name']
if json.loads((run_path / 'identity.json').read_text()) != {'name': context['name'], 'uuid': context['uuid']}:
raise SmokeError('Refusing lifecycle cleanup for a foreign identity')
with QMPClient(run_path / 'qmp.sock', timeout=2) as client:
if (client.command('query-name') != {'name': context['name']}
or client.command('query-uuid') != {'UUID': context['uuid']}):
raise SmokeError('Refusing lifecycle cleanup for foreign live QMP')
if cli(context, 'stop', context['name'], '--force').get('status') != 'stopped':
raise SmokeError('Owned lifecycle VM did not stop')
elif cleanup_status != 'stopped':
raise SmokeError('Cannot safely verify the lifecycle VM for cleanup')
else:
raise SmokeError('Unknown lifecycle mode')
'''
def lifecycle_guest_script():
helpers = (lifecycle_socket, lifecycle_snapshot, verify_lifecycle_survival,
login_session_records, is_open_login, require_initial_login,
no_open_logins, capture_logout_snapshot)
return ('import json, os, socket, stat, subprocess, sys, time, uuid\n'
'from pathlib import Path\nsys.path.insert(0, "/usr/lib/hydra")\n'
'from hydra.qmp import QMPClient, QMPError\n'
'class SmokeError(RuntimeError): pass\n' +
'\n'.join(inspect.getsource(helper) for helper in helpers) + LIFECYCLE_BODY)
def installed_admin(console, password):
console.buffer = b''
prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
console.send(f'sudo -k -p "$(printf %s {prompt_encoded} | base64 -d)" -i\n')
console.expect(rb'HYDRA_SUDO_PASSWORD: ', 30)
console.send(password + '\n')
console.expect(rb'root[^\r\n]*[#$] ', 30)
def installed_logout_probe(console, password):
# Exit only the guest sudo shell, then work in its actual getty PAM login.
console.send('exit\n')
console.expect(rb'hydratest[^\r\n]*[#$] ', 30)
if console.command('id -un', 10).strip() != b'hydratest':
raise SmokeError('Logout probe did not return to installed user login')
base = '/home/hydratest/hydra-logout-' + secrets.token_hex(4)
script = base + '/probe.py'
preparation = (f'umask 077; mkdir {shlex.quote(base)} && cat > {shlex.quote(script)} '
"<<'HYDRA_LIFECYCLE_SOURCE'\n" + lifecycle_guest_script() + '\nHYDRA_LIFECYCLE_SOURCE\n' +
f'python3 {shlex.quote(script)} start {shlex.quote(base)}')
ready = console.command(preparation, 120)
if ready.startswith(b'SKIP:'):
installed_admin(console, password)
return ready
if b'READY:' not in ready:
raise SmokeError('Lifecycle VM did not become ready before logout')
console.buffer = b''
console.send('exit\n')
console.expect(rb'login: ', 30)
# This is intentionally a period at the actual login prompt with no
# replacement login. The guest monitor independently confirms no open logins.
time.sleep(20)
console.send('hydratest\n')
console.expect(rb'Password: ', 30)
console.send(password + '\n')
console.expect(rb'hydratest[^\r\n]*[#$] ', 60)
result = console.command(f'python3 {shlex.quote(script)} finish {shlex.quote(base)}', 90)
installed_admin(console, password)
return result
def reserve_disk(path):
try:
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
os.close(fd)
except OSError as exc:
raise SmokeError('Test disk must be a newly created regular file') from exc
def owned_file(path, scratch):
path, scratch = Path(path), Path(scratch)
try:
info = path.lstat()
except OSError as exc:
raise SmokeError('Missing disposable drive') from exc
if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1
or path.parent.resolve() != scratch.resolve()
or info.st_uid != os.getuid()):
raise SmokeError('Drive must be an owned, unlinked regular file directly in scratch')
def qemu_path(path):
# QEMU keyval arguments use commas as separators. Refuse ambiguous paths.
if any(c in str(path) for c in ',\n\r\x00'):
raise SmokeError('QEMU file paths cannot contain comma or control characters')
return str(path)
def qemu_argv(qemu, scratch, disk, code, variables, iso, memory):
scratch = Path(scratch)
owned_file(disk, scratch)
owned_file(variables, scratch)
argv = [qemu, '-machine', 'q35,accel=kvm', '-cpu', 'host', '-smp', '2',
'-m', str(memory), '-display', 'none', '-monitor', 'none',
'-no-reboot', '-nic', 'user,model=virtio-net-pci',
'-drive', f'if=pflash,format=raw,readonly=on,file={qemu_path(code)}',
'-drive', f'if=pflash,format=raw,file={qemu_path(variables)}',
'-drive', f'if=virtio,format=qcow2,file={qemu_path(disk)}',
'-chardev', f'socket,id=console,path={qemu_path(scratch / "serial.sock")},server=on,wait=off',
'-serial', 'chardev:console', '-qmp', f'unix:{qemu_path(scratch / "qmp.sock")},server=on,wait=off']
if iso is not None:
if not Path(iso).is_file():
raise SmokeError('Trusted ISO must exist as a regular file')
argv += ['-drive', f'file={qemu_path(iso)},media=cdrom,format=raw,readonly=on',
'-boot', 'order=d,strict=on']
else:
argv += ['-boot', 'order=c,strict=on']
return argv
def frame_command(command, token):
if not re.fullmatch(r'[a-zA-Z0-9]+', token):
raise SmokeError('Invalid framing token')
encoded = base64.b64encode(command.encode()).decode()
# Linux canonical serial TTY input is bounded per physical line. Continue
# the shell expression across short lines for substantial guest probes.
chunks = ' \\\n'.join(shlex.quote(encoded[i:i + 768]) for i in range(0, len(encoded), 768)) or "''"
# Literal control bytes appear only in output, never in the echoed command.
return (f"printf '\\036{token}:BEGIN\\037'; "
f"sh -c \"$(printf %s {chunks} | base64 -d)\"; "
f"rc=$?; printf '\\036{token}:END:%s\\037' \"$rc\"")
def parse_frame(data, token):
begin = b'\x1e' + token.encode() + b':BEGIN\x1f'
start = data.find(begin)
if start < 0:
return None
end = re.search(rb'\x1e' + token.encode() + rb':END:(\d+)\x1f', data[start + len(begin):])
if end is None:
return None
output = data[start + len(begin):start + len(begin) + end.start()]
return int(end.group(1)), output
def installer_command(token):
# A literal control-byte return frame cannot be forged by terminal echo.
# Keep the installer on this controlling TTY for getpass, then mark every
# shell return, including an unexpected status zero without completion.
return ('sudo -n hydra-install --disk /dev/vda --user hydratest --execute '
'--confirm-erase /dev/vda; rc=$?; '
f"printf '\\036{token}:INSTALL_EXIT:%s\\037' \"$rc\"")
class Console:
def __init__(self, connection, logfile, passwords):
self.sock = connection
self.log = open(logfile, 'ab', buffering=0)
self.passwords = [p.encode() for p in passwords if p]
self.pending_log = b''
self.buffer = b''
self.closed = False
def record(self, data, final=False):
self.pending_log += data
for password in self.passwords:
self.pending_log = self.pending_log.replace(password, b'[REDACTED]')
keep = 0
if not final:
for password in self.passwords:
for length in range(1, len(password)):
if self.pending_log.endswith(password[:length]):
keep = max(keep, length)
if keep:
self.log.write(self.pending_log[:-keep])
self.pending_log = self.pending_log[-keep:]
else:
self.log.write(self.pending_log)
self.pending_log = b''
def close(self):
if not self.closed:
self.record(b'', final=True)
self.log.close()
self.sock.close()
self.closed = True
def send(self, data):
self.sock.sendall(data.encode() if isinstance(data, str) else data)
def read(self, timeout):
ready, _, _ = select.select([self.sock], [], [], max(0, timeout))
if not ready:
return b''
data = self.sock.recv(65536)
if not data:
raise SmokeError('Guest console closed')
self.record(data)
self.buffer += data
if len(self.buffer) > 16 * 1024 * 1024:
raise SmokeError('Console exceeded bounded capture size')
return data
def expect(self, pattern, timeout, failure_pattern=None):
deadline = time.monotonic() + timeout
while True:
match = re.search(pattern, self.buffer)
failure = re.search(failure_pattern, self.buffer) if failure_pattern else None
if failure and (not match or failure.start() < match.start()):
evidence = failure.group()
for password in self.passwords:
evidence = evidence.replace(password, b'[REDACTED]')
raise SmokeError('Installer exited before expected console state: ' +
evidence.decode(errors='replace'))
if match:
self.buffer = self.buffer[match.end():]
return match
remaining = deadline - time.monotonic()
if remaining <= 0:
raise SmokeError('Timed out waiting for guest console state')
self.read(min(remaining, 1))
def expect_installer(self, pattern, timeout, token):
failure = (rb'hydra-install: [^\r\n]*[\r\n]|\x1e' +
re.escape(token.encode()) + rb':INSTALL_EXIT:[0-9]+\x1f')
return self.expect(pattern, timeout, failure_pattern=failure)
def finish_installer(self, token, timeout=30):
# Completion text can arrive before the process exits and readline
# resumes. Input sent during that gap can be discarded by the TTY.
deadline = time.monotonic() + timeout
returned = self.expect(rb'\x1e' + re.escape(token.encode()) +
rb':INSTALL_EXIT:([0-9]+)\x1f', timeout)
status = int(returned.group(1))
if status:
raise SmokeError(f'Installer returned status {status} after completion')
self.expect(rb'hydra-live[^\r\n]*[#$] ', max(0, deadline - time.monotonic()))
def command(self, command, timeout=60):
token = secrets.token_hex(12)
self.buffer = b''
self.send(frame_command(command, token) + '\n')
deadline = time.monotonic() + timeout
while True:
parsed = parse_frame(self.buffer, token)
if parsed is not None:
status, output = parsed
self.buffer = b''
if status:
raise SmokeError(f'Guest probe failed with status {status}: {output.decode(errors="replace")[-3000:]}')
return output.replace(b'\r\n', b'\n')
remaining = deadline - time.monotonic()
if remaining <= 0:
raise SmokeError('Timed out waiting for framed guest command')
self.read(min(remaining, 1))
def connect(path, process, timeout=30):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if process.poll() is not None:
raise SmokeError(f'Test QEMU exited with status {process.returncode}')
connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
connection.connect(str(path))
return connection
except OSError:
connection.close()
time.sleep(0.1)
raise SmokeError('Timed out connecting to private test socket')
class VM:
def __init__(self, args, scratch, disk, iso, password, phase):
self.scratch = scratch
for name in ('serial.sock', 'qmp.sock'):
(scratch / name).unlink(missing_ok=True)
self.stderr = open(scratch / f'{phase}-qemu.log', 'ab', buffering=0)
self.process = subprocess.Popen(qemu_argv(args.qemu, scratch, disk,
args.ovmf_code, scratch / 'OVMF_VARS.fd', iso, args.memory),
stdin=subprocess.DEVNULL, stdout=self.stderr, stderr=self.stderr)
self.console = None
self.qmp = None
try:
self.console = Console(connect(scratch / 'serial.sock', self.process),
scratch / f'{phase}-serial.log', [password])
ready = connect(scratch / 'qmp.sock', self.process)
ready.close()
self.qmp = QMPClient(scratch / 'qmp.sock', timeout=5)
self.kvm = self.qmp.command('query-kvm')
if not self.kvm.get('enabled'):
raise SmokeError('Disposable QEMU is not using KVM')
except BaseException:
self.close()
raise
def wait_poweroff(self, timeout):
deadline = time.monotonic() + timeout
serial_open = True
while self.process.poll() is None:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise SmokeError('Guest failed to power off in time')
if serial_open:
try:
self.console.read(min(remaining, 0.25))
except SmokeError as exc:
if str(exc) != 'Guest console closed':
raise
serial_open = False
else:
try:
self.process.wait(timeout=remaining)
except subprocess.TimeoutExpired as exc:
raise SmokeError('Guest failed to power off in time') from exc
# Capture bytes queued immediately before exit, even if poll observed
# the child exiting before the first serial read.
while serial_open:
try:
if not self.console.read(0):
break
except SmokeError as exc:
if str(exc) != 'Guest console closed':
raise
serial_open = False
rc = self.process.returncode
if rc:
raise SmokeError(f'Test QEMU failed during poweroff ({rc})')
def close(self):
# Only the Popen child owned by this harness is signalled.
if self.qmp:
if self.process.poll() is None:
try:
self.qmp.command('quit')
self.process.wait(timeout=5)
except (QMPError, subprocess.TimeoutExpired):
pass
self.qmp.close()
if self.process.poll() is None:
self.process.terminate()
try:
self.process.wait(timeout=10)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait(timeout=10)
if self.console:
self.console.close()
self.stderr.close()
LIVE_CHECK = r'''set -eu
[ "$(id -un)" = hydra-live ]
test -e /sys/firmware/efi
test -f /etc/hydra/live-mode/active
findmnt -n -o FSTYPE / | grep -qx overlay
test -f /run/initramfs/live/LiveOS/squashfs.img
sudo -n true
for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
for service in NetworkManager hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
'''
INSTALLED_CHECK = r'''set -eu
assert_no_match() {
if grep "$@"; then
printf 'Forbidden live configuration remains\n' >&2
exit 1
else
status=$?
[ "$status" -eq 1 ] || exit "$status"
fi
}
[ "$(findmnt -n -o FSTYPE /)" = ext4 ]
[ "$(readlink -f "$(findmnt -n -o SOURCE /)")" = /dev/vda2 ]
assert_no_match -Eq 'root=live:|rd\.live' /proc/cmdline
test ! -e /etc/hydra/live-mode
test ! -L /etc/hydra/live-mode
if getent passwd hydra-live; then
printf 'Live account remains\n' >&2
exit 1
else
status=$?
[ "$status" -eq 2 ] || exit "$status"
fi
remaining=$(find /etc/systemd/system /etc/sudoers.d \( -name 90-hydra-live.conf -o -name 90-hydra-live \) -print)
[ -z "$remaining" ]
test ! -e /etc/greetd/live.toml
test ! -L /etc/greetd/live.toml
for directory in /etc/systemd/system/getty@*.service.d /etc/systemd/system/serial-getty@*.service.d /etc/sudoers.d; do
if [ -d "$directory" ]; then
assert_no_match -R -E -- '--autologin|NOPASSWD' "$directory"
elif [ -e "$directory" ] || [ -L "$directory" ]; then
printf 'Unexpected non-directory configuration path\n' >&2
exit 1
fi
done
getent shadow root | cut -d: -f2 | grep -Eq '^[!*]'
getent shadow hydratest | cut -d: -f2 | grep -Eq '^\$'
for service in NetworkManager greetd hydra-network hydra-dnsmasq; do systemctl is-active "$service"; done
[ "$(systemctl show hydra-bridge-permissions -p Result --value)" = success ]
[ "$(systemctl show hydra-bridge-permissions -p ExecMainStatus --value)" = 0 ]
ip -4 -o addr show hydrabr0 | grep -F '10.203.0.1/24'
ip -4 route show default | grep '^default '
nft list table ip hydra_nat
nmcli -t -f STATE general | grep -E '^connected'
getent ahostsv4 gentoo.org
for tool in hydra hydra-install qemu-system-x86_64 qemu-img remote-viewer foot labwc; do command -v "$tool"; done
hydra --help
qemu-system-x86_64 --version
qemu-img --version
test -r /etc/greetd/config.toml
find /usr/share -name OVMF_CODE.fd -print -quit | grep .
'''
NETWORK_CHECK = r'''set -eu
[ "$(cat /proc/sys/net/ipv4/ip_forward)" = 1 ]
python3 - <<'NETWORK'
import pathlib, secrets, subprocess, time
tag = secrets.token_hex(3)
lan, guest = 'hl' + tag, 'hv' + tag
uplink, lanpeer = 'hu' + tag, 'hp' + tag
bridgepeer, guestpeer = 'hb' + tag, 'hg' + tag
created = []
links = []
server = None
def run(*args):
return subprocess.run(args, check=True, capture_output=True, text=True, timeout=20).stdout
try:
for name in (lan, guest):
run('ip', 'netns', 'add', name)
created.append(name)
run('ip', 'link', 'add', uplink, 'type', 'veth', 'peer', 'name', lanpeer)
links.append(uplink)
run('ip', 'link', 'set', lanpeer, 'netns', lan)
run('ip', 'addr', 'add', '198.18.0.1/30', 'dev', uplink)
run('ip', 'link', 'set', uplink, 'up')
run('ip', '-n', lan, 'addr', 'add', '198.18.0.2/30', 'dev', lanpeer)
run('ip', '-n', lan, 'link', 'set', lanpeer, 'up')
run('ip', '-n', lan, 'link', 'set', 'lo', 'up')
run('ip', '-n', lan, 'route', 'add', '10.203.0.0/24', 'via', '198.18.0.1')
run('ip', 'link', 'add', bridgepeer, 'type', 'veth', 'peer', 'name', guestpeer)
links.append(bridgepeer)
run('ip', 'link', 'set', guestpeer, 'netns', guest)
run('ip', 'link', 'set', bridgepeer, 'master', 'hydrabr0')
run('ip', 'link', 'set', bridgepeer, 'up')
run('ip', '-n', guest, 'addr', 'add', '10.203.0.250/24', 'dev', guestpeer)
run('ip', '-n', guest, 'link', 'set', guestpeer, 'up')
run('ip', '-n', guest, 'link', 'set', 'lo', 'up')
run('ip', '-n', guest, 'route', 'add', 'default', 'via', '10.203.0.1')
port = 43117
server_code = "import socket; s=socket.socket(); s.settimeout(15); s.bind(('10.203.0.250',43117)); s.listen(1); c,a=s.accept(); c.sendall(b'HYDRA_LAN_TO_VM_OK'); c.close(); s.close()"
server = subprocess.Popen(['ip', 'netns', 'exec', guest, 'python3', '-c', server_code], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
client_code = "import socket,time; deadline=time.monotonic()+10\nwhile True:\n try:\n s=socket.create_connection(('10.203.0.250',43117),1); break\n except OSError:\n if time.monotonic()>deadline: raise\n time.sleep(.1)\nexpected=b'HYDRA_LAN_TO_VM_OK'; received=b''\nwhile len(received)<len(expected):\n remaining=deadline-time.monotonic()\n if remaining<=0: raise TimeoutError('TCP marker deadline exceeded')\n s.settimeout(remaining)\n part=s.recv(len(expected)-len(received))\n if not part: raise RuntimeError('Truncated TCP marker')\n received+=part\nassert received==expected; s.close(); print('PASS: routed new LAN-to-VM TCP connection on arbitrary port 43117')"
print(run('ip', 'netns', 'exec', lan, 'python3', '-c', client_code), end='')
assert server.wait(timeout=10) == 0
finally:
if server is not None and server.poll() is None:
server.terminate()
try: server.wait(timeout=5)
except subprocess.TimeoutExpired: server.kill(); server.wait(timeout=5)
for link in reversed(links):
subprocess.run(['ip', 'link', 'del', link], check=False, capture_output=True, timeout=10)
for name in reversed(created):
subprocess.run(['ip', 'netns', 'del', name], check=False, capture_output=True, timeout=10)
NETWORK
'''
# A nested VM hardware smoke proves KVM initialisation/QMP, not guest OS boot.
NESTED_CHECK = r'''set -eu
if ! test -c /dev/kvm; then printf 'SKIP: guest /dev/kvm unavailable\n'; exit 0; fi
python3 - <<'NESTED'
import json, pathlib, socket, subprocess, tempfile, time
with tempfile.TemporaryDirectory(prefix='hydra-nested-') as tmp:
root = pathlib.Path(tmp)
disk = root / 'fresh.qcow2'
subprocess.run(['qemu-img', 'create', '-f', 'qcow2', str(disk), '64M'], check=True, timeout=30)
log = open(root / 'qemu.log', 'wb')
proc = subprocess.Popen(['qemu-system-x86_64', '-machine', 'q35,accel=kvm', '-cpu', 'host', '-m', '128', '-S', '-display', 'none', '-nodefaults', '-nic', 'bridge,br=hydrabr0,model=virtio-net-pci', '-drive', 'if=virtio,format=qcow2,file=' + str(disk), '-qmp', 'unix:' + str(root / 'qmp') + ',server=on,wait=off'], stdin=subprocess.DEVNULL, stdout=log, stderr=log)
sock = socket.socket(socket.AF_UNIX)
sock.settimeout(10)
try:
deadline = time.monotonic() + 15
while True:
try: sock.connect(str(root / 'qmp')); break
except OSError:
if proc.poll() is not None or time.monotonic() >= deadline: raise RuntimeError('Nested KVM start failed')
time.sleep(.1)
stream = sock.makefile('rwb', buffering=0)
assert 'QMP' in json.loads(stream.readline())
for request in ({'execute':'qmp_capabilities'}, {'execute':'query-kvm'}, {'execute':'quit'}):
stream.write(json.dumps(request).encode() + b'\n')
while True:
response = json.loads(stream.readline())
if 'return' in response or 'error' in response: break
assert 'error' not in response, response
if request['execute'] == 'query-kvm':
assert response['return']['enabled'], response
print('PASS: normal-user nested KVM VM and bridge helper; fresh qcow2; private QMP')
assert proc.wait(timeout=10) == 0
finally:
sock.close()
if proc.poll() is None:
proc.terminate()
try: proc.wait(timeout=5)
except subprocess.TimeoutExpired: proc.kill(); proc.wait(timeout=5)
log.close()
NESTED
'''
def digest(path):
result = hashlib.sha256()
with open(path, 'rb') as source:
for block in iter(lambda: source.read(1024 * 1024), b''):
result.update(block)
return result.hexdigest()
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--iso', required=True, type=Path)
parser.add_argument('--ovmf-code', required=True, type=Path)
parser.add_argument('--ovmf-vars', required=True, type=Path)
parser.add_argument('--qemu', default='qemu-system-x86_64')
parser.add_argument('--qemu-img', default='qemu-img')
parser.add_argument('--scratch-parent', type=Path, default=Path(tempfile.gettempdir()))
parser.add_argument('--memory', type=int, default=4096)
parser.add_argument('--boot-timeout', type=int, default=360)
parser.add_argument('--install-timeout', type=int, default=1800)
args = parser.parse_args(argv)
for path in (args.iso, args.ovmf_code, args.ovmf_vars):
if not path.is_file():
parser.error(f'Missing input file: {path}')
if args.memory < 2048 or args.boot_timeout <= 0 or args.install_timeout <= 0:
parser.error('Memory must be >=2048 MiB and timeouts positive')
args.iso = args.iso.resolve()
args.ovmf_code = args.ovmf_code.resolve()
args.ovmf_vars = args.ovmf_vars.resolve()
scratch = Path(tempfile.mkdtemp(prefix='hydra-iso-smoke-', dir=args.scratch_parent))
if len(str(scratch / 'serial.sock').encode()) >= 104:
parser.error('Scratch path too long for Unix sockets; use /tmp')
print(f'Acceptance artifacts: {scratch}', flush=True)
report = {'schema': 1, 'status': 'running', 'scratch': str(scratch),
'iso': str(args.iso), 'iso_sha256': digest(args.iso),
'checks': {}, 'physical_hardware': 'pending',
'nested_scope': 'Normal-user KVM VM launch, bridge helper and QMP; no nested guest OS boot',
'ovmf_code_sha256': digest(args.ovmf_code),
'ovmf_vars_input_sha256': digest(args.ovmf_vars)}
password = secrets.token_urlsafe(24)
sentinel = secrets.token_hex(32)
vm = None
def checkpoint(name, output):
report['checks'][name] = {'status': 'skipped' if output.startswith(b'SKIP:') else 'passed',
'output': output.decode(errors='replace')}
(scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
try:
disk = scratch / 'fresh-install.qcow2'
reserve_disk(disk)
subprocess.run([args.qemu_img, 'create', '-f', 'qcow2', str(disk), '32G'], check=True, timeout=30,
stdout=subprocess.DEVNULL)
shutil.copyfile(args.ovmf_vars, scratch / 'OVMF_VARS.fd')
os.chmod(scratch / 'OVMF_VARS.fd', 0o600)
vm = VM(args, scratch, disk, args.iso, password, 'live')
checkpoint('live_qmp_kvm', json.dumps(vm.kvm).encode())
console = vm.console
console.expect(rb'hydra-live[^\r\n]*[#$] ', args.boot_timeout)
checkpoint('live_boot', console.command(LIVE_CHECK, 90))
console.buffer = b''
installer_token = secrets.token_hex(12)
console.send(installer_command(installer_token) + '\n')
console.expect_installer(rb'Installed account password: ', 120, installer_token)
console.send(password + '\n')
console.expect_installer(rb'Repeat password: ', 30, installer_token)
console.send(password + '\n')
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
args.install_timeout, installer_token)
console.finish_installer(installer_token)
checkpoint('real_installer', b'Installer completed via guest controlling TTY')
console.command("sudo -n mkdir -p /mnt/hydra-validation && sudo -n mount /dev/vda2 /mnt/hydra-validation && printf %s " +
shlex.quote(sentinel) + " | sudo -n tee /mnt/hydra-validation/var/lib/hydra-acceptance-sentinel >/dev/null && sudo -n umount /mnt/hydra-validation", 60)
console.send('sudo -n poweroff\n')
vm.wait_poweroff(90)
vm.close()
vm = None
checkpoint('live_poweroff', b'QEMU exited before ISO-free reboot')
vm = VM(args, scratch, disk, None, password, 'installed')
checkpoint('installed_qmp_kvm', json.dumps(vm.kvm).encode())
console = vm.console
console.expect(rb'login: ', args.boot_timeout)
console.send('hydratest\n')
console.expect(rb'Password: ', 30)
console.send(password + '\n')
console.expect(rb'hydratest[^\r\n]*[#$] ', 60)
output = console.command('id -un; if sudo -n true 2>/dev/null; then exit 1; fi', 30)
if output.strip() != b'hydratest':
raise SmokeError('Installed account identity mismatch')
checkpoint('installed_password_login', output)
console.buffer = b''
prompt_encoded = base64.b64encode(b'HYDRA_SUDO_PASSWORD: ').decode()
console.send(f'sudo -p "$(printf %s {prompt_encoded} | base64 -d)" -i\n')
console.expect(rb'HYDRA_SUDO_PASSWORD: ', 30)
console.send(password + '\n')
console.expect(rb'root[^\r\n]*[#$] ', 30)
observed = console.command('cat /var/lib/hydra-acceptance-sentinel', 30)
if observed != sentinel.encode():
raise SmokeError('Persistence sentinel did not survive installed disk boot')
checkpoint('disk_persistence', b'Sentinel content verified from command output frame')
checkpoint('installed_contract', console.command(INSTALLED_CHECK, 120))
checkpoint('lan_to_vm_tcp', console.command(NETWORK_CHECK, 120))
checkpoint('nested_kvm', console.command('runuser -u hydratest -- sh -c ' + shlex.quote(NESTED_CHECK), 90))
checkpoint('installed_user_logout', installed_logout_probe(console, password))
console.send('poweroff\n')
vm.wait_poweroff(90)
checkpoint('installed_poweroff', b'Normal installed guest shutdown')
report['status'] = 'passed'
return 0
except (SmokeError, QMPError, OSError, subprocess.SubprocessError) as exc:
report['status'] = 'failed'
report['error'] = str(exc).replace(password, '[REDACTED]')
print(f'Acceptance failed: {report["error"]}', file=sys.stderr)
return 1
finally:
if vm:
vm.close()
report['completed_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
(scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n')
print(f'Report: {scratch / "report.json"}', flush=True)
if __name__ == '__main__':
raise SystemExit(main())
docs/install.md
-94
View File
@@ -1,94 +0,0 @@
# Install HydraOS
HydraOS targets x86_64 UEFI systems. Disable Secure Boot for this first image.
Boot the ISO and open the recovery terminal. The live session uses a locked
`hydra-live` account with local sudo access; changes in the live filesystem are
volatile. The image provides no default SSH login or universal installed password.
`hydra-install` lists disks without selecting one. Inspect the devices, then ask
for a plan using the whole disk you intend to erase:
```sh
hydra-install
sudo hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER
```
Replace the placeholders with an actual whole disk and your desired account
name. The installer rejects mounted devices, the running root, the live medium,
and their backing disks. These checks help prevent mistakes; verify the chosen
disk and preserve backups before executing the plan. Physical disk installation
requires the operator's explicit choice.
To execute, enter the identical resolved whole-disk path in the confirmation:
```sh
sudo hydra-install --disk /dev/CHOSEN_DISK --user YOUR_USER \
--execute --confirm-erase /dev/CHOSEN_DISK
```
Execution erases the disk, creates a 1 GiB FAT32 EFI partition and an ext4 root,
extracts the prepared payload, and prompts twice on the local terminal for your
installed account password. Passwords have no command-line or configuration
option. The installed root account stays locked. Defaults are Europe/Moscow and
English UTF-8, with Russian UTF-8 also available; `--timezone` accepts an IANA
zone present in the payload.
The installer removes the live account, local password-free sudo rule, and
live autologin overrides, then builds a normal disk initramfs and fallback UEFI
loader. Wait for `Installation completed`, power off, remove the ISO, and boot
the installed disk. Log in with the account and password you supplied. On
failure, read the printed live diagnostic path; when a target root was mounted,
a nonsecret snapshot is also saved at `/var/log/hydra-install.log` on that disk.
## Disposable VM acceptance
Run the repository harness with QEMU/KVM, `qemu-img`, Python 3, and a generic
x86_64 QEMU OVMF CODE/VARS pair. A Xen-specific OVMF build is unsuitable: the
first validation attempt with that variant produced no guest boot output.
Use trusted image and matching firmware files. For example:
```sh
install -d -m 700 "$HOME/.cache/hydra-tests"
python3 dev/iso-smoke.py --iso /absolute/path/to/hydra-os.iso \
--ovmf-code /absolute/path/to/OVMF_CODE.fd \
--ovmf-vars /absolute/path/to/OVMF_VARS.fd \
--scratch-parent "$HOME/.cache/hydra-tests"
```
The harness creates a private directory under `--scratch-parent` (`/tmp` by default), a new 32 GiB sparse
qcow2 disk, and a private copy of OVMF variables. There is no existing disk
argument. QEMU attaches the ISO read-only; serial and QMP use private Unix
sockets. Tool overrides `--qemu` and `--qemu-img` support temporary development
environments. `--scratch-parent`, `--memory`, `--boot-timeout`, and
`--install-timeout` control retained artifacts and bounded resource use.
Prefer a disk-backed scratch directory: an installed disk stored on a tmpfs
such as `/tmp` consumes host RAM. The example keeps test data in the user's cache.
The test boots the live image, runs the real installer through the guest's
controlling terminal using a generated password, writes a persistence sentinel,
shuts down, and launches a second QEMU with the ISO absent. It verifies password
login, disk root, removed live overrides, services, bridge/NAT configuration,
DNS, recovery tools, and a routed new TCP connection from a synthetic LAN to a
synthetic VM endpoint. Network namespaces and veth interfaces exist only inside
the disposable HydraOS guest. When guest KVM is available, it launches a further
fresh-disk QEMU as the installed account to exercise KVM, the bridge helper,
and private QMP; this checks VM launch, not a nested operating system boot.
It also logs the installed account out and back in, checking that the same VM,
private sockets and user audio services survive a measured interval without an
open login session. This establishes process/socket continuity, not audio playback.
`report.json` records each observed check, ISO/firmware SHA256, failures, and
skipped nested virtualization. Serial/QEMU logs and the qcow2 remain in the
printed scratch directory for review; passwords are redacted even across serial
read boundaries. Cleanup controls only the QEMU process created by this run.
A passing unit suite establishes harness behavior, not ISO acceptance. An
actual successful report from the built ISO is required for boot/install/reboot
acceptance.
## Checks requiring the laptop
Disposable QEMU results cannot establish ASUS FX517ZC hardware support. Before
using HydraOS daily, boot the live ISO and observe Wi-Fi, touchpad, audio,
brightness, suspend/resume, USB device selection, input capture, and actual
display refresh. Physical installation and these checks remain pending until
explicitly performed and recorded.
docs/releases/2026-10-13/acceptance.json
-57
View File
@@ -1,57 +0,0 @@
{
"format": 1,
"status": "passed",
"completed_utc": "2026-10-13T02:49:50Z",
"artifact": {
"file": "hydra-os.iso",
"bytes": 2345109504,
"sha256": "bdff621037ce9220767b7869499823c9b840b76c0b38268bc7f56a7b54390ce0",
"source_commit": "903e7cf35339e50783a1bb3790b06e3204efab4d",
"source_inputs_sha256": "cfe9ddf4549bdedcd91888adc8c4b00639e370b6296c361898d3898daf76d4de",
"kernel": "6.18.54-gentoo-dist-bin",
"package_count": 559,
"rootfs_manifest_sha256": "c53ce549e98e0b5a56091d7960a2737534882e3f64a2497a12e3b2a27e522c48"
},
"validation": {
"harness_commit": "4a636924df4e630788dcbbe8d0002261fa9a7df9",
"harness_sha256": "64863d2f856cd3cee1e0c28af5aac2db2706b6403bf5deb32f64f57c33df1a53",
"ovmf_code_sha256": "f3ff7e73448ed2845ee15356f394882f5618eb5dab92c9a30ec6ee0e1468553a",
"ovmf_vars_input_sha256": "6ed987af3a3c155be71665f510eae3e007eda9b8b94afd59d45e91c4a11565cc",
"tools": "Ubuntu QEMU8.2.2, Python3.12.3",
"disk": "new 32 GiB sparse qcow2, only this disposable VM",
"iso_unmodified": true,
"manual_guest_changes": false,
"full_private_report_sha256": "accfc85840681da69eeacabf82c9007dceb66952dac907e0e545b1061a9dee10",
"checks": {
"live_qmp_kvm": "passed",
"live_boot": "passed",
"real_installer": "passed",
"live_poweroff": "passed",
"installed_qmp_kvm": "passed",
"installed_password_login": "passed",
"disk_persistence": "passed",
"installed_contract": "passed",
"lan_to_vm_tcp": "passed",
"nested_kvm": "passed",
"installed_user_logout": "passed",
"installed_poweroff": "passed"
}
},
"scope": {
"nested_virtualization": "Normal-user KVM VM launch, bridge helper and QMP; no nested guest OS boot",
"audio": "Socket/service continuity only; signal/playback untested",
"logout": "Actual getty logout, no open login for measured 15-second survival interval, relogin; unchanged VM process/start/UUID, runtime directories, private sockets and user audio service identities",
"graphics": "Private graphical live recovery observed; actual two-guest labwc/viewer and profile-refresh probes recorded separately; no physical scanout assertion"
},
"pending_hardware": [
"ASUS live boot",
"Wi-Fi and Ethernet on laptop",
"touchpad and physical input capture",
"brightness",
"physical audio playback",
"suspend/resume",
"USB device attachment and guest drivers",
"physical display refresh and delivered FPS"
],
"publication": "Source, patches and these records are in Git. ISO, VM disks, raw logs and private runtime identities remain local."
}
docs/releases/2026-10-13/hydra-os.iso.sha256
-1
View File
@@ -1 +0,0 @@
bdff621037ce9220767b7869499823c9b840b76c0b38268bc7f56a7b54390ce0 hydra-os.iso
docs/releases/2026-10-13/rootfs-manifest.json
-6399
View File
File diff suppressed because it is too large Load Diff
docs/superpowers/plans/2026-10-13-gentoo-image.md
+19 -19
View File
@@ -34,40 +34,40 @@
**Files:** `image/sources.json`, `image/packages`, `image/portage/`, `scripts/hydra-build`, `hydra/build.py`, `tests/test_build.py`.
**Interfaces:** `hydra-build fetch|prepare|packages|rootfs|iso|status`, `--work PATH`, `--cache PATH`, optional `--podman PATH`. Each step resumable and validates prerequisite manifests. Outputs WORK/rootfs payload or dedicated builder container, WORK/artifacts/, build manifest. No host root required.
- [x] Test pinned download hash mismatch, required signaturefingerprint, safe workpath/refusal of root, resume state vs missingactualartifact, scoped Podman argv.
- [x] Observe RED; implement verifiedfetch and isolatedimport/container naming. Use stage3 20261011T170049Z SHA256 50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b and Portage4696089e782c20b5f0c04952f6698d6a8cc1acb4 archiveSHA256 d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca.
- [x] Configure officialbinhost signaturechecking, selective ~amd64 for labwc/wlroots, QEMU x86_64 only, required SPICE/GL/USB flags. Preserve licenses and installed packageDB.
- [x] Run actual dependency resolution and package install; record exact versions/USE and any selective sourcebuild decisions. No claim of bit-identical build.
- [x] Run tests and commit builder inputs/scripts.
- [ ] Test pinned download hash mismatch, required signaturefingerprint, safe workpath/refusal of root, resume state vs missingactualartifact, scoped Podman argv.
- [ ] Observe RED; implement verifiedfetch and isolatedimport/container naming. Use stage3 20261011T170049Z SHA256 50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b and Portage4696089e782c20b5f0c04952f6698d6a8cc1acb4 archiveSHA256 d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca.
- [ ] Configure officialbinhost signaturechecking, selective ~amd64 for labwc/wlroots, QEMU x86_64 only, required SPICE/GL/USB flags. Preserve licenses and installed packageDB.
- [ ] Run actual dependency resolution and package install; record exact versions/USE and any selective sourcebuild decisions. No claim of bit-identical build.
- [ ] Run tests and commit builder inputs/scripts.
### Task 2: Host rootfs and bootable image
**Files:** `image/rootfs/`, `image/grub.cfg`, `image/dracut.conf`, `image/configure-rootfs.sh`, `tests/test_image.py`.
**Interfaces:** Consumes builder and tested CLI/session from graphics plan. Produces labelled HYDRA_OS ISO with boot/vmlinuz, boot/initramfs-live.img, LiveOS/squashfs.img and package/buildmanifest.
- [x] Test service/config generation, live-only overrides separation, exact kernelversion used for dracut, image label/cmdline agreement, networksubnet conflict handling.
- [x] Implement rootfs overlay: greetd/minimalsession, recovery, localliveadministration, installedauth baseline, audio, NetworkManager, hydrabr0/DHCP/NAT and narrowlyscoped bridgehelper. Include laptopfirmware/bootdrivers and OVMF.
- [x] Build genericliveinitramfs with dmsquash-live and OverlayFS; squashfspayload retains kernel/packageDB. Assemble UEFIISO with GRUB; save SHA256 and manifest.
- [x] Run image checks and disposable OVMF/KVM liveboot; verify rootfilesystem ready and expected tools. Commit rootfs/imageconfig.
- [ ] Test service/config generation, live-only overrides separation, exact kernelversion used for dracut, image label/cmdline agreement, networksubnet conflict handling.
- [ ] Implement rootfs overlay: greetd/minimalsession, recovery, localliveadministration, installedauth baseline, audio, NetworkManager, hydrabr0/DHCP/NAT and narrowlyscoped bridgehelper. Include laptopfirmware/bootdrivers and OVMF.
- [ ] Build genericliveinitramfs with dmsquash-live and OverlayFS; squashfspayload retains kernel/packageDB. Assemble UEFIISO with GRUB; save SHA256 and manifest.
- [ ] Run image checks and disposable OVMF/KVM liveboot; verify rootfilesystem ready and expected tools. Commit rootfs/imageconfig.
### Task 3: Explicit guarded installer
**Files:** `bin/hydra-install`, `hydra/installer.py`, `tests/test_installer.py`.
**Interfaces:** `hydra-install --disk DEVICE --user USER [--execute --confirm-erase DEVICE]`; without execute prints exactplan. Environment/default rootfs path points to live squashfs, never running root. Password via getpass/TTY, never CLI/log. Separate nonsecret dryrun from execution.
- [x] Test missing/mismatched confirmation, disk mounted directly or through children/ancestors, root/live source backingdisk, invaliddevice, symlink resolution, invalidusername, failedcommands abort subsequentsteps.
- [x] Observe RED; implement sysfs/lsblk topology validation, GPT1GiBESP/ext4, payload extraction, fstabUUIDs, user/password setup and liveoverridesremoval. Explicit target, no automaticdiscoveryselection.
- [x] Install regularinitramfs and GRUBfallback --removable --no-nvram. Checked subprocessargv; handle failure diagnostics and mounts cleanup without deletingtargetdata.
- [x] Test fakecommandfailure paths then complete actualinstallation inside fresh VM using disposable qcow2. Rootprivilege exists only in testguest.
- [x] Commit installer and negative tests.
- [ ] Test missing/mismatched confirmation, disk mounted directly or through children/ancestors, root/live source backingdisk, invaliddevice, symlink resolution, invalidusername, failedcommands abort subsequentsteps.
- [ ] Observe RED; implement sysfs/lsblk topology validation, GPT1GiBESP/ext4, payload extraction, fstabUUIDs, user/password setup and liveoverridesremoval. Explicit target, no automaticdiscoveryselection.
- [ ] Install regularinitramfs and GRUBfallback --removable --no-nvram. Checked subprocessargv; handle failure diagnostics and mounts cleanup without deletingtargetdata.
- [ ] Test fakecommandfailure paths then complete actualinstallation inside fresh VM using disposable qcow2. Rootprivilege exists only in testguest.
- [ ] Commit installer and negative tests.
### Task 4: Installed-system validation and publication
**Files:** `dev/iso-smoke.py`, `docs/install.md`, `docs/verification.md`, `README.md`.
**Interfaces:** Consumes ISO and installer. Produces evidence from boot/install/reboot and releaseartifact checksum, plus pendinghardwarechecklist.
- [x] Boot ISO under OVMF/KVM with fresh disk; install only there; write persistence sentinel.
- [x] Remove ISO and boot installeddisk; verify sentinel, services, QEMUtools, network and shell/recovery. Check no liveoverlay/autologinadminoverride remains.
- [x] Test nested guest VM if KVM nesting available; distinguish virtualgraphics from real laptopresults.
- [x] Run full test suite and independent review. Correct material defects; document remaining physicalchecks without claiming them passed.
- [x] Save ISO locally with checksum, push source/patches/manifests to suppliedrepo. BinaryISO only uploaded when suitable repository release/storage path exists; never commit ISO to Git.
- [ ] Boot ISO under OVMF/KVM with fresh disk; install only there; write persistence sentinel.
- [ ] Remove ISO and boot installeddisk; verify sentinel, services, QEMUtools, network and shell/recovery. Check no liveoverlay/autologinadminoverride remains.
- [ ] Test nested guest VM if KVM nesting available; distinguish virtualgraphics from real laptopresults.
- [ ] Run full test suite and independent review. Correct material defects; document remaining physicalchecks without claiming them passed.
- [ ] Save ISO locally with checksum, push source/patches/manifests to suppliedrepo. BinaryISO only uploaded when suitable repository release/storage path exists; never commit ISO to Git.
docs/superpowers/plans/2026-10-13-graphics-session.md
+23 -19
View File
@@ -50,12 +50,12 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
- `load_profile(path: Path) -> Profile`, `build_qemu_argv(profile: Profile, paths: VMPaths) -> list[str]`, `QMPClient.command(name: str, arguments: dict | None = None) -> dict`.
- `VMPaths` provides disk, log, lock and private QMP/SPICE sockets. Session consumes stable name and socket paths only.
- [x] Write validation/QMP fake-server tests: invalid names/resources, missing ISO, paths containing spaces, greeting/capabilities, asynchronous events, IDs, timeouts.
- [x] Run `python3 -m unittest discover -s tests -v`; observe failure before implementation.
- [x] Implement validation and bounded QMP; subprocess argv arrays, named environment tool overrides for tests.
- [x] Add lifecycle tests: simultaneous start, existing-disk preservation, stale socket, failed boot, shutdown vs force, QMP VM identity.
- [x] Implement locked safe disk creation, detached QEMU, virtio disk/network, SPICE no clipboard/file transfer, optional Intel GL. No display dependency in manager and no PID-only stop.
- [x] Run complete suite, help, scratch QEMU lifecycle probe. Commit `feat: add independent VM manager`.
- [ ] Write validation/QMP fake-server tests: invalid names/resources, missing ISO, paths containing spaces, greeting/capabilities, asynchronous events, IDs, timeouts.
- [ ] Run `python3 -m unittest discover -s tests -v`; observe failure before implementation.
- [ ] Implement validation and bounded QMP; subprocess argv arrays, named environment tool overrides for tests.
- [ ] Add lifecycle tests: simultaneous start, existing-disk preservation, stale socket, failed boot, shutdown vs force, QMP VM identity.
- [ ] Implement locked safe disk creation, detached QEMU, virtio disk/network, SPICE no clipboard/file transfer, optional Intel GL. No display dependency in manager and no PID-only stop.
- [ ] Run complete suite, help, scratch QEMU lifecycle probe. Commit `feat: add independent VM manager`.
### Task 2: Minimal fullscreen session
@@ -63,25 +63,29 @@ unittest; `dev/` holds pinned tools and disposable integration probes.
**Interfaces:** Consumes Task1 Profile/VMPaths. Produces `hydra view NAME [--windowed]`, `hydra session [--nested]`, labwc config only in selected runtime. Viewer title is deterministic from validated name; each VM maps to one workspace. Recovery opens host terminal; session never shuts down a VM.
- [x] Probe actual labwc version/overrideInhibition and remote-viewer kiosk USB behavior on disposable state; record versions.
- [x] Resolve manual USB dialog with a small upstream viewer adaptation if necessary; never auto-redirect USB.
- [x] Write tests for valid escaped XML, exact window/workspace/fullscreen rules, no default keyboard/mouse bindings, overrideInhibition on host keys, missing viewer/display errors.
- [x] Run session tests and observe missing implementation failure.
- [x] Implement session/viewer commands and pinned temporary tool environment; no host installation.
- [x] Run full tests and labwc config parsing. Commit `feat: add fullscreen HydraOS session`.
- [ ] Probe actual labwc version/overrideInhibition and remote-viewer kiosk USB behavior on disposable state; record versions.
- [ ] Resolve manual USB dialog with a small upstream viewer adaptation if necessary; never auto-redirect USB.
- [ ] Write tests for valid escaped XML, exact window/workspace/fullscreen rules, no default keyboard/mouse bindings, overrideInhibition on host keys, missing viewer/display errors.
- [ ] Run session tests and observe missing implementation failure.
- [ ] Implement session/viewer commands and pinned temporary tool environment; no host installation.
- [ ] Run full tests and labwc config parsing. Commit `feat: add fullscreen HydraOS session`.
### 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.
- [x] Run two disposable KVM instances/viewers, never existing lab disks; verify private QMP and independent lifecycle.
- [x] Verify viewer/session exit preserves QEMU and reattach works; switching neither restarts nor reconnects guests.
- [x] Check input only in disposable session, without injecting into current host or user's guests; add regression tests for discovered defects.
- [x] Record held-key, physical brightness and 144Hz checks needing dedicated TTY. Nested tests cannot prove them.
- [x] Run full suite and independent review; fix material findings with reproducing tests.
- [x] Commit evidence and push reviewed prototype branch to supplied repository.
- [ ] 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.
## Next plan
docs/verification.md
-15
View File
@@ -1,20 +1,5 @@
# Graphics and recovery verification
The [2026-10-13 preview record](releases/2026-10-13/acceptance.json) identifies
the exact ISO, firmware and harness used for the passing twelve-check acceptance
run. The unmodified ISO installed onto a fresh 32 GiB qcow2 disk and booted from
that disk with no ISO attached. Password login, persistent data, removed live
overrides, installed services, DNS lookup, routed LAN-to-VM TCP, normal-user
nested KVM/bridge access and normal shutdown passed. An actual logout followed
by 15 seconds without an open login and a new login preserved the same VM,
QMP/SPICE sockets and user audio services. Audio playback was not measured.
Live graphical recovery was also observed through a private QEMU screenshot;
the earlier two-guest compositor and profile-refresh probes cover fullscreen
switching and hidden workload continuity. These virtual checks leave the physical
ASUS checks listed below pending. No current host configuration or existing user
VM disk was changed for this validation.
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
hydra/build.py
-559
View File
@@ -1,559 +0,0 @@
"""Verified, resumable Gentoo builds in a dedicated rootless Podman store."""
import argparse
import contextlib
import fcntl
import hashlib
import json
import lzma
import os
from pathlib import Path
import re
import shutil
import shlex
import subprocess
import sys
import tempfile
import urllib.parse
import urllib.request
class BuildError(RuntimeError):
pass
def package_job_script(command, state='/var/lib/hydra-build', package_db='/var/db/pkg', inputs=''):
"""The contained job owns its log/result even if the outer caller disconnects."""
return f'''import fcntl, json, os, pathlib, subprocess, sys, time
state = pathlib.Path({str(state)!r})
database = pathlib.Path({str(package_db)!r})
state.mkdir(parents=True, exist_ok=True)
lock = (state / 'packages.lock').open('a')
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
sys.exit(75)
def write(name, data):
temporary = state / (name + '.tmp')
temporary.write_text(json.dumps(data, indent=2) + '\\n')
temporary.replace(state / name)
status = {{'running': True, 'pid': os.getpid(), 'inputs': {inputs!r}, 'started': int(time.time())}}
write('packages.status.json', status)
with (state / 'packages.log').open('a') as log:
try:
result = subprocess.run({command!r}, stdout=log, stderr=subprocess.STDOUT)
code = result.returncode
except OSError as error:
log.write(str(error) + '\\n')
code = 127
if code == 0:
inventory = []
for use in sorted(database.glob('*/*/USE')):
package = use.parent
inventory.append({{'package': str(package.relative_to(database)), 'use': use.read_text().split(),
'build_time': (package / 'BUILD_TIME').read_text().strip() if (package / 'BUILD_TIME').exists() else None}})
write('packages.json', inventory)
status.update(running=False, exit_code=code, finished=int(time.time()))
write('packages.status.json', status)
sys.exit(code)
'''
def rootfs_retry_guard_script():
"""Refuse deletion of a still-mounted export, including nested mountpoints."""
return r"""mounts=$(findmnt --raw --noheadings --output TARGET)
while IFS= read -r mountpoint; do
case "$mountpoint" in
"$root"|"$root/"*) echo 'Refusing to remove a mounted rootfs export' >&2; exit 1 ;;
esac
done <<< "$mounts"
if [ -e "$root" ] || [ -L "$root" ]; then
[ ! -L "$root" ] && [ -f "$root/.hydra-export-owned" ] || { echo 'Unowned rootfs output' >&2; exit 1; }
rm -rf -- "$root"
fi
"""
def verify_hash(path, expected):
digest = hashlib.sha256()
with Path(path).open('rb') as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b''):
digest.update(chunk)
if digest.hexdigest() != expected:
raise BuildError(f'SHA256 mismatch: {path}')
def require_signature(status, fingerprint):
valid = [line.split() for line in status.splitlines() if line.startswith('[GNUPG:] VALIDSIG ')]
if not valid or not any(fields[-1].upper() == fingerprint.upper() for fields in valid):
raise BuildError('Required primary signature fingerprint is missing')
if any(token in status for token in ('BADSIG', 'ERRSIG', 'REVKEYSIG', 'EXPKEYSIG')):
raise BuildError('Invalid release signature')
class Builder:
def __init__(self, project, work, cache, podman=None, gpg=None, source_dir=None):
if os.geteuid() == 0:
raise BuildError('Run as the normal user, never host root')
self.project = Path(project).resolve()
self.source_dir = Path(source_dir or project).resolve()
self.work, self.cache = Path(work).resolve(), Path(cache).resolve()
for path in (self.work, self.cache):
if path == self.project or not path.is_relative_to(self.project):
raise BuildError('Work and cache paths must be inside the project, below its root')
if self.work == self.cache or self.work.is_relative_to(self.cache) or self.cache.is_relative_to(self.work):
raise BuildError('Work and cache must be separate directories')
self.podman = podman or shutil.which('podman') or 'podman'
self.gpg = gpg or shutil.which('gpg') or 'gpg'
runtime = Path(os.environ.get('XDG_RUNTIME_DIR', f'/run/user/{os.getuid()}'))
identifier = hashlib.sha256(str(self.work).encode()).hexdigest()[:12]
self.runroot = runtime / ('hydra-build-podman' if self.work == self.project / 'build' else f'hydra-build-{identifier}')
self.name = 'hydra-builder-' + identifier
self.image = 'localhost/hydra-stage3:20261011'
self._stage3_layers = {}
def podman_argv(self, *args):
return [self.podman, '--root', str(self.work / 'podman'), '--runroot', str(self.runroot), '--storage-driver', 'overlay', *args]
def container_run_args(self):
return ['run', '--detach', '--name', self.name, '--hostname', 'hydra-builder',
'--network', 'host', '--cap-add', 'SYS_ADMIN', '--security-opt', 'seccomp=unconfined',
'--volume', f'{self.source_dir}:/hydra-source:ro', '--volume', f'{self.cache}:/hydra-cache:ro',
self.image, '/bin/sleep', 'infinity']
def run(self, argv, capture=False, check=True):
print('+ ' + ' '.join(str(arg) for arg in argv), file=sys.stderr, flush=True)
result = subprocess.run(argv, text=True, stdout=subprocess.PIPE if capture else None,
stderr=subprocess.PIPE if capture else None)
if check and result.returncode:
raise BuildError(f'Command failed ({result.returncode}): {argv[0]}\n{result.stderr or ""}')
return result
def pod(self, *args, **kwargs):
return self.run(self.podman_argv(*args), **kwargs)
def exec(self, command, capture=False):
return self.pod('exec', self.name, '/bin/bash', '-euc', command, capture=capture)
def inspect(self, kind, name):
result = self.pod(kind, 'inspect', name, capture=True, check=False)
if result.returncode:
raise BuildError(f'{kind} provenance is unavailable; use a fresh work directory')
entries = json.loads(result.stdout)
if not isinstance(entries, list) or len(entries) != 1 or not isinstance(entries[0], dict):
raise BuildError(f'Unexpected {kind} provenance')
return entries[0]
def check_mounts(self, container):
for destination, path, label in (('/hydra-source', self.source_dir, 'source'),
('/hydra-cache', self.cache, 'cache')):
mounts = [item for item in container.get('Mounts', []) if item.get('Destination') == destination]
if len(mounts) != 1 or mounts[0].get('Source') != str(path) or mounts[0].get('RW') is not False:
raise BuildError(f'Existing builder {label} mount differs; use its original checkout/cache or a fresh work directory')
def stage3_layer(self, inputs):
stage = inputs['stage3']
archive = self.cache / Path(urllib.parse.urlparse(stage['url']).path).name
verify_hash(archive, stage['sha256'])
if stage['sha256'] not in self._stage3_layers:
digest = hashlib.sha256()
with lzma.open(archive, 'rb') as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b''):
digest.update(chunk)
self._stage3_layers[stage['sha256']] = 'sha256:' + digest.hexdigest()
return self._stage3_layers[stage['sha256']]
def check_base(self, image, inputs):
expected = self.stage3_layer(inputs)
if image.get('RootFS', {}).get('Layers') != [expected]:
raise BuildError('Existing builder stage3 differs from pinned archive; use a fresh work directory')
return expected
def read_container_file(self, path):
# Podman cp also reads a stopped container, so provenance precedes start.
with tempfile.TemporaryDirectory(prefix='hydra-read-') as directory:
target = Path(directory) / 'data'
result = self.pod('cp', f'{self.name}:{path}', str(target), capture=True, check=False)
return target.read_text() if result.returncode == 0 else None
def check_snapshot(self, inputs, allow_pending=False):
marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot')
if marker is None and allow_pending:
marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot-input')
if marker is None or marker.strip() != inputs['portage']['commit']:
raise BuildError('Existing builder snapshot differs or is unverified; use a fresh work directory')
def active_package_job(self):
if not (self.work / 'podman').is_dir():
return False
result = self.pod('top', self.name, 'hpid', 'args', capture=True, check=False)
if result.returncode == 0 and any(re.search(r'(?:^|\s)(?:\S*/)?emerge(?:\s|$)', line)
for line in result.stdout.splitlines()[1:]):
return True
probe = self.pod('exec', self.name, 'python3', '-c', '''import fcntl, pathlib, sys
path = pathlib.Path('/var/lib/hydra-build/packages.lock')
if path.exists():
with path.open('r') as lock:
try:
fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
except BlockingIOError:
sys.exit(75)
''', capture=True, check=False)
return probe.returncode == 75
def manifest(self):
path = self.work / 'manifest.json'
return json.loads(path.read_text()) if path.exists() else {}
def input_id(self):
digest = hashlib.sha256()
for directory in ('image', 'patches'):
for path in sorted((self.source_dir / directory).rglob('*')):
if path.is_file():
digest.update(str(path.relative_to(self.source_dir)).encode())
digest.update(path.read_bytes())
return digest.hexdigest()
def record(self, step, **details):
data = self.manifest()
data[step] = {'complete': True, 'inputs': self.input_id(), **details}
data['builder'] = self.name
path = self.work / 'manifest.json'
temp = path.with_suffix('.tmp')
temp.write_text(json.dumps(data, indent=2) + '\n')
temp.replace(path)
def complete(self, step):
entry = self.manifest().get(step, {})
if not entry.get('complete') or entry.get('inputs') != self.input_id():
return False
if step in ('rootfs', 'iso'):
from hydra import image
if entry.get('source_inputs') != image.source_identity(self.source_dir):
return False
if not image.artifacts_valid(self.work / 'artifacts', entry):
return False
return self.complete('packages' if step == 'rootfs' else 'rootfs')
if step == 'packages':
try:
self.package_inventory()
except (BuildError, OSError):
return False
return self.complete('prepare')
if step == 'prepare':
try:
container = self.inspect('container', self.name)
self.check_mounts(container)
inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
self.check_base(self.inspect('image', container['Image']), inputs)
self.check_snapshot(inputs)
return True
except (BuildError, OSError, ValueError, KeyError, lzma.LZMAError):
return False
if step == 'fetch':
inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
try:
for item in [inputs['stage3'], inputs['portage'], *inputs.get('local_packages', [])]:
path = self.cache / item.get('filename', Path(urllib.parse.urlparse(item['url']).path).name)
verify_hash(path, item['sha256'])
except (OSError, BuildError):
return False
return True
return False
def package_inventory(self):
from hydra import image
checksum = self.manifest().get('packages', {}).get('inventory_sha256')
if not isinstance(checksum, str) or not re.fullmatch('[a-f0-9]{64}', checksum):
raise BuildError('Package inventory has no recorded checksum')
try:
return image.read_package_inventory(self.work / 'artifacts/packages.json', checksum)
except (image.ImageError, OSError) as error:
raise BuildError(f'Package inventory validation failed: {error}') from error
def download(self, url, filename=None):
parsed = urllib.parse.urlparse(url)
if parsed.scheme != 'https':
raise BuildError('Build input URLs must use HTTPS')
target = self.cache / (filename or Path(parsed.path).name)
if not target.exists():
temp = target.with_suffix(target.suffix + '.partial')
print(f'Downloading {url}', flush=True)
with urllib.request.urlopen(url, timeout=120) as response, temp.open('wb') as stream:
shutil.copyfileobj(response, stream)
temp.replace(target)
return target
def fetch(self):
inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
stage = inputs['stage3']
archive = self.download(stage['url'])
verify_hash(archive, stage['sha256'])
snapshot = self.download(inputs['portage']['url'], inputs['portage']['filename'])
verify_hash(snapshot, inputs['portage']['sha256'])
for item in inputs.get('local_packages', []):
verify_hash(self.download(item['url'], item['filename']), item['sha256'])
signature = self.download(stage['signature_url'])
keyring = self.download(inputs['keyring_url'])
verify_hash(keyring, inputs['keyring_sha256'])
with tempfile.TemporaryDirectory(prefix='hydra-gpg-', dir=self.work) as home:
result = self.run([self.gpg, '--batch', '--homedir', home, '--no-default-keyring', '--keyring', str(keyring),
'--status-fd', '1', '--verify', str(signature), str(archive)], capture=True)
require_signature(result.stdout, stage['fingerprint'])
(self.work / 'artifacts/stage3-signature.txt').write_text(result.stdout + result.stderr)
self.record('fetch', sources=inputs)
return inputs
def prepare(self):
existing = self.pod('container', 'exists', self.name, capture=True, check=False).returncode == 0
container = self.inspect('container', self.name) if existing else None
if container is not None:
self.check_mounts(container) # Reject another worktree before changing any checkpoint.
if self.active_package_job():
raise BuildError('A contained package job is running; wait for it to finish before preparation')
inputs = self.fetch() # Reverify cached bytes even when resuming.
if container is not None:
layer = self.check_base(self.inspect('image', container['Image']), inputs)
self.check_snapshot(inputs, allow_pending=True)
elif self.pod('image', 'exists', self.image, capture=True, check=False).returncode:
self.pod('import', str(self.cache / Path(urllib.parse.urlparse(inputs['stage3']['url']).path).name), self.image)
layer = self.check_base(self.inspect('image', self.image), inputs)
else:
layer = self.check_base(self.inspect('image', self.image), inputs)
if container is None:
self.pod(*self.container_run_args())
else:
self.pod('start', self.name)
snapshot = inputs['portage']['filename']
# Snapshot extraction happens only inside the container; host paths are read-only.
self.exec(f'''mkdir -p /var/db/repos/gentoo
if [ ! -f /var/db/repos/gentoo/.hydra-snapshot ]; then
printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot-input
tar --extract --gzip --file /hydra-cache/{snapshot} --strip-components=1 --directory /var/db/repos/gentoo
printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot
fi
cp -a /hydra-source/image/portage/. /etc/portage/
mkdir -p /var/db/repos/hydra /var/cache/distfiles
cp -a /hydra-source/image/overlay/. /var/db/repos/hydra/
cp /hydra-cache/brightnessctl-0.5.1.tar.gz /var/cache/distfiles/brightnessctl-0.5.1.tar.gz
mkdir -p /etc/portage/patches/app-emulation/qemu /etc/portage/patches/app-emulation/virt-viewer
cp /hydra-source/patches/qemu-virtio-refresh-rate.patch /etc/portage/patches/app-emulation/qemu/10-hydra-refresh.patch
cp /hydra-source/patches/virt-viewer-hydra-session.patch /etc/portage/patches/app-emulation/virt-viewer/10-hydra-session.patch
if [ -d /etc/portage/gnupg ] && [ ! -s /etc/portage/gnupg/mykeyid ]; then
mv /etc/portage/gnupg /etc/portage/gnupg.hydra-untrusted
fi
getuto
gpg --batch --homedir /etc/portage/gnupg --import /hydra-cache/service-keys.gpg
chmod a+r /etc/portage/gnupg/pubring.kbx /etc/portage/gnupg/trustdb.gpg
''')
self.record('prepare', image=container['Image'] if container else self.image,
stage3_sha256=inputs['stage3']['sha256'], stage3_layer=layer,
source_dir=str(self.source_dir), cache_dir=str(self.cache), snapshot=inputs['portage']['commit'])
def packages(self, pretend=False):
self.prepare()
atoms = [line.strip() for line in (self.source_dir / 'image/packages').read_text().splitlines()
if line.strip() and not line.startswith('#')]
import shlex
command = ['emerge', '--verbose', '--update', '--newuse', '--deep', '--backtrack=100',
'--usepkg-exclude=app-emulation/qemu app-emulation/virt-viewer']
if pretend:
self.exec(shlex.join([*command, '--pretend', *atoms]))
return
inputs = self.input_id()
self.record('packages', complete=False, job_log='/var/lib/hydra-build/packages.log')
patches = hashlib.sha256()
for name in ('qemu-virtio-refresh-rate.patch', 'virt-viewer-hydra-session.patch'):
patches.update((self.source_dir / 'patches' / name).read_bytes())
patch_id = patches.hexdigest()
forced = ['emerge', '--verbose', '--oneshot', '--usepkg=n', '--getbinpkg=n',
'app-emulation/qemu', 'app-emulation/virt-viewer']
package_commands = shlex.join([*command, *atoms]) + '\n' + f'''if [ "$(cat /var/lib/hydra-build/patched-source.sha256 2>/dev/null || true)" != {shlex.quote(patch_id)} ]; then
{shlex.join(forced)}
printf '%s\\n' {shlex.quote(patch_id)} > /var/lib/hydra-build/patched-source.sha256
fi'''
script = package_job_script(['/bin/bash', '-euc', package_commands], inputs=inputs)
self.exec(shlex.join(['python3', '-c', script]))
if self.input_id() != inputs:
raise BuildError('Image inputs changed during package installation; completion was not recorded')
inventory = self.exec('cat /var/lib/hydra-build/packages.json', capture=True).stdout
from hydra import image
try:
image.validate_package_inventory(json.loads(inventory))
except (ValueError, image.ImageError) as error:
raise BuildError(f'Package inventory is malformed: {error}') from error
target = self.work / 'artifacts/packages.json'
temporary = target.with_suffix('.tmp')
temporary.write_text(inventory)
temporary.replace(target)
self.record('packages', inventory='artifacts/packages.json', patched_source=patch_id,
inventory_sha256=image.digest(target))
def assembly(self, snapshot_digest=None):
"""A separate rootless container processes an exported copy, never the builder."""
if snapshot_digest is None:
state = self.manifest().get('rootfs', {})
snapshot_digest = state.get('snapshot_sha256')
if not isinstance(snapshot_digest, str) or not re.fullmatch('[a-f0-9]{64}', snapshot_digest):
raise BuildError('Verified exported snapshot is required')
name = self.name + '-image-' + snapshot_digest[:12]
image = 'localhost/hydra-image:' + snapshot_digest
exists = self.pod('container', 'exists', name, capture=True, check=False).returncode == 0
if not exists:
self.pod('import', str(self.work / 'artifacts/builder-rootfs.tar'), image)
self.pod('run', '--detach', '--name', name, '--hostname', 'hydra-image-assembler',
'--network', 'none', '--cap-add', 'SYS_ADMIN', '--security-opt', 'seccomp=unconfined',
'--volume', f'{self.source_dir}:/hydra-source:ro',
'--volume', f'{self.work}:/hydra-output:rw', image, '/bin/sleep', 'infinity')
container = self.inspect('container', name)
for destination, source, writable in (('/hydra-source', self.source_dir, False),
('/hydra-output', self.work, True)):
mounts = [mount for mount in container.get('Mounts', []) if mount.get('Destination') == destination]
if len(mounts) != 1 or mounts[0].get('Source') != str(source) or mounts[0].get('RW') is not writable:
raise BuildError('Assembly mount provenance differs; refusing to write')
provenance = self.inspect('image', container['Image'])
if provenance.get('RootFS', {}).get('Layers') != ['sha256:' + snapshot_digest]:
raise BuildError('Assembly image does not match the exported builder snapshot')
if exists:
self.pod('start', name)
return name
def rootfs(self):
from hydra import image
if self.active_package_job():
raise BuildError('A contained package job is running; wait before export')
if not self.complete('packages'):
raise BuildError('Verified completed packages are required before rootfs export')
if self.complete('rootfs'):
return
source_inputs = image.source_identity(self.source_dir)
self.record('rootfs', complete=False)
original = self.work / 'artifacts/builder-rootfs.tar'
if original.is_symlink() or not original.parent.resolve().is_relative_to(self.work):
raise BuildError('Raw export output must remain inside the dedicated work directory')
original.touch(mode=0o600, exist_ok=True)
original.chmod(0o600) # The unsanitized intermediate can contain builder-private trust.
self.pod('export', '--output', str(original), self.name)
snapshot = image.digest(original)
assembler = self.assembly(snapshot)
commit = self.run(['git', '-C', str(self.source_dir), 'rev-parse', 'HEAD'], capture=True).stdout.strip()
script = 'root=/hydra-output/rootfs\n' + rootfs_retry_guard_script() + r"""mkdir -p "$root"
touch "$root/.hydra-export-owned"
tar --extract --preserve-permissions --numeric-owner --file /hydra-output/artifacts/builder-rootfs.tar --directory "$root"
/hydra-source/image/configure-rootfs.sh "$root"
cp "$(python3 /hydra-source/hydra/image.py kernel-image "$root")" /hydra-output/artifacts/vmlinuz
cp "$root/boot/initramfs-live.img" /hydra-output/artifacts/initramfs-live.img
cp "$root/usr/share/hydra/build-manifest.json" /hydra-output/artifacts/rootfs-manifest.json
# Keep ownership for safe retries outside the shipped payload.
mksquashfs "$root" /hydra-output/artifacts/squashfs.img -noappend -comp zstd -processors 4 -e boot/initramfs-live.img .hydra-export-owned
chmod 0644 /hydra-output/artifacts/vmlinuz /hydra-output/artifacts/initramfs-live.img /hydra-output/artifacts/squashfs.img /hydra-output/artifacts/rootfs-manifest.json
"""
self.pod('exec', '--env', f'HYDRA_SOURCE_COMMIT={commit}', assembler, '/bin/bash', '-euc', script)
if image.source_identity(self.source_dir) != source_inputs:
raise BuildError('Source inputs changed during rootfs assembly; completion not recorded')
manifest = json.loads((self.work / 'artifacts/rootfs-manifest.json').read_text())
if manifest.get('packages') != self.package_inventory():
raise BuildError('Payload package inventory differs from the validated package checkpoint')
if manifest.get('source_files') != image.source_files(self.source_dir):
raise BuildError('Payload source manifest is incomplete or stale')
artifacts = {name: image.digest(self.work / 'artifacts' / name) for name in
('vmlinuz', 'initramfs-live.img', 'squashfs.img', 'rootfs-manifest.json', 'packages.json')}
self.record('rootfs', source_inputs=source_inputs, snapshot_sha256=snapshot,
assembler=assembler, kernel=manifest['kernel'], artifacts=artifacts)
def iso(self):
from hydra import image
if not self.complete('rootfs'):
raise BuildError('Verified rootfs artifacts are required before ISO assembly')
if self.complete('iso'):
return
source_inputs = image.source_identity(self.source_dir)
assembler = self.assembly()
self.record('iso', complete=False)
script = r"""tree=/hydra-output/iso-tree
[ ! -L "$tree" ] || { echo 'ISO output cannot be a symlink' >&2; exit 1; }
mkdir -p "$tree/boot/grub" "$tree/LiveOS"
cp /hydra-output/artifacts/vmlinuz "$tree/boot/vmlinuz"
cp /hydra-output/artifacts/initramfs-live.img "$tree/boot/initramfs-live.img"
cp /hydra-output/artifacts/squashfs.img "$tree/LiveOS/squashfs.img"
cp /hydra-source/image/grub.cfg "$tree/boot/grub/grub.cfg"
cp /hydra-output/artifacts/rootfs-manifest.json "$tree/build-manifest.json"
cp /hydra-output/artifacts/packages.json "$tree/packages.json"
grub-mkrescue --directory=/usr/lib/grub/x86_64-efi --output=/hydra-output/artifacts/hydra-os.iso "$tree" -- -volid HYDRA_OS
xorriso -indev /hydra-output/artifacts/hydra-os.iso -report_el_torito plain -pvd_info > /hydra-output/artifacts/iso-inspection.txt 2>&1
chmod 0644 /hydra-output/artifacts/hydra-os.iso /hydra-output/artifacts/iso-inspection.txt
"""
self.pod('exec', assembler, '/bin/bash', '-euc', script)
inspection = (self.work / 'artifacts/iso-inspection.txt').read_text()
if 'HYDRA_OS' not in inspection or not re.search(r'El Torito boot img.*UEFI', inspection):
raise BuildError('ISO does not report the expected volume label and UEFI boot image')
if image.source_identity(self.source_dir) != source_inputs:
raise BuildError('Source inputs changed during ISO assembly; completion not recorded')
checksum = image.digest(self.work / 'artifacts/hydra-os.iso')
(self.work / 'artifacts/hydra-os.iso.sha256').write_text(checksum + ' hydra-os.iso\n')
artifacts = {name: image.digest(self.work / 'artifacts' / name) for name in
('hydra-os.iso', 'hydra-os.iso.sha256', 'iso-inspection.txt', 'packages.json')}
self.record('iso', source_inputs=source_inputs, label=image.LABEL, artifacts=artifacts)
def status(self):
data = self.manifest()
for step in ('fetch', 'prepare', 'packages', 'rootfs', 'iso'):
if step in data:
data[step]['recorded_complete'] = data[step].get('complete', False)
data[step]['complete'] = self.complete(step)
data['mutation_lock_held'] = self.build_running()
data['package_job_running'] = self.active_package_job()
data['running'] = data['mutation_lock_held'] or data['package_job_running']
print(json.dumps(data, indent=2))
def build_running(self):
try:
with (self.work / '.hydra-build.lock').open('r') as lock:
try:
fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
except BlockingIOError:
return True
except FileNotFoundError:
pass
return False
@contextlib.contextmanager
def locked(self):
self.work.mkdir(parents=True, exist_ok=True)
self.cache.mkdir(parents=True, exist_ok=True)
(self.work / 'artifacts').mkdir(exist_ok=True)
with (self.work / '.hydra-build.lock').open('a') as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError as exc:
raise BuildError('Another Hydra build is running in this work directory') from exc
yield
def main(argv=None):
source = Path(__file__).resolve().parents[1]
# Linked worktrees share project-owned caches/build state in the primary checkout.
common = subprocess.run(['git', '-C', str(source), 'rev-parse', '--path-format=absolute', '--git-common-dir'],
text=True, capture_output=True)
project = Path(common.stdout.strip()).parent if common.returncode == 0 else source
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('step', choices=('fetch', 'prepare', 'packages', 'rootfs', 'iso', 'status'))
parser.add_argument('--work', type=Path, default=project / 'build')
parser.add_argument('--cache', type=Path, default=project / 'cache')
parser.add_argument('--podman')
parser.add_argument('--gpg')
parser.add_argument('--pretend', action='store_true', help='resolve packages without installing')
options = parser.parse_args(argv)
try:
builder = Builder(project, options.work, options.cache, options.podman, options.gpg, source)
if options.step == 'status':
builder.status()
return 0
with builder.locked():
if options.step == 'packages':
builder.packages(options.pretend)
else:
getattr(builder, options.step)()
return 0
except (BuildError, OSError, ValueError, lzma.LZMAError) as exc:
print(f'hydra-build: {exc}', file=sys.stderr)
return 1
hydra/image.py
-404
View File
@@ -1,404 +0,0 @@
"""Prepare only exported Gentoo payloads and configure the dedicated guest network."""
import argparse
import bz2
import io
import runpy
import hashlib
import ipaddress
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
LABEL = 'HYDRA_OS'
BRIDGE = 'hydrabr0'
SUBNET = ipaddress.ip_network('10.203.0.0/24')
CONNECTION_UUID = 'c84844bd-2955-4a61-9463-5e540140f9a2'
class ImageError(RuntimeError):
pass
def root_path(root, relative, follow_leaf=True):
root = Path(root).resolve()
if root == Path('/') or not root.is_dir():
raise ImageError('Expected an isolated exported rootfs directory')
relative = Path(relative)
if relative.is_absolute() or '..' in relative.parts:
raise ImageError('Payload path must be relative and remain inside the export')
path = root / relative
checked = path if follow_leaf else path.parent
if not checked.resolve().is_relative_to(root):
raise ImageError(f'Payload symlink escapes the export: {relative}')
return path
def remove(root, relative):
path = root_path(root, relative, follow_leaf=False)
if path.is_symlink() or path.is_file():
path.unlink()
elif path.is_dir():
shutil.rmtree(path)
def write(root, relative, content, mode=0o644):
path = root_path(root, relative)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
path.chmod(mode)
return path
def sanitize_saved_environments(root):
"""Use Portage's parser to scrub exported declarations without evaluating them."""
database = root_path(root, 'var/db/pkg')
paths = sorted(database.glob('*/*/environment.bz2'))
result = {'files_scanned': len(paths), 'filtered_files': 0, 'variable_names': []}
if not paths:
return result
scripts = sorted(root_path(root, 'usr/lib/portage').glob('*/filter-bash-environment.py'))
if not scripts:
raise ImageError('Installed Portage environment filter is required for export')
script = root_path(root, scripts[0].relative_to(root))
api = runpy.run_path(str(script))
pattern = re.compile(rb'(?i:(?:HTTP|HTTPS|FTP|ALL|NO)_PROXY|.*(?:PASSWORD|PASSWD|SECRET|TOKEN|API_KEY|ACCESS_KEY|CREDENTIAL|AUTHORIZATION).*)')
declarations = re.compile(rb'^(?:declare|export|readonly)\s+(?:-[^ ]+\s+)*([A-Za-z_][A-Za-z0-9_]*)=', re.M)
names = set()
for path in paths:
path = root_path(root, path.relative_to(root))
try:
original = bz2.decompress(path.read_bytes())
except (OSError, ValueError) as error:
raise ImageError('Cannot read an exported package environment') from error
filtered = io.BytesIO()
api['filter_bash_environment'](pattern, io.BytesIO(original), filtered)
clean = filtered.getvalue()
if clean != original:
names.update(name.decode('ascii') for name in declarations.findall(original)
if pattern.fullmatch(name))
path.write_bytes(bz2.compress(clean))
result['filtered_files'] += 1
result['variable_names'] = sorted(names)
return result
def sanitize_rootfs(root):
"""Erase builder identity/secrets in the exported copy, retaining Gentoo updates."""
for relative in ('etc/portage/gnupg', 'etc/portage/gnupg.hydra-untrusted',
'var/lib/hydra-build', 'root', 'home', 'var/cache', 'var/tmp',
'tmp', 'run', 'var/log', 'var/lib/systemd/random-seed',
'var/lib/NetworkManager', 'etc/NetworkManager/system-connections',
'hydra-source', 'hydra-cache', 'hydra-output'):
remove(root, relative)
for name in root_path(root, 'etc/ssh').glob('ssh_host_*'):
remove(root, str(name.relative_to(root)))
for relative, mode in [('root', 0o700), ('home', 0o755), ('tmp', 0o1777),
('var/tmp', 0o1777), ('run', 0o755), ('var/log', 0o755),
('var/cache', 0o755), ('dev', 0o755), ('proc', 0o755), ('sys', 0o755)]:
path = root_path(root, relative)
path.mkdir(parents=True, exist_ok=True)
path.chmod(mode)
remove(root, 'etc/machine-id')
write(root, 'etc/machine-id', '')
remove(root, 'var/lib/dbus/machine-id')
path = root_path(root, 'var/lib/dbus/machine-id', follow_leaf=False)
path.parent.mkdir(parents=True, exist_ok=True)
path.symlink_to('/etc/machine-id')
write(root, 'etc/hostname', 'hydra-os\n')
write(root, 'etc/hosts', '127.0.0.1 localhost\n::1 localhost\n127.0.1.1 hydra-os\n')
remove(root, 'etc/resolv.conf')
root_path(root, 'etc/resolv.conf', follow_leaf=False).symlink_to('/run/NetworkManager/resolv.conf')
audit = sanitize_saved_environments(root)
write(root, 'usr/share/hydra/environment-sanitization.json', json.dumps(audit, indent=2) + '\n')
def copy_overlay(root, overlay):
for source in sorted(Path(overlay).rglob('*')):
destination = root_path(root, source.relative_to(overlay))
if source.is_symlink():
raise ImageError('Source overlay must contain ordinary directories/files')
if source.is_dir():
destination.mkdir(parents=True, exist_ok=True)
elif source.is_file():
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source, destination)
destination.chmod(source.stat().st_mode & 0o777)
def _kernel_version(root):
modules = root_path(root, 'lib/modules')
versions = sorted(path.name for path in modules.iterdir() if path.is_dir()) if modules.is_dir() else []
if len(versions) != 1 or not re.fullmatch(r'[A-Za-z0-9_.+-]+', versions[0]):
raise ImageError('Exactly one installed payload kernel is required')
return versions[0]
def kernel_image(root):
version = _kernel_version(root)
candidates = [root_path(root, 'boot/' + prefix + version)
for prefix in ('kernel-', 'vmlinuz-')]
images = [path for path in candidates if path.is_file()]
if len(images) == 2 and images[0].samefile(images[1]):
return images[1] # Stable installer/GRUB vmlinuz alias for Gentoo kernel-bin.
if len(images) != 1:
raise ImageError('Installed modules require exactly one matching boot kernel image')
return images[0]
def kernel_version(root):
kernel_image(root)
return _kernel_version(root)
def live_kernel_args():
return [f'root=live:CDLABEL={LABEL}', 'rd.live.image', 'rd.live.overlay.overlayfs=1',
'ro', 'console=tty0', 'console=ttyS0,115200n8']
def source_files(source):
source = Path(source)
result = {}
for directory in ('image', 'hydra', 'bin', 'session', 'patches'):
for path in sorted((source / directory).rglob('*')):
if path.is_file() and '__pycache__' not in path.parts and path.suffix != '.pyc':
result[str(path.relative_to(source))] = digest(path)
return result
def source_identity(source):
return hashlib.sha256(json.dumps(source_files(source), sort_keys=True).encode()).hexdigest()
def payload_manifest(root, source):
packages = []
database = root_path(root, 'var/db/pkg')
for use in sorted(database.glob('*/*/USE')):
package = use.parent
built = package / 'BUILD_TIME'
packages.append({'package': str(package.relative_to(database)), 'use': use.read_text().split(),
'build_time': built.read_text().strip() if built.is_file() else None})
commit = os.environ.get('HYDRA_SOURCE_COMMIT')
if not commit:
git = subprocess.run(['git', '-C', str(source), 'rev-parse', 'HEAD'], text=True, capture_output=True)
commit = git.stdout.strip() if git.returncode == 0 else None
return {'format': 1, 'label': LABEL, 'kernel': kernel_version(root),
'live_kernel_args': live_kernel_args(), 'source_commit': commit,
'source_files': source_files(source), 'sources': json.loads((Path(source) / 'image/sources.json').read_text()),
'packages': packages}
def artifacts_valid(directory, state):
artifacts = state.get('artifacts')
if not isinstance(artifacts, dict) or not artifacts:
return False
for name, expected in artifacts.items():
if not isinstance(name, str) or Path(name).name != name or not isinstance(expected, str):
return False
path = Path(directory) / name
try:
if path.is_symlink() or not path.is_file() or digest(path) != expected:
return False
except OSError:
return False
return True
def validate_package_inventory(data):
if not isinstance(data, list) or not data:
raise ImageError('Package inventory must be a nonempty array')
seen = set()
for record in data:
if not isinstance(record, dict) or set(record) != {'package', 'use', 'build_time'}:
raise ImageError('Malformed package inventory entry')
package = record['package']
if (not isinstance(package, str) or
not re.fullmatch(r'[A-Za-z0-9+_.-]+/[A-Za-z0-9+_.-]+-[0-9][A-Za-z0-9+_.-]*', package) or
package in seen):
raise ImageError('Invalid or duplicate package inventory identity')
seen.add(package)
flags = record['use']
if (not isinstance(flags, list) or any(not isinstance(flag, str) or
not re.fullmatch(r'[A-Za-z0-9+_.@-]+', flag) for flag in flags) or len(set(flags)) != len(flags)):
raise ImageError('Invalid package inventory USE flags')
built = record['build_time']
if built is not None and (not isinstance(built, str) or not re.fullmatch(r'[0-9]+', built)):
raise ImageError('Invalid package inventory build time')
return data
def read_package_inventory(path, expected_digest=None):
path = Path(path)
if path.is_symlink() or not path.is_file():
raise ImageError('Package inventory must be an ordinary file')
if expected_digest is not None and digest(path) != expected_digest:
raise ImageError('Package inventory checksum differs from checkpoint')
try:
return validate_package_inventory(json.loads(path.read_text()))
except (ValueError, UnicodeError) as error:
raise ImageError('Package inventory is not valid JSON') from error
def enable_user_audio(root):
root_path(root, 'etc/systemd/user')
result = subprocess.run(['systemctl', '--root=' + str(root), '--global', 'enable',
'pipewire.socket', 'pipewire-pulse.socket', 'wireplumber.service'],
text=True, capture_output=True)
if result.returncode:
raise ImageError('Cannot enable user audio sockets: ' + result.stderr.strip())
def configure_rootfs(root, source):
root, source = Path(root), Path(source)
# Refuse incomplete integrations before changing the exported copy.
for name in ('bin/hydra', 'bin/hydra-install', 'hydra/session.py', 'hydra/installer.py'):
if not (source / name).is_file():
raise ImageError(f'Required committed runtime/installer asset is missing: {name}')
version = kernel_version(root)
kernel = kernel_image(root)
alias = root_path(root, 'boot/vmlinuz-' + version)
if not alias.exists():
alias.symlink_to(kernel.name)
sanitize_rootfs(root)
copy_overlay(root, source / 'image/rootfs')
write(root, 'etc/hydra/live-mode/dracut.conf', (source / 'image/dracut.conf').read_text())
root_path(root, 'etc/hydra/live-mode/dracut.conf.d').mkdir()
package = root_path(root, 'usr/lib/hydra/hydra')
package.mkdir(parents=True, exist_ok=True)
for module in sorted((source / 'hydra').glob('*.py')):
shutil.copyfile(module, root_path(root, 'usr/lib/hydra/hydra/' + module.name))
# Session helpers derive this checkout-compatible launcher from __file__.
for name in ('hydra', 'hydra-install'):
write(root, 'usr/lib/hydra/bin/' + name, (source / 'bin' / name).read_text(), 0o755)
for name, module in (('hydra', 'cli'), ('hydra-install', 'installer')):
write(root, 'usr/bin/' + name,
'#!/usr/bin/python3\nimport sys\nsys.path.insert(0, "/usr/lib/hydra")\n'
f'from hydra.{module} import main\nraise SystemExit(main())\n', 0o755)
write(root, 'usr/bin/hydra-session', '#!/bin/sh\nexec /usr/bin/hydra session "$@"\n', 0o755)
write(root, 'usr/bin/hydra-network', '#!/usr/bin/python3\nimport sys\nsys.path.insert(0, "/usr/lib/hydra")\n'
'from hydra.image import network_main\nraise SystemExit(network_main())\n', 0o755)
write(root, 'usr/share/hydra/build-manifest.json', json.dumps(payload_manifest(root, source), indent=2) + '\n')
make = root_path(root, 'etc/portage/make.conf')
if make.is_file():
settings = make.read_text()
settings = re.sub(r'(?<=[\s"\'])-(?:usersandbox|sandbox|network-sandbox|ipc-sandbox|pid-sandbox)(?=[\s"\']|$)', '', settings)
make.write_text(settings)
for path in root_path(root, 'etc/sudoers.d').iterdir():
path.chmod(0o440)
return version
def check_network_conflicts(routes, addresses):
try:
if not isinstance(routes, list) or not isinstance(addresses, list):
raise ValueError('Expected JSON arrays')
for route in routes:
dst = route.get('dst')
if dst in ('default', None):
continue
network = ipaddress.ip_network(dst, strict=False)
if network.version != 4:
raise ValueError('Unexpected non-IPv4 route')
if network.overlaps(SUBNET) and not (route.get('dev') == BRIDGE and network.subnet_of(SUBNET)):
raise ImageError(f'Guest subnet {SUBNET} conflicts with route {dst} on {route.get("dev")}')
for interface in addresses:
for addr in interface['addr_info']:
if addr.get('family') != 'inet':
continue
network = ipaddress.ip_network(f'{addr["local"]}/{addr["prefixlen"]}', strict=False)
owned = interface['ifname'] == BRIDGE and addr['local'] == '10.203.0.1' and addr['prefixlen'] == 24
if network.overlaps(SUBNET) and not owned:
raise ImageError(f'Guest subnet {SUBNET} conflicts with address on {interface["ifname"]}')
except (ValueError, KeyError, TypeError, AttributeError) as error:
raise ImageError('Cannot safely interpret IPv4 routes/addresses') from error
def configure_network(run=subprocess.run):
def command(argv, **kwargs):
result = run(argv, text=True, capture_output=True, **kwargs)
if result.returncode:
raise ImageError(f'Network command failed: {argv[0]}: {result.stderr.strip()}')
return result.stdout
routes = json.loads(command(['ip', '-json', '-4', 'route', 'show', 'table', 'all']))
addresses = json.loads(command(['ip', '-json', '-4', 'address', 'show']))
check_network_conflicts(routes, addresses)
probe = run(['nmcli', '--get-values', 'connection.uuid,connection.type,connection.interface-name,ipv4.method,ipv4.addresses',
'connection', 'show', 'uuid', CONNECTION_UUID], text=True, capture_output=True)
exists = probe.returncode == 0
if exists and probe.stdout.strip().splitlines() != [CONNECTION_UUID, 'bridge', BRIDGE, 'manual', '10.203.0.1/24']:
raise ImageError('Hydra bridge connection exists with unexpected configuration')
if not exists and probe.returncode != 10:
raise ImageError('Cannot query NetworkManager connections')
links = json.loads(command(['ip', '-json', '-details', 'link', 'show']))
for link in links:
if link.get('ifname') == BRIDGE and (not exists or link.get('linkinfo', {}).get('info_kind') != 'bridge'):
raise ImageError('Refusing to replace an existing interface named hydrabr0')
if not exists:
command(['nmcli', 'connection', 'add', 'type', 'bridge', 'con-name', 'hydra-guest',
'ifname', BRIDGE, 'connection.uuid', CONNECTION_UUID,
'connection.autoconnect', 'no', 'bridge.stp', 'no',
'ipv4.method', 'manual', 'ipv4.addresses', '10.203.0.1/24',
'ipv4.never-default', 'yes', 'ipv6.method', 'disabled'])
# Activation is always gated by the fresh route/address check, including restarts.
command(['nmcli', 'connection', 'up', 'uuid', CONNECTION_UUID])
command(['sysctl', '-w', 'net.ipv4.ip_forward=1'])
table = run(['nft', 'list', 'table', 'ip', 'hydra_nat'], text=True, capture_output=True)
rules = ('delete table ip hydra_nat\n' if table.returncode == 0 else '') + '''table ip hydra_nat {
chain postrouting {
type nat hook postrouting priority srcnat; policy accept;
ip saddr 10.203.0.0/24 oifname != "hydrabr0" masquerade
}
}
'''
command(['nft', '-f', '-'], input=rules)
def network_main():
try:
configure_network()
except (ImageError, OSError, ValueError) as error:
print(f'hydra-network: {error}', file=sys.stderr)
return 1
return 0
def digest(path):
value = hashlib.sha256()
with Path(path).open('rb') as stream:
for block in iter(lambda: stream.read(1024 * 1024), b''):
value.update(block)
return value.hexdigest()
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('step', choices=('configure', 'kernel', 'kernel-image', 'network', 'audio'))
parser.add_argument('root', nargs='?', type=Path)
parser.add_argument('--source', type=Path)
options = parser.parse_args(argv)
try:
if options.step == 'network':
return network_main()
if options.root is None:
raise ImageError('An isolated rootfs directory is required')
if options.step == 'kernel':
print(kernel_version(options.root))
elif options.step == 'kernel-image':
print(kernel_image(options.root))
elif options.step == 'audio':
enable_user_audio(options.root)
else:
if options.source is None:
raise ImageError('Source checkout is required')
configure_rootfs(options.root, options.source)
except (ImageError, OSError, ValueError) as error:
print(f'hydra-image: {error}', file=sys.stderr)
return 1
return 0
if __name__ == '__main__':
raise SystemExit(main())
hydra/installer.py
-765
View File
@@ -1,765 +0,0 @@
"""Explicit whole-disk installer. Importing this module never mutates storage.
Live payload contract: /etc/hydra/live-mode/active and remove-on-install.json
must be regular files. The manifest is a JSON array of absolute paths matching
LIVE_REMOVABLE. Store all other live-only configuration in live-mode itself;
ordinary installed configuration must already be present in the payload.
"""
from __future__ import annotations
import argparse
from dataclasses import dataclass
import getpass
import fcntl
import json
import os
from pathlib import Path, PurePosixPath
import re
import shlex
import shutil
import stat
import struct
import subprocess
import sys
import tempfile
import warnings
from typing import TextIO
DEFAULT_PAYLOAD = '/run/initramfs/live/LiveOS/squashfs.img'
LIVE_REMOVABLE = (
'/etc/systemd/system/getty@*.service.d/90-hydra-live.conf',
'/etc/systemd/system/serial-getty@*.service.d/90-hydra-live.conf',
'/etc/systemd/system/greetd.service.d/90-hydra-live.conf',
'/etc/sudoers.d/90-hydra-live',
'/etc/greetd/live.toml',
'/etc/NetworkManager/system-connections/90-hydra-live.nmconnection',
)
DEVICE = re.compile(r'/dev/[A-Za-z0-9_.+:/-]+\Z')
KERNEL = re.compile(r'[A-Za-z0-9][A-Za-z0-9_.+-]*\Z')
KNOWN_TYPES = {'disk', 'part', 'loop', 'rom', 'crypt', 'lvm', 'mpath', 'dm',
'raid0', 'raid1', 'raid4', 'raid5', 'raid6', 'raid10', 'md', 'zram'}
class InstallError(RuntimeError):
"""A failed safety check or operation; never suppress this in execution."""
@dataclass
class Request:
disk: str
user: str
payload: str = DEFAULT_PAYLOAD
execute: bool = False
confirm_erase: str | None = None
timezone: str = 'Europe/Moscow'
kernel: str | None = None
class Runner:
"""Checked argv-only commands. Secret-bearing commands redact all output."""
def run(self, argv, *, input=None, sensitive=False):
try:
result = subprocess.run(list(argv), input=input, text=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=False, check=False)
except OSError as exc:
raise InstallError(f'Cannot run {argv[0]}: {exc.strerror}') from exc
if result.returncode:
detail = '' if sensitive else ': ' + result.stderr.strip()[:2000]
raise InstallError(f'Command failed ({result.returncode}): {shlex.join(argv)}{detail}')
return '' if sensitive else result.stdout
class Topology:
def __init__(self):
self.nodes = {}
self.edges = {}
self.parents = {}
self.protected = set()
@classmethod
def from_json(cls, raw):
try:
data = json.loads(raw)
except (ValueError, TypeError) as exc:
raise InstallError('Malformed lsblk JSON') from exc
if not isinstance(data, dict) or not isinstance(data.get('blockdevices'), list) or not data['blockdevices']:
raise InstallError('lsblk must return a nonempty blockdevices array')
result = cls()
parents = []
def visit(node, ancestor=None, trail=()):
if not isinstance(node, dict):
raise InstallError('Malformed lsblk node')
path, kind = node.get('path'), node.get('type')
if not isinstance(path, str) or not DEVICE.fullmatch(path) or '..' in PurePosixPath(path).parts:
raise InstallError('Invalid lsblk device path')
if path in trail or kind not in KNOWN_TYPES:
raise InstallError('Cyclic or unknown lsblk topology')
mounts = node.get('mountpoints')
size, identity = node.get('size'), node.get('maj:min')
if not isinstance(mounts, list) or any(m is not None and not isinstance(m, str) for m in mounts):
raise InstallError('Malformed lsblk mountpoints')
if isinstance(size, bool) or not isinstance(size, int) or size < 0:
raise InstallError('Invalid lsblk size')
empty_optical = (kind == 'rom' and not any(mounts)
and not node.get('children') and node.get('pkname') is None
and ancestor is None)
if size == 0 and not empty_optical:
raise InstallError('Zero-size device is not an unrelated empty optical drive')
if not isinstance(identity, str) or not re.fullmatch(r'\d+:\d+', identity):
raise InstallError('Invalid lsblk device identity')
if path in result.nodes and any(result.nodes[path][key] != node[key] for key in ('type', 'size', 'maj:min', 'mountpoints')):
raise InstallError('Inconsistent duplicate lsblk node')
result.nodes[path] = node
result.edges.setdefault(path, set())
result.parents.setdefault(path, set())
if any(m for m in mounts):
result.protected.add(path)
pkname = node.get('pkname')
if pkname is not None:
if not isinstance(pkname, str) or not pkname:
raise InstallError('Invalid lsblk parent')
parent = pkname if pkname.startswith('/dev/') else '/dev/' + pkname
if not DEVICE.fullmatch(parent):
raise InstallError('Invalid lsblk parent path')
parents.append((path, parent))
if ancestor:
parents.append((path, ancestor))
children = node.get('children', [])
if not isinstance(children, list):
raise InstallError('Malformed lsblk children')
for child in children:
visit(child, path, trail + (path,))
for node in data['blockdevices']:
visit(node)
identities = {}
for path, node in result.nodes.items():
identity = node['maj:min']
if identity in identities and identities[identity] != path:
raise InstallError('Conflicting block-device identities')
identities[identity] = path
directed = {path: set() for path in result.nodes}
for child, parent in parents:
result.link(child, parent)
directed[child].add(parent)
visiting, visited = set(), set()
def check_cycle(path):
if path in visiting:
raise InstallError('Cyclic block-device parent references')
if path in visited:
return
visiting.add(path)
for parent in directed[path]:
check_cycle(parent)
visiting.remove(path)
visited.add(path)
for path in directed:
check_cycle(path)
for path, node in result.nodes.items():
if node['size'] == 0 and result.edges[path]:
raise InstallError('Empty optical drive has block-device dependencies')
# An undirected child edge does not establish backing ancestry.
# Each layer must identify its own parent, even when another
# apparently unrelated disk is the requested installation target.
if node['type'] not in {'disk', 'rom', 'loop', 'zram'} and not directed[path]:
raise InstallError(f"Device has no known backing parent: {path} ({node['type']})")
return result
def link(self, child, parent):
if child == parent or child not in self.nodes or parent not in self.nodes:
raise InstallError('Unresolved or cyclic block-device parent')
# Backing links discovered after lsblk must obey the same directed
# ancestry rule. An undirected shared disk is not itself a cycle.
seen, pending = set(), [parent]
while pending:
current = pending.pop()
if current == child:
raise InstallError('Cyclic block-device backing ancestry')
if current not in seen:
seen.add(current)
pending.extend(self.parents[current] - seen)
self.parents[child].add(parent)
self.edges[child].add(parent)
self.edges[parent].add(child)
def protect(self, path):
if path not in self.nodes:
raise InstallError(f'Cannot resolve mounted/live source: {path}')
if self.nodes[path]['size'] == 0:
raise InstallError(f'Empty optical drive unexpectedly backs mounted/live storage: {path}')
self.protected.add(path)
def component(self, path):
seen, pending = set(), [path]
while pending:
current = pending.pop()
if current not in seen:
seen.add(current)
pending.extend(self.edges[current] - seen)
return seen
class SystemProbe:
"""Read-only Linux topology probes, including loop backing files."""
def __init__(self, runner=None):
self.runner = runner or Runner()
def resolve(self, path):
return os.path.realpath(path)
def physical_disk(self, path):
try:
info = os.stat(path)
if not stat.S_ISBLK(info.st_mode):
return False
device = Path(f'/sys/dev/block/{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}')
if not device.exists() or (device / 'partition').exists():
return False
# Reject loop, device mapper, md, ram and other virtual targets.
return '/virtual/' not in str(device.resolve()) and (device / 'device').exists()
except OSError:
return False
def snapshot(self, payload):
raw = self.runner.run(['lsblk', '--json', '--paths', '--bytes', '--output',
'PATH,TYPE,PKNAME,MOUNTPOINTS,SIZE,MAJ:MIN'])
# lsblk PATH can use /dev/mapper aliases while findmnt uses /dev/dm-N.
try:
decoded = json.loads(raw)
def canonicalize(node):
if not isinstance(node, dict):
raise ValueError()
if not isinstance(node.get('path'), str) or not DEVICE.fullmatch(node['path']) or '..' in PurePosixPath(node['path']).parts:
raise ValueError()
node['path'] = self.resolve(node['path'])
if isinstance(node.get('pkname'), str):
parent = node['pkname']
node['pkname'] = self.resolve(parent if parent.startswith('/dev/') else '/dev/' + parent)
children = node.get('children', [])
if not isinstance(children, list):
raise ValueError()
for child in children:
canonicalize(child)
for node in decoded['blockdevices']:
canonicalize(node)
topo = Topology.from_json(json.dumps(decoded))
except (ValueError, TypeError, KeyError) as exc:
raise InstallError('Malformed lsblk JSON') from exc
mounts = self._mounts(self.runner.run(['findmnt', '--json', '--output',
'SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS']))
for mount in mounts:
source = self._block_source(mount, topo)
if source:
topo.protect(source)
if not Path(payload).is_file():
raise InstallError(f'Prepared squashfs payload is not a regular file: {payload}')
source = self._source_for_path(self.resolve(payload), topo)
if not source:
# tmpfs/overlay payload location cannot establish live-medium ancestry.
raise InstallError('Cannot identify the payload backing block device; use --payload on the mounted live medium')
topo.protect(source)
# Loops can hide further loops, including ancestors of a mounted
# partition or device mapper. Resolve protected components to a fixed
# point, independent of lsblk ordering and payload discovery order.
inspected = set()
while True:
protected_devices = set()
for path in topo.protected:
protected_devices.update(topo.component(path))
pending = [path for path in topo.nodes if path in protected_devices
and topo.nodes[path]['type'] == 'loop' and path not in inspected]
if not pending:
break
for path in pending:
identity = self._loop_backing_identity(path, topo.nodes[path]['maj:min'])
sources = [p for p, n in topo.nodes.items() if n['maj:min'] == identity]
if len(sources) != 1:
raise InstallError(f'Unknown physical backing for mounted loop: {path} ({identity})')
source = sources[0]
topo.link(path, source)
topo.protect(source)
inspected.add(path)
return topo
@staticmethod
def _loop_backing_identity(path, expected_identity):
# LOOP_GET_STATUS64 reports the backing inode's filesystem device,
# independent of names which dracut switch_root may have relocated.
# lo_rdevice is nonzero when the backing inode is itself a block device.
try:
descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC)
try:
info = os.fstat(descriptor)
identity = f'{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}'
if not stat.S_ISBLK(info.st_mode) or identity != expected_identity:
raise InstallError(f'Mounted loop device identity changed: {path}')
status = fcntl.ioctl(descriptor, 0x4C05, bytes(232))
if not isinstance(status, bytes) or len(status) != 232:
raise InstallError(f'Malformed loop status: {path}')
filesystem_device, _, block_device = struct.unpack_from('=QQQ', status)
finally:
os.close(descriptor)
except (OSError, struct.error) as exc:
raise InstallError(f'Cannot inspect mounted loop backing identity: {path}') from exc
device = block_device or filesystem_device
return f'{os.major(device)}:{os.minor(device)}'
@staticmethod
def _mounts(raw):
try:
data = json.loads(raw)
roots = data['filesystems']
if not isinstance(roots, list) or not roots:
raise ValueError()
result = []
def visit(item):
if not isinstance(item, dict) or any(not isinstance(item.get(k), str) for k in ('source', 'target', 'fstype', 'maj:min')):
raise ValueError()
result.append(item)
children = item.get('children', [])
if not isinstance(children, list):
raise ValueError()
for child in children:
visit(child)
for root in roots:
visit(root)
return result
except (ValueError, TypeError, KeyError) as exc:
raise InstallError('Malformed findmnt topology') from exc
def _block_source(self, mount, topo):
if mount['fstype'] in {'zfs', 'btrfs'}:
raise InstallError('Cannot safely resolve all backing disks of mounted ' + mount['fstype'])
source = mount['source'].split('[', 1)[0]
identity = mount['maj:min']
matches = [p for p, n in topo.nodes.items() if n['maj:min'] == identity]
if source.startswith('/dev/'):
path = self.resolve(source)
if path in topo.nodes:
if topo.nodes[path]['maj:min'] != identity:
raise InstallError('Inconsistent mounted block identity')
topo.protect(path)
return path
if not matches:
raise InstallError(f'Cannot resolve mounted/live source: {source}')
if matches:
topo.protect(matches[0])
return matches[0]
if identity.split(':', 1)[0] != '0' and mount['fstype'] not in {'zfs', 'nfs', 'nfs4', 'cifs', '9p', 'virtiofs'}:
raise InstallError(f'Unknown mounted block source: {source}')
return None
def _source_for_path(self, path, topo):
raw = self.runner.run(['findmnt', '--json', '--target', path, '--output',
'SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS'])
mounts = self._mounts(raw)
if len(mounts) != 1:
raise InstallError('Ambiguous backing filesystem')
return self._block_source(mounts[0], topo)
def validate_request(request, probe):
if not isinstance(request.disk, str) or not DEVICE.fullmatch(request.disk) or '..' in PurePosixPath(request.disk).parts:
raise InstallError('Choose an explicit /dev whole-disk device')
if not re.fullmatch(r'[a-z_][a-z0-9_-]{0,31}', request.user) or request.user in {'root', 'hydra-live', 'greetd'}:
raise InstallError('Choose a valid nonreserved lowercase Unix username')
if not re.fullmatch(r'[A-Za-z_+-]+(?:/[A-Za-z0-9_+-]+)+', request.timezone):
raise InstallError('Invalid timezone; use an IANA name such as Europe/Moscow')
if not Path(request.payload).is_absolute() or any(ord(c) < 32 for c in request.payload):
raise InstallError('Payload must be an absolute path without control characters')
if request.kernel is not None and not KERNEL.fullmatch(request.kernel):
raise InstallError('Invalid kernel version')
target = probe.resolve(request.disk)
if not DEVICE.fullmatch(target) or not probe.physical_disk(target):
raise InstallError(f'Target is not a physical whole block disk: {target}')
if request.execute and request.confirm_erase != target:
raise InstallError(f'Execution requires --confirm-erase {target} exactly')
topo = probe.snapshot(request.payload)
if target not in topo.nodes or topo.nodes[target]['type'] != 'disk':
raise InstallError('Target is not a known whole disk')
if topo.nodes[target]['size'] < 4 * 1024 ** 3:
raise InstallError('Target disk is smaller than the minimum 4 GiB')
forbidden = topo.component(target) & topo.protected
if forbidden:
raise InstallError('Target backs mounted/root/live storage: ' + ', '.join(sorted(forbidden)))
return target
def partitions(target):
suffix = 'p' if target[-1].isdigit() else ''
return target + suffix + '1', target + suffix + '2'
def partition_commands(target):
esp, root = partitions(target)
return [
['parted', '--script', target, 'mklabel', 'gpt'],
['parted', '--script', target, 'unit', 'MiB', 'mkpart', 'ESP', 'fat32', '1MiB', '1025MiB'],
['parted', '--script', target, 'set', '1', 'esp', 'on'],
['parted', '--script', target, 'unit', 'MiB', 'mkpart', 'root', 'ext4', '1025MiB', '100%'],
['partprobe', target], ['udevadm', 'settle'],
['mkfs.vfat', '-F', '32', '-n', 'HYDRA_EFI', esp],
['mkfs.ext4', '-F', '-L', 'HYDRA_ROOT', root],
]
def grub_command(root):
return ['chroot', str(root), 'grub-install', '--target=x86_64-efi',
'--efi-directory=/boot/efi', '--bootloader-id=HydraOS', '--removable', '--no-nvram']
def render_plan(request, target, root):
esp, part = partitions(target)
commands = partition_commands(target) + [
['mount', part, str(root)], ['unsquashfs', '-f', '-d', str(root), request.payload],
['mount', esp, str(Path(root) / 'boot/efi')],
['blkid', '-s', 'UUID', '-o', 'value', part],
['blkid', '-s', 'UUID', '-o', 'value', esp],
['mount', '--rbind', '/dev', str(Path(root) / 'dev')],
['mount', '--make-rslave', str(Path(root) / 'dev')],
['mount', '-t', 'proc', 'proc', str(Path(root) / 'proc')],
['mount', '--rbind', '/sys', str(Path(root) / 'sys')],
['mount', '--make-rslave', str(Path(root) / 'sys')],
['chroot', str(root), 'locale-gen'],
['chroot', str(root), 'useradd', '--create-home', '--groups', 'wheel,kvm,video,render,audio', '--shell', '/bin/bash', request.user],
['chroot', str(root), 'chpasswd'],
['chroot', str(root), 'passwd', '--lock', 'root'],
['chroot', str(root), 'systemd-machine-id-setup'],
['chroot', str(root), 'dracut', '--force', '--no-hostonly', '--omit', 'dmsquash-live',
'/boot/initramfs-<kernel>.img', '<kernel>'],
grub_command(root), ['chroot', str(root), 'grub-mkconfig', '-o', '/boot/grub/grub.cfg'],
]
lines = [f'HydraOS {"EXECUTION" if request.execute else "dry-run"}: erase all data on {target}.',
'GPT: exactly 1 GiB FAT32 ESP, remaining space ext4 root.',
f'Prepared payload: {request.payload}; account: {request.user}; timezone: {request.timezone}.',
'Commands (UUIDs/kernel discovered after unpacking; no commands run in dry-run):']
lines.extend(' ' + shlex.join(c) for c in commands)
lines.extend(['Write UUID fstab, en_US.UTF-8 and ru_RU.UTF-8, timezone, installed dracut/GRUB defaults.',
'Remove allowlisted live-mode files, hydra-live account and /home/hydra-live; reset machine-id.',
'Ensure wheel,kvm,video,render,audio groups exist before creating the installed account.',
'chpasswd receives the TTY-entered password on private stdin; root remains locked.',
'Unmount tracked sys/proc/dev/ESP/root in reverse order on success or failure.',
f'To execute, add --execute --confirm-erase {target} (as root; no automatic sudo).'])
return '\n'.join(lines)
def target_path(root, relative, *, symlinks=False):
"""Resolve a path inside a target, interpreting absolute symlinks as chroot paths."""
root = Path(root).resolve()
parts = list(PurePosixPath(relative).parts)
if parts and parts[0] == '/':
parts.pop(0)
resolved, hops = [], 0
while parts:
part = parts.pop(0)
if part in ('', '.'):
continue
if part == '..':
if not resolved:
raise InstallError('Target path escapes root')
resolved.pop()
continue
candidate = root.joinpath(*resolved, part)
if candidate.is_symlink():
if not symlinks:
raise InstallError(f'Refusing symlink traversal in target: {relative}')
hops += 1
if hops > 40:
raise InstallError('Target symlink cycle')
link = PurePosixPath(os.readlink(candidate))
extra = list(link.parts)
if link.is_absolute():
resolved = []
extra.pop(0)
parts = extra + parts
else:
resolved.append(part)
return root.joinpath(*resolved)
def remove_live_files(root):
directory = target_path(root, 'etc/hydra/live-mode')
if not directory.is_dir():
raise InstallError('Payload missing /etc/hydra/live-mode contract')
manifest = target_path(root, 'etc/hydra/live-mode/remove-on-install.json')
marker = target_path(root, 'etc/hydra/live-mode/active')
if not manifest.is_file() or not marker.is_file():
raise InstallError('Payload live-mode marker/manifest missing')
try:
paths = json.loads(manifest.read_text())
except (OSError, ValueError) as exc:
raise InstallError('Invalid live-mode removal manifest') from exc
if not isinstance(paths, list):
raise InstallError('Live removal manifest must be a JSON array')
removals = []
for path in paths:
if not isinstance(path, str) or not path.startswith('/') or '..' in PurePosixPath(path).parts or str(PurePosixPath(path)) != path:
raise InstallError('Invalid path in live removal manifest')
if not any(re.fullmatch(re.escape(pattern).replace(r'\*', r'[A-Za-z0-9_.-]+'), path) for pattern in LIVE_REMOVABLE):
raise InstallError(f'Live manifest path is not allowlisted: {path}')
leaf = PurePosixPath(path)
parent = target_path(root, str(leaf.parent))
candidate = parent / leaf.name
if candidate.exists() and not candidate.is_file() and not candidate.is_symlink():
raise InstallError('Live manifest may only remove files/symlinks')
removals.append(candidate)
for path in removals:
path.unlink(missing_ok=True)
# shutil.rmtree never follows directory symlinks inside this directory.
shutil.rmtree(directory)
def remove_live_account(root, run):
"""Remove the payload's fixed local live identity, before binding live /proc."""
passwd = target_path(root, 'etc/passwd')
try:
rows = [row.split(':') for row in passwd.read_text().splitlines() if row.split(':', 1)[0] == 'hydra-live']
except OSError as exc:
raise InstallError('Payload missing readable /etc/passwd') from exc
if len(rows) > 1:
raise InstallError('Duplicate live account in payload')
if rows:
row = rows[0]
if len(row) != 7 or not row[2].isdigit() or int(row[2]) < 1000 or row[5] != '/home/hydra-live':
raise InstallError('Live account must have a nonroot UID and exactly /home/hydra-live')
home = target_path(root, 'home/hydra-live')
if home.exists() and not home.is_dir():
raise InstallError('Live home must be an ordinary directory')
if rows:
# /proc is not mounted yet, so the live session's identical UID cannot
# make userdel mistake a running live process for a target process.
run(['chroot', str(root), 'userdel', 'hydra-live'])
if home.exists():
shutil.rmtree(home)
for spool in ('var/mail/hydra-live', 'var/spool/mail/hydra-live'):
parent = target_path(root, str(PurePosixPath(spool).parent), symlinks=True)
leaf = parent / 'hydra-live'
if leaf.is_file() or leaf.is_symlink():
leaf.unlink()
def select_kernel(root, requested):
modules = target_path(root, '/lib/modules', symlinks=True)
if not modules.is_dir():
raise InstallError('Payload has no installed kernel modules')
versions = sorted(p.name for p in modules.iterdir() if not p.is_symlink() and p.is_dir() and KERNEL.fullmatch(p.name))
if requested:
if requested not in versions:
raise InstallError(f'Requested kernel not installed: {requested}')
version = requested
elif len(versions) == 1:
version = versions[0]
else:
raise InstallError('Select --kernel explicitly when payload has zero or multiple kernels: ' + ', '.join(versions))
if not target_path(root, '/boot/vmlinuz-' + version, symlinks=True).is_file():
raise InstallError(f'Payload missing /boot/vmlinuz-{version}')
return version
def write_config(root, relative, value, mode=0o644):
path = target_path(root, relative)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(value)
path.chmod(mode)
def install(request, runner, probe, root, password, log: TextIO):
"""Execute into an empty mount directory; inject a runner for disk-free tests."""
if not request.execute:
raise InstallError('Installation requires --execute')
if not isinstance(password, str) or not password or any(c in password for c in '\r\n\x00'):
raise InstallError('Password must be nonempty and contain no newline or NUL')
target = validate_request(request, probe)
root = Path(root)
if not root.is_absolute() or root.is_symlink() or not root.is_dir() or any(root.iterdir()):
raise InstallError('Target mount directory must be an empty absolute directory')
mounted = []
error = None
journal = []
def emit(message):
journal.append(message)
print(message, file=log, flush=True)
def run(argv, **kw):
emit('+ ' + shlex.join(argv))
return runner.run(argv, **kw)
def mount(argv, path, recursive=False):
run(argv)
mounted.append((str(path), recursive))
try:
run(['unsquashfs', '-s', request.payload])
# Recheck after preflight, directly before the first destructive command.
if validate_request(request, probe) != target:
raise InstallError('Target device changed during validation')
for argv in partition_commands(target):
run(argv)
esp, part = partitions(target)
mount(['mount', part, str(root)], root)
run(['unsquashfs', '-f', '-d', str(root), request.payload])
kernel = select_kernel(root, request.kernel)
remove_live_files(root)
remove_live_account(root, run)
if not target_path(root, '/usr/share/zoneinfo/' + request.timezone, symlinks=True).is_file():
raise InstallError('Timezone does not exist in payload: ' + request.timezone)
for folder in ('boot/efi', 'dev', 'proc', 'sys'):
target_path(root, folder).mkdir(parents=True, exist_ok=True)
mount(['mount', esp, str(root / 'boot/efi')], root / 'boot/efi')
root_uuid = run(['blkid', '-s', 'UUID', '-o', 'value', part]).strip()
esp_uuid = run(['blkid', '-s', 'UUID', '-o', 'value', esp]).strip()
if any(not re.fullmatch(r'[A-Fa-f0-9-]+', value) for value in (root_uuid, esp_uuid)):
raise InstallError('Missing or invalid partition UUID')
write_config(root, 'etc/fstab', f'UUID={root_uuid} / ext4 defaults 0 1\nUUID={esp_uuid} /boot/efi vfat umask=0077 0 2\n')
write_config(root, 'etc/locale.gen', 'en_US.UTF-8 UTF-8\nru_RU.UTF-8 UTF-8\n')
write_config(root, 'etc/locale.conf', 'LANG=en_US.UTF-8\n')
write_config(root, 'etc/timezone', request.timezone + '\n')
localtime = target_path(root, 'etc/localtime', symlinks=True)
# Replace even an absolute /etc/localtime symlink without touching the host.
localtime_link = target_path(root, 'etc') / 'localtime'
if localtime_link.is_symlink():
localtime_link.unlink()
elif localtime.exists():
localtime.unlink()
(root / 'etc/localtime').symlink_to('/usr/share/zoneinfo/' + request.timezone)
write_config(root, 'etc/machine-id', '')
dbus_machine_id = target_path(root, 'var/lib/dbus') / 'machine-id'
dbus_machine_id.unlink(missing_ok=True)
write_config(root, 'etc/dracut.conf.d/90-hydra-installed.conf',
'hostonly="no"\nhostonly_cmdline="no"\nomit_dracutmodules+=" dmsquash-live "\n# No root=live: or rd.live arguments on the installed disk.\n')
write_config(root, 'etc/default/grub', 'GRUB_DISTRIBUTOR="HydraOS"\nGRUB_CMDLINE_LINUX=""\nGRUB_TIMEOUT=3\n')
write_config(root, 'etc/sudoers.d/90-hydra-admin', '%wheel ALL=(ALL:ALL) ALL\n', 0o440)
mount(['mount', '--rbind', '/dev', str(root / 'dev')], root / 'dev', True)
run(['mount', '--make-rslave', str(root / 'dev')])
mount(['mount', '-t', 'proc', 'proc', str(root / 'proc')], root / 'proc')
mount(['mount', '--rbind', '/sys', str(root / 'sys')], root / 'sys', True)
run(['mount', '--make-rslave', str(root / 'sys')])
def chroot(*args, **kw):
return run(['chroot', str(root), *args], **kw)
chroot('locale-gen')
for group in ('wheel', 'kvm', 'video', 'render', 'audio'):
chroot('groupadd', '--force', '--system', group)
chroot('useradd', '--create-home', '--groups', 'wheel,kvm,video,render,audio', '--shell', '/bin/bash', request.user)
chroot('chpasswd', input=f'{request.user}:{password}\n', sensitive=True)
password = None
chroot('passwd', '--lock', 'root')
chroot('systemd-machine-id-setup')
chroot('dracut', '--force', '--no-hostonly', '--omit', 'dmsquash-live',
'/boot/initramfs-' + kernel + '.img', kernel)
run(grub_command(root))
chroot('grub-mkconfig', '-o', '/boot/grub/grub.cfg')
run(['sync'])
except (InstallError, OSError, KeyboardInterrupt) as exc:
error = exc
emit(f'Installation stopped: {exc}. Target data retained for diagnosis.')
finally:
password = None
if any(path == str(root) for path, _ in mounted):
try:
write_config(root, 'var/log/hydra-install.log', '\n'.join(journal) + '\n', 0o600)
except (InstallError, OSError) as exc:
emit(f'Unable to retain diagnostic in target: {exc}')
cleanup_errors = []
for path, recursive in reversed(mounted):
try:
run(['umount', *(['--recursive'] if recursive else []), path])
except (InstallError, OSError) as exc:
cleanup_errors.append(str(exc))
emit(f'cleanup failed: {exc}')
if cleanup_errors and error is None:
error = InstallError('Mount cleanup failed: ' + '; '.join(cleanup_errors))
if error:
raise InstallError(str(error)) from error
print('Installation finished; remove the live ISO before booting the installed disk.', file=log, flush=True)
def preflight(request, runner):
if any(Path(marker).exists() for marker in ('/.dockerenv', '/run/.containerenv', '/run/systemd/container')):
raise InstallError('Refusing disk installation in a container; boot the live ISO in a disposable VM')
if os.geteuid() != 0:
raise InstallError('Execution requires root inside the live environment; no automatic sudo')
if not sys.stdin.isatty():
raise InstallError('Execution requires a TTY for password entry')
if not Path('/sys/firmware/efi').is_dir():
raise InstallError('Boot the live medium in UEFI mode before installation')
for command in ('lsblk', 'findmnt', 'parted', 'partprobe', 'udevadm', 'mkfs.vfat',
'mkfs.ext4', 'mount', 'umount', 'unsquashfs', 'blkid', 'chroot', 'sync'):
if shutil.which(command) is None:
raise InstallError('Required live installer tool missing: ' + command)
runner.run(['unsquashfs', '-s', request.payload])
def prompt_password():
try:
# Only the prompt output stream is needed here; getpass opens its own
# raw controlling-TTY input. Buffered r+ requires seekability.
with open('/dev/tty', 'w') as tty:
if not tty.isatty():
raise InstallError('Password entry requires a controlling TTY')
with warnings.catch_warnings():
warnings.simplefilter('error', getpass.GetPassWarning)
password = getpass.getpass('Installed account password: ', stream=tty)
if password != getpass.getpass('Repeat password: ', stream=tty):
raise InstallError('Passwords differ')
return password
except (OSError, getpass.GetPassWarning) as exc:
raise InstallError('Cannot securely read password from the controlling TTY') from exc
def main(argv=None):
parser = argparse.ArgumentParser(description='HydraOS explicit whole-disk installer; dry-run by default')
parser.add_argument('--disk', help='explicit physical whole-disk target; never automatically selected')
parser.add_argument('--user', help='installed administrator username')
parser.add_argument('--payload', default=DEFAULT_PAYLOAD, help='prepared live squashfs root payload')
parser.add_argument('--execute', action='store_true')
parser.add_argument('--confirm-erase', help='must exactly match the resolved /dev target with --execute')
parser.add_argument('--timezone', default='Europe/Moscow')
parser.add_argument('--kernel', help='required if payload contains multiple kernel module versions')
args = parser.parse_args(argv)
runner = Runner()
try:
if not args.disk:
if args.execute or args.confirm_erase:
raise InstallError('--execute requires an explicit --disk and --user')
topo = Topology.from_json(runner.run(['lsblk', '--json', '--paths', '--bytes', '--output',
'PATH,TYPE,PKNAME,MOUNTPOINTS,SIZE,MAJ:MIN']))
print('Disks (no target selected; mounted disks are ineligible):')
for path, node in sorted(topo.nodes.items()):
if node['type'] == 'disk':
state = 'in use' if topo.component(path) & topo.protected else 'unmounted; requires full guard check'
print(f' {path} {node["size"] / 1024**3:.1f} GiB {state}')
print('Choose --disk DEVICE --user USER to inspect the dry-run plan.')
return 0
if not args.user:
raise InstallError('--user is required with --disk')
request = Request(**vars(args))
probe = SystemProbe(runner)
target = validate_request(request, probe)
print(render_plan(request, target, '/mnt/hydra-target'))
if not request.execute:
return 0
preflight(request, runner)
password = prompt_password()
state = Path(tempfile.mkdtemp(prefix='hydra-install-', dir='/run'))
state.chmod(0o700)
root = state / 'target'
root.mkdir()
logfile = state / 'installer.log'
print(f'Diagnostic log: {logfile}')
with logfile.open('w') as log:
log.write(render_plan(request, target, str(root)) + '\n')
install(request, runner, probe, root, password, log)
password = None
print(f'Installation completed. Diagnostic log: {logfile}')
return 0
except (InstallError, OSError, KeyboardInterrupt) as exc:
print(f'hydra-install: {exc}', file=sys.stderr)
return 1
if __name__ == '__main__':
raise SystemExit(main())
image/README.md
-47
View File
@@ -1,47 +0,0 @@
HydraOS builds start from the pinned amd64 systemd stage3 and pinned Gentoo
repository in `sources.json`. Run the builder as the normal user with Python 3,
Podman and GnuPG available; temporary tool environments are sufficient. It
refuses host root and work/cache paths outside the primary project checkout.
```
scripts/hydra-build fetch
scripts/hydra-build prepare
scripts/hydra-build packages --pretend
scripts/hydra-build packages
scripts/hydra-build rootfs
scripts/hydra-build iso
scripts/hydra-build status
```
`--work PATH`, `--cache PATH`, `--podman PATH` and `--gpg PATH` override defaults.
Linked worktrees use the primary checkout's `build/` and `cache/`. The build
lock prevents simultaneous writes in the same work directory. Interrupted
package installation resumes through Portage; successful steps write
`build/manifest.json`, signature evidence and the installed package/version/USE
inventory in `build/artifacts/`. Downloaded inputs are reverified on every
prepare; missing or changed files cannot be trusted from a checkpoint alone.
The dedicated rootless Podman store is `build/podman`, separate from default
Podman and Distrobox storage. The build container mounts inputs and repository
read-only and retains its writable layer across failures. Host networking makes
the existing local VPN/proxy available. `SYS_ADMIN` operates within the rootless
user namespace; package sandbox features incompatible with it are disabled.
No host disks or `/dev/kvm` are passed into the builder. Hydra never prunes
containers/images or changes host services.
Official Gentoo binpkgs require signatures. USE mismatches and selective testing
keywords can cause source builds; concurrency is four compiler processes and
one emerge job. Keep licenses and Portage's `/var/db/pkg`, world and repository
for installed-system updates. Rolling binhost contents can change; resulting
versions/USE are recorded without claiming bit-identical output. The pinned
service-keyring hash may need deliberate review when Gentoo rotates keys.
`rootfs` exports the dedicated builder into a private intermediate and assembles
an isolated sanitized payload, generic live initramfs and squashfs. `iso` checks
those artifacts and emits the UEFI `HYDRA_OS` ISO, SHA256 and build manifest in
`build/artifacts/`. The working builder is retained for package retries. The
installed Gentoo repository uses the standard rsync backend with signed
metamanifest verification; the builder continues to use its pinned snapshot.
QEMU and remote-viewer retain local patches and are excluded from upstream
binary packages on installed updates. Disposable boot/install validation is
required before treating an emitted ISO as release ready.
image/configure-rootfs.sh
-50
View File
@@ -1,50 +0,0 @@
#!/bin/bash
set -euo pipefail
if [[ $# != 1 || "$1" != /hydra-output/rootfs ]]; then
echo "configure-rootfs: expected the isolated /hydra-output/rootfs export" >&2
exit 1
fi
root=$1
python3 /hydra-source/hydra/image.py configure "$root" --source /hydra-source
# This is a separate rootfs inside the assembly container, never the host root.
mounted=()
cleanup_mounts() {
status=$?
trap - EXIT
cleanup_failed=0
for ((index=${#mounted[@]}-1; index>=0; index--)); do
umount "${mounted[index]}" || cleanup_failed=1
done
if ((cleanup_failed)); then
echo 'configure-rootfs: temporary mount cleanup failed; refusing success' >&2
status=1
fi
exit "$status"
}
trap cleanup_mounts EXIT
mount --bind /dev "$root/dev"
mounted+=("$root/dev")
mount -t proc proc "$root/proc"
mounted+=("$root/proc")
mount --bind /sys "$root/sys"
mounted+=("$root/sys")
for group in wheel kvm video render audio; do
chroot "$root" getent group "$group" >/dev/null || chroot "$root" groupadd -r "$group"
done
if ! chroot "$root" getent passwd hydra-live >/dev/null; then
chroot "$root" useradd --uid 1000 --create-home --home-dir /home/hydra-live --shell /bin/bash --groups wheel,kvm,video,render,audio hydra-live
fi
chroot "$root" usermod --lock hydra-live
chroot "$root" usermod --lock root
chroot "$root" locale-gen
ln -snf /usr/share/zoneinfo/Europe/Moscow "$root/etc/localtime"
# Enable the ordinary per-user sound services without starting a user manager.
python3 /hydra-source/hydra/image.py audio "$root"
# Set rootfs permissions now; capabilities are regenerated in the boot namespace.
chroot "$root" chown root:kvm /usr/libexec/qemu-bridge-helper
chroot "$root" chmod 0750 /usr/libexec/qemu-bridge-helper
chroot "$root" visudo --check
systemctl --root="$root" enable NetworkManager.service greetd.service hydra-network.service hydra-dnsmasq.service hydra-bridge-permissions.service getty@tty2.service serial-getty@ttyS0.service
systemctl --root="$root" set-default graphical.target
kernel=$(python3 /hydra-source/hydra/image.py kernel "$root")
chroot "$root" dracut --force --no-hostonly --no-hostonly-cmdline --conf /etc/hydra/live-mode/dracut.conf --confdir /etc/hydra/live-mode/dracut.conf.d --add dmsquash-live /boot/initramfs-live.img "$kernel"
image/dracut.conf
-6
View File
@@ -1,6 +0,0 @@
hostonly="no"
hostonly_cmdline="no"
add_dracutmodules+=" dmsquash-live "
filesystems+=" squashfs overlay iso9660 vfat ext4 "
add_drivers+=" vmd nvme xhci_pci thunderbolt usb_storage uas sd_mod virtio_pci virtio_blk virtio_scsi virtio_net virtio_gpu i915 loop squashfs overlay sr_mod ahci "
compress="zstd"
image/grub.cfg
-12
View File
@@ -1,12 +0,0 @@
set default=0
set timeout=5
insmod all_video
insmod serial
serial --unit=0 --speed=115200
terminal_input console serial
terminal_output console serial
menuentry "HydraOS live (volatile host root)" {
search --no-floppy --label HYDRA_OS --set=root
linux /boot/vmlinuz root=live:CDLABEL=HYDRA_OS rd.live.image rd.live.overlay.overlayfs=1 ro console=tty0 console=ttyS0,115200n8
initrd /boot/initramfs-live.img
}
image/overlay/metadata/layout.conf
-3
View File
@@ -1,3 +0,0 @@
masters = gentoo
thin-manifests = true
sign-manifests = false
image/overlay/profiles/repo_name
-1
View File
@@ -1 +0,0 @@
hydra
image/overlay/sys-apps/brightnessctl/Manifest
-1
View File
@@ -1 +0,0 @@
DIST brightnessctl-0.5.1.tar.gz 8284 BLAKE2B 7d2226aae10fddc606c5ce71ed4a04eda583acf245ea1532d8730c73dc02ea7aeed47a0e25ca7f9d40e7b0170bb08d87617e9f4aa082495e78894982870ca7cc SHA512 41ac86357b3b5a3d54a043140a6bdb1fb68d5080a16e86692059ad58f4b9134a8532432a3f664c70c665a4ad73ea8a94cc9b88fa84abc28f0356b8b397aaebc7
image/overlay/sys-apps/brightnessctl/brightnessctl-0.5.1.ebuild
-18
View File
@@ -1,18 +0,0 @@
# HydraOS local package for the pinned Gentoo snapshot.
EAPI=8
inherit toolchain-funcs udev
DESCRIPTION="Read and control backlight brightness without privileged launchers"
HOMEPAGE="https://github.com/Hummer12007/brightnessctl"
SRC_URI="https://github.com/Hummer12007/brightnessctl/archive/refs/tags/${PV}.tar.gz -> ${P}.tar.gz"
LICENSE="MIT"
SLOT="0"
KEYWORDS="amd64"
RDEPEND="virtual/udev"
src_compile() {
emake CC="$(tc-getCC)" VERSION="${PV}"
}
src_install() {
# Upstream's udev mode uses 0755, never its alternate setuid mode.
emake DESTDIR="${D}" PREFIX="${EPREFIX}/usr" UDEVDIR="${EPREFIX}$(get_udevdir)/rules.d" INSTALL_UDEV_RULES=1 install
dodoc README.md LICENSE
}
image/packages
-31
View File
@@ -1,31 +0,0 @@
# Host, boot medium, and recovery tools. Keep Portage's installed database/world.
app-emulation/qemu
~sys-firmware/edk2-bin-202408
>=gui-wm/labwc-0.20.0
app-emulation/virt-viewer
gui-apps/foot
media-fonts/dejavu
gui-libs/greetd
net-misc/networkmanager
net-dns/dnsmasq
net-firewall/nftables
media-video/pipewire
media-video/wireplumber
app-admin/sudo
sys-kernel/gentoo-kernel-bin
sys-kernel/linux-firmware
sys-firmware/sof-firmware
sys-firmware/intel-microcode
sys-kernel/dracut
sys-boot/grub
sys-fs/squashfs-tools
sys-fs/dosfstools
sys-fs/e2fsprogs
sys-block/parted
sys-apps/util-linux
sys-fs/mtools
dev-libs/libisoburn
dev-lang/python
sec-keys/openpgp-keys-gentoo-release
sec-keys/openpgp-keys-gentoo-auth
sys-apps/brightnessctl
image/portage/binrepos.conf/gentoo.conf
-4
View File
@@ -1,4 +0,0 @@
[gentoo]
sync-uri = https://distfiles.gentoo.org/releases/amd64/binpackages/23.0/x86-64/
priority = 10
verify-signature = yes
image/portage/make.conf
-16
View File
@@ -1,16 +0,0 @@
COMMON_FLAGS="-O2 -pipe -march=x86-64 -mtune=generic"
CFLAGS="${COMMON_FLAGS}"
CXXFLAGS="${COMMON_FLAGS}"
FCFLAGS="${COMMON_FLAGS}"
FFLAGS="${COMMON_FLAGS}"
MAKEOPTS="-j4"
EMERGE_DEFAULT_OPTS="--jobs=1 --getbinpkg --binpkg-respect-use=y --autounmask=n --usepkg-exclude='app-emulation/qemu app-emulation/virt-viewer'"
FEATURES="getbinpkg binpkg-request-signature -usersandbox -sandbox -network-sandbox -ipc-sandbox -pid-sandbox"
ACCEPT_LICENSE="@FREE @BINARY-REDISTRIBUTABLE linux-fw-redistributable no-source-code Intel-SDP"
VIDEO_CARDS="intel virgl"
USE="X dbus wayland opengl alsa"
QEMU_SOFTMMU_TARGETS="x86_64"
QEMU_USER_TARGETS=""
GRUB_PLATFORMS="efi-64"
BINPKG_GPG_VERIFY_GPG_HOME="/etc/portage/gnupg"
PORTAGE_NICENESS="10"
image/portage/package.accept_keywords/hydra
-2
View File
@@ -1,2 +0,0 @@
>=gui-wm/labwc-0.20.0 ~amd64
gui-libs/wlroots ~amd64
image/portage/package.use/hydra
-10
View File
@@ -1,10 +0,0 @@
app-emulation/qemu spice usb usbredir virgl opengl wayland pipewire -gtk -sdl
app-emulation/virt-viewer spice -libvirt
net-misc/spice-gtk usbredir
media-video/pipewire sound-server bluetooth dbus pipewire-alsa
net-misc/networkmanager wifi
sys-boot/grub mount
media-libs/freetype harfbuzz
sys-kernel/installkernel dracut
sys-fs/squashfs-tools zstd
image/portage/repos.conf/gentoo.conf
-5
View File
@@ -1,5 +0,0 @@
[gentoo]
location = /var/db/repos/gentoo
sync-type = git
sync-uri = https://github.com/gentoo/gentoo.git
auto-sync = no
image/portage/repos.conf/hydra.conf
-4
View File
@@ -1,4 +0,0 @@
[hydra]
location = /var/db/repos/hydra
masters = gentoo
auto-sync = no
image/rootfs/etc/default/grub
-3
View File
@@ -1,3 +0,0 @@
GRUB_DISTRIBUTOR="HydraOS"
GRUB_TIMEOUT=5
GRUB_CMDLINE_LINUX_DEFAULT="console=tty0 console=ttyS0,115200n8"
image/rootfs/etc/dracut.conf.d/50-hydra.conf
-3
View File
@@ -1,3 +0,0 @@
hostonly="no"
hostonly_cmdline="no"
omit_dracutmodules+=" dmsquash-live "
image/rootfs/etc/greetd/config.toml
-5
View File
@@ -1,5 +0,0 @@
[terminal]
vt = 1
[default_session]
command = "/usr/bin/agreety --cmd /usr/bin/hydra-session"
user = "greetd"
image/rootfs/etc/greetd/live.toml
-8
View File
@@ -1,8 +0,0 @@
[terminal]
vt = 1
[default_session]
command = "/usr/bin/agreety --cmd /usr/bin/hydra-session"
user = "greetd"
[initial_session]
command = "/usr/bin/hydra-session"
user = "hydra-live"
image/rootfs/etc/hydra/defaults.toml
-5
View File
@@ -1,5 +0,0 @@
network = "bridge"
bridge = "hydrabr0"
width = 1920
height = 1080
refresh_millihz = 144000
image/rootfs/etc/hydra/dnsmasq.conf
-11
View File
@@ -1,11 +0,0 @@
interface=hydrabr0
bind-dynamic
except-interface=lo
listen-address=10.203.0.1
dhcp-range=10.203.0.20,10.203.0.254,255.255.255.0,12h
dhcp-option=option:router,10.203.0.1
dhcp-option=option:dns-server,10.203.0.1
dhcp-leasefile=/var/lib/hydra/dnsmasq.leases
resolv-file=/run/NetworkManager/resolv.conf
domain-needed
bogus-priv
image/rootfs/etc/hydra/live-mode/active
-1
View File
@@ -1 +0,0 @@
HydraOS live medium: changes to the host root are not persistent.
image/rootfs/etc/hydra/live-mode/remove-on-install.json
-7
View File
@@ -1,7 +0,0 @@
[
"/etc/greetd/live.toml",
"/etc/sudoers.d/90-hydra-live",
"/etc/systemd/system/greetd.service.d/90-hydra-live.conf",
"/etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf",
"/etc/systemd/system/serial-getty@ttyS0.service.d/90-hydra-live.conf"
]
image/rootfs/etc/locale.conf
-1
View File
@@ -1 +0,0 @@
LANG=en_US.UTF-8
image/rootfs/etc/locale.gen
-2
View File
@@ -1,2 +0,0 @@
en_US.UTF-8 UTF-8
ru_RU.UTF-8 UTF-8
image/rootfs/etc/motd
-5
View File
@@ -1,5 +0,0 @@
HydraOS: fullscreen VMs with an ordinary Gentoo host underneath.
Live host changes are volatile. Use sudo hydra-install for the guarded installer.
Use hydra --help, nmcli, or nmtui to configure VMs and network connections.
Ctrl+Alt+F2 opens local recovery; return to the graphical session on Ctrl+Alt+F1.
Guest bridge: 10.203.0.0/24. Routed LAN access requires a route on your LAN router.
image/rootfs/etc/portage/bashrc
-6
View File
@@ -1,6 +0,0 @@
# Preserve the scoped bridge-helper permission after ordinary QEMU updates.
post_pkg_postinst() {
if [[ ${CATEGORY}/${PN} == app-emulation/qemu ]]; then
"${EROOT}usr/libexec/hydra-bridge-permissions" || die "Hydra bridge permissions failed"
fi
}
image/rootfs/etc/portage/repos.conf/gentoo.conf
-11
View File
@@ -1,11 +0,0 @@
[gentoo]
location = /var/db/repos/gentoo
sync-type = rsync
sync-uri = rsync://rsync.gentoo.org/gentoo-portage
auto-sync = yes
sync-rsync-verify-metamanifest = yes
sync-rsync-verify-max-age = 3
sync-openpgp-key-package = sec-keys/openpgp-keys-gentoo-release
sync-openpgp-key-path = /usr/share/openpgp-keys/gentoo-release.asc
sync-openpgp-keyserver = hkps://keys.gentoo.org
sync-webrsync-verify-signature = yes
image/rootfs/etc/qemu/bridge.conf
-1
View File
@@ -1 +0,0 @@
allow hydrabr0
image/rootfs/etc/sudoers.d/50-hydra-wheel
-1
View File
@@ -1 +0,0 @@
%wheel ALL=(ALL:ALL) ALL
image/rootfs/etc/sudoers.d/90-hydra-live
-1
View File
@@ -1 +0,0 @@
hydra-live ALL=(ALL:ALL) NOPASSWD: ALL
image/rootfs/etc/sysctl.d/50-hydra-network.conf
-1
View File
@@ -1 +0,0 @@
net.ipv4.ip_forward = 1
image/rootfs/etc/systemd/logind.conf.d/50-hydra.conf
-3
View File
@@ -1,3 +0,0 @@
[Login]
KillUserProcesses=no
UserStopDelaySec=infinity
image/rootfs/etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf
-3
View File
@@ -1,3 +0,0 @@
[Service]
ExecStart=
ExecStart=-/sbin/agetty --autologin hydra-live --noclear %I $TERM
image/rootfs/etc/systemd/system/greetd.service.d/90-hydra-live.conf
-3
View File
@@ -1,3 +0,0 @@
[Service]
ExecStart=
ExecStart=/usr/bin/greetd --config /etc/greetd/live.toml
image/rootfs/etc/systemd/system/hydra-bridge-permissions.service
-8
View File
@@ -1,8 +0,0 @@
[Unit]
Description=Restrict QEMU bridge helper to HydraOS VM users
Before=greetd.service
[Service]
Type=oneshot
ExecStart=/usr/libexec/hydra-bridge-permissions
[Install]
WantedBy=multi-user.target
image/rootfs/etc/systemd/system/hydra-dnsmasq.service
-12
View File
@@ -1,12 +0,0 @@
[Unit]
Description=DHCP and DNS for HydraOS guests
Requires=hydra-network.service
PartOf=hydra-network.service
After=hydra-network.service NetworkManager.service
[Service]
Type=simple
StateDirectory=hydra
ExecStart=/usr/sbin/dnsmasq --keep-in-foreground --conf-file=/etc/hydra/dnsmasq.conf
Restart=on-failure
[Install]
WantedBy=multi-user.target
image/rootfs/etc/systemd/system/hydra-network.service
-12
View File
@@ -1,12 +0,0 @@
[Unit]
Description=HydraOS guest bridge and Internet NAT
Requires=NetworkManager.service
PartOf=NetworkManager.service
After=NetworkManager.service
Before=hydra-dnsmasq.service
[Service]
Type=oneshot
RemainAfterExit=yes
ExecStart=/usr/bin/hydra-network
[Install]
WantedBy=multi-user.target
image/rootfs/etc/systemd/system/serial-getty@ttyS0.service.d/90-hydra-live.conf
-3
View File
@@ -1,3 +0,0 @@
[Service]
ExecStart=
ExecStart=-/sbin/agetty --autologin hydra-live --noclear %I $TERM
image/rootfs/usr/libexec/hydra-bridge-permissions
-7
View File
@@ -1,7 +0,0 @@
#!/bin/sh
set -eu
helper=/usr/libexec/qemu-bridge-helper
[ -f "$helper" ] && [ ! -L "$helper" ]
chown root:kvm "$helper"
chmod 0750 "$helper"
setcap cap_net_admin+ep "$helper"
image/sources.json
-27
View File
@@ -1,27 +0,0 @@
{
"stage3": {
"url": "https://distfiles.gentoo.org/releases/amd64/autobuilds/20261011T170049Z/stage3-amd64-systemd-20261011T170049Z.tar.xz",
"sha256": "50634e8cb87384b94a6dbf78cd2707766c45a5a98f3823e7287d114ee998970b",
"signature_url": "https://distfiles.gentoo.org/releases/amd64/autobuilds/20261011T170049Z/stage3-amd64-systemd-20261011T170049Z.tar.xz.asc",
"fingerprint": "13EBBDBEDE7A12775DFDB1BABB572E0E2D182910"
},
"portage": {
"url": "https://github.com/gentoo/gentoo/archive/4696089e782c20b5f0c04952f6698d6a8cc1acb4.tar.gz",
"filename": "gentoo-4696089e782c20b5f0c04952f6698d6a8cc1acb4.tar.gz",
"sha256": "d24b2056f7ae26506193374b0d9f25c82d4bf547e0ee17279940169418bd3bca",
"commit": "4696089e782c20b5f0c04952f6698d6a8cc1acb4"
},
"keyring_url": "https://qa-reports.gentoo.org/output/service-keys.gpg",
"keyring_sha256": "9fe40114c956c4c4d593f6e961ac9a994b0772863f4e685608fbdd8d77e39748",
"binhost": "https://distfiles.gentoo.org/releases/amd64/binpackages/23.0/x86-64/",
"local_packages": [
{
"atom": "sys-apps/brightnessctl",
"version": "0.5.1",
"url": "https://github.com/Hummer12007/brightnessctl/archive/refs/tags/0.5.1.tar.gz",
"filename": "brightnessctl-0.5.1.tar.gz",
"sha256": "a68869e23f56ac4f2e28f1783002810ddbf10f95e1af9b48b2912fb169f46994",
"license": "MIT"
}
]
}
scripts/hydra-build
-9
View File
@@ -1,9 +0,0 @@
#!/usr/bin/env python3
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hydra.build import main
if __name__ == '__main__':
raise SystemExit(main())
tests/fixtures/installer-dracut-findmnt.json
Vendored
-205
View File
@@ -1,205 +0,0 @@
{
"filesystems": [
{
"source": "LiveOS_rootfs",
"target": "/",
"fstype": "overlay",
"maj:min": "0:31",
"options": "rw,relatime,lowerdir=/run/rootfsbase,upperdir=/run/overlayfs,workdir=/run/ovlwork,uuid=on",
"children": [
{
"source": "tmpfs",
"target": "/tmp",
"fstype": "tmpfs",
"maj:min": "0:40",
"options": "rw,nosuid,nodev,nr_inodes=1048576,inode64,usrquota"
},
{
"source": "devtmpfs",
"target": "/dev",
"fstype": "devtmpfs",
"maj:min": "0:6",
"options": "rw,nosuid,size=1930828k,nr_inodes=482707,mode=755,inode64",
"children": [
{
"source": "hugetlbfs",
"target": "/dev/hugepages",
"fstype": "hugetlbfs",
"maj:min": "0:36",
"options": "rw,nosuid,nodev,relatime,pagesize=2M"
},
{
"source": "mqueue",
"target": "/dev/mqueue",
"fstype": "mqueue",
"maj:min": "0:21",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "tmpfs",
"target": "/dev/shm",
"fstype": "tmpfs",
"maj:min": "0:24",
"options": "rw,nosuid,nodev,inode64,usrquota"
},
{
"source": "devpts",
"target": "/dev/pts",
"fstype": "devpts",
"maj:min": "0:25",
"options": "rw,nosuid,noexec,relatime,gid=5,mode=600,ptmxmode=000"
}
]
},
{
"source": "sysfs",
"target": "/sys",
"fstype": "sysfs",
"maj:min": "0:23",
"options": "rw,nosuid,nodev,noexec,relatime",
"children": [
{
"source": "fusectl",
"target": "/sys/fs/fuse/connections",
"fstype": "fusectl",
"maj:min": "0:37",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "tracefs",
"target": "/sys/kernel/tracing",
"fstype": "tracefs",
"maj:min": "0:13",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "debugfs",
"target": "/sys/kernel/debug",
"fstype": "debugfs",
"maj:min": "0:8",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "securityfs",
"target": "/sys/kernel/security",
"fstype": "securityfs",
"maj:min": "0:7",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "cgroup2",
"target": "/sys/fs/cgroup",
"fstype": "cgroup2",
"maj:min": "0:27",
"options": "rw,nosuid,nodev,noexec,relatime,nsdelegate,memory_recursiveprot,memory_hugetlb_accounting"
},
{
"source": "none",
"target": "/sys/fs/pstore",
"fstype": "pstore",
"maj:min": "0:28",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "efivarfs",
"target": "/sys/firmware/efi/efivars",
"fstype": "efivarfs",
"maj:min": "0:29",
"options": "rw,nosuid,nodev,noexec,relatime"
},
{
"source": "bpf",
"target": "/sys/fs/bpf",
"fstype": "bpf",
"maj:min": "0:30",
"options": "rw,nosuid,nodev,noexec,relatime,mode=700"
},
{
"source": "configfs",
"target": "/sys/kernel/config",
"fstype": "configfs",
"maj:min": "0:19",
"options": "rw,nosuid,nodev,noexec,relatime"
}
]
},
{
"source": "proc",
"target": "/proc",
"fstype": "proc",
"maj:min": "0:22",
"options": "rw,nosuid,nodev,noexec,relatime",
"children": [
{
"source": "systemd-1",
"target": "/proc/sys/fs/binfmt_misc",
"fstype": "autofs",
"maj:min": "0:33",
"options": "rw,relatime,fd=38,pgrp=1,timeout=0,minproto=5,maxproto=5,direct,pipe_ino=5489",
"children": [
{
"source": "binfmt_misc",
"target": "/proc/sys/fs/binfmt_misc",
"fstype": "binfmt_misc",
"maj:min": "0:42",
"options": "rw,nosuid,nodev,noexec,relatime"
}
]
}
]
},
{
"source": "tmpfs",
"target": "/run",
"fstype": "tmpfs",
"maj:min": "0:26",
"options": "rw,nosuid,nodev,size=797732k,nr_inodes=819200,mode=755,inode64",
"children": [
{
"source": "none",
"target": "/run/credentials/systemd-journald.service",
"fstype": "tmpfs",
"maj:min": "0:35",
"options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
},
{
"source": "tmpfs",
"target": "/run/user/1000",
"fstype": "tmpfs",
"maj:min": "0:46",
"options": "rw,nosuid,nodev,relatime,size=398864k,nr_inodes=99716,mode=700,uid=1000,gid=1000,inode64"
},
{
"source": "/dev/sr0",
"target": "/run/initramfs/live",
"fstype": "iso9660",
"maj:min": "11:0",
"options": "ro,relatime,nojoliet,check=s,map=n,blocksize=2048,iocharset=utf8"
},
{
"source": "/dev/loop0",
"target": "/run/rootfsbase",
"fstype": "squashfs",
"maj:min": "7:0",
"options": "ro,relatime,errors=continue"
},
{
"source": "none",
"target": "/run/credentials/getty@tty2.service",
"fstype": "tmpfs",
"maj:min": "0:47",
"options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
},
{
"source": "none",
"target": "/run/credentials/serial-getty@ttyS0.service",
"fstype": "tmpfs",
"maj:min": "0:48",
"options": "ro,nosuid,nodev,noexec,relatime,nosymfollow,size=1024k,nr_inodes=1024,mode=700,inode64,noswap"
}
]
}
]
}
]
}
tests/fixtures/installer-dracut-lsblk.json
Vendored
-40
View File
@@ -1,40 +0,0 @@
{
"blockdevices": [
{
"path": "/dev/loop0",
"type": "loop",
"size": 2220142592,
"maj:min": "7:0",
"pkname": null,
"mountpoints": [
"/run/rootfsbase"
]
},
{
"path": "/dev/sr0",
"type": "rom",
"size": 2345107456,
"maj:min": "11:0",
"pkname": null,
"mountpoints": [
"/run/initramfs/live"
]
},
{
"path": "/dev/sr1",
"type": "rom",
"size": 1073741312,
"maj:min": "11:1",
"pkname": null,
"mountpoints": []
},
{
"path": "/dev/vda",
"type": "disk",
"size": 34359738368,
"maj:min": "253:0",
"pkname": null,
"mountpoints": []
}
]
}
tests/test_build.py
-320
View File
@@ -1,320 +0,0 @@
import hashlib
import contextlib
import io
import importlib.util
import json
import lzma
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
class BuilderTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.project = Path(self.tmp.name)
spec = importlib.util.spec_from_file_location('hydra_build', Path(__file__).parents[1] / 'hydra/build.py')
self.assertIsNotNone(spec, 'builder module is required')
self.assertTrue(Path(spec.origin).is_file(), 'builder implementation is missing')
self.module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(self.module)
def tearDown(self):
self.tmp.cleanup()
def builder(self):
return self.module.Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
def test_bad_cached_hash_is_rejected(self):
artifact = self.project / 'archive'
artifact.write_bytes(b'tampered')
with self.assertRaisesRegex(self.module.BuildError, 'SHA256'):
self.module.verify_hash(artifact, hashlib.sha256(b'authentic').hexdigest())
def test_wrong_primary_signature_is_rejected(self):
status = '[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 DEADBEEF\n'
with self.assertRaisesRegex(self.module.BuildError, 'fingerprint'):
self.module.require_signature(status, '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910')
def test_signature_accepts_only_successful_matching_primary(self):
fingerprint = '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910'
self.module.require_signature('[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 ' + fingerprint, fingerprint)
with self.assertRaises(self.module.BuildError):
self.module.require_signature('[GNUPG:] BADSIG ABC test', fingerprint)
def test_refuses_root_and_outside_project(self):
with patch.object(self.module.os, 'geteuid', return_value=0):
with self.assertRaisesRegex(self.module.BuildError, 'root'):
self.builder()
with patch.object(self.module.os, 'geteuid', return_value=1000):
with self.assertRaisesRegex(self.module.BuildError, 'project'):
self.module.Builder(self.project, self.project.parent / 'unsafe', self.project / 'cache')
def test_podman_store_is_scoped_to_work(self):
argv = self.builder().podman_argv('container', 'exists', 'builder')
self.assertEqual(argv[0], 'podman')
self.assertEqual(argv[argv.index('--root') + 1], str(self.project / 'build/podman'))
self.assertIn('--runroot', argv)
self.assertEqual(argv[-3:], ['container', 'exists', 'builder'])
def test_resume_requires_actual_inventory(self):
builder = self.builder()
builder.work.mkdir()
(builder.work / 'manifest.json').write_text(json.dumps({'packages': {'complete': True}}))
self.assertFalse(builder.complete('packages'))
def test_symlink_escape_is_rejected(self):
(self.project / 'escape').symlink_to(self.project.parent, target_is_directory=True)
with self.assertRaisesRegex(self.module.BuildError, 'project'):
self.module.Builder(self.project, self.project / 'escape/unsafe', self.project / 'cache')
def test_changed_configuration_invalidates_fetch_checkpoint(self):
image = self.project / 'image'
image.mkdir()
config = image / 'sources.json'
config.write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
builder = self.builder()
with builder.locked():
(builder.cache / 'stage.tar').write_bytes(b'good')
(builder.cache / 'repo.tar').write_bytes(b'repo')
builder.record('fetch')
self.assertTrue(builder.complete('fetch'))
config.write_text('{"changed":true}')
self.assertFalse(builder.complete('fetch'))
def test_missing_fetch_archive_invalidates_checkpoint(self):
image = self.project / 'image'
image.mkdir()
(image / 'sources.json').write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
builder = self.builder()
with builder.locked():
builder.record('fetch')
self.assertFalse(builder.complete('fetch'))
def test_second_builder_cannot_share_active_work(self):
first, second = self.builder(), self.builder()
with first.locked():
with self.assertRaisesRegex(self.module.BuildError, 'running'):
with second.locked():
self.fail('overlapping build was allowed')
def test_container_uses_host_proxy_network_and_readonly_inputs(self):
builder = self.builder()
args = builder.container_run_args()
self.assertEqual(args[args.index('--network') + 1], 'host')
self.assertIn(str(self.project) + ':/hydra-source:ro', args)
self.assertIn(str(self.project / 'cache') + ':/hydra-cache:ro', args)
self.assertNotIn('--privileged', args)
def test_distinct_workspaces_have_distinct_podman_runtime_storage(self):
other = self.module.Builder(self.project, self.project / 'other-build', self.project / 'cache')
self.assertNotEqual(self.builder().runroot, other.runroot)
def existing_builder(self, source=None, cache=None, layer=None, snapshot='old-snapshot'):
"""A real executable models only the external Podman boundary."""
source = source or self.project
cache = cache or self.project / 'cache'
image = self.project / 'image'
image.mkdir(exist_ok=True)
inputs = {'stage3': {'url': 'https://example.test/stage.tar.xz',
'sha256': hashlib.sha256(lzma.compress(b'original layer')).hexdigest()},
'portage': {'filename': 'repo.tar.gz', 'commit': 'old-snapshot'}}
(image / 'sources.json').write_text(json.dumps(inputs))
builder = self.builder()
(builder.work / 'podman').mkdir(parents=True)
fixture = self.project / 'podman-fixture.json'
fixture.write_text(json.dumps({'Image': 'original-image-id',
'Mounts': [{'Source': str(source), 'Destination': '/hydra-source', 'RW': False},
{'Source': str(cache), 'Destination': '/hydra-cache', 'RW': False}],
'RootFS': {'Layers': [layer or 'sha256:' + hashlib.sha256(b'original layer').hexdigest()]},
'snapshot': snapshot}))
executable = self.project / 'fixture-podman'
executable.write_text('''#!/usr/bin/env python3
import json, pathlib, sys
root = pathlib.Path(__file__).parent
data = json.loads((root / 'podman-fixture.json').read_text())
args = sys.argv[7:]
if 'exists' in args:
sys.exit(0)
if 'inspect' in args:
print(json.dumps([data]))
elif args[0] == 'top':
print('HPID ARGS')
if data.get('active_job'):
print('123 /usr/bin/python3 /usr/lib/python-exec/python3/emerge --update package')
elif args[0] == 'cp':
snapshot = data.get('snapshot_pending') if args[1].endswith('.hydra-snapshot-input') else data['snapshot']
if snapshot is None:
sys.exit(1)
pathlib.Path(args[-1]).write_text(snapshot + '\\n')
elif args[0] == 'exec' and data.get('stopped'):
sys.exit(125)
elif args[0] == 'exec' and 'packages.lock' in args[-1]:
sys.exit(75 if data.get('job_lock') else 0)
elif args[0] == 'exec' and args[-1].startswith('cat '):
snapshot = data['snapshot'] or (data.get('snapshot_pending') if '.hydra-snapshot-input' in args[-1] else None)
if snapshot is None:
sys.exit(1)
print(snapshot)
else:
if args[0] == 'start':
data['stopped'] = False
(root / 'podman-fixture.json').write_text(json.dumps(data))
with (root / 'mutations.jsonl').open('a') as stream:
stream.write(json.dumps(args) + '\\n')
''')
executable.chmod(0o755)
builder.podman = str(executable)
with builder.locked():
(builder.cache / 'stage.tar.xz').write_bytes(lzma.compress(b'original layer'))
return builder, inputs
def assert_prepare_refused(self, builder, inputs, reason):
with patch.object(builder, 'fetch', return_value=inputs):
with self.assertRaisesRegex(self.module.BuildError, reason):
builder.prepare()
self.assertFalse((self.project / 'mutations.jsonl').exists(), 'refusal must precede container/config mutations')
self.assertNotIn('prepare', builder.manifest(), 'refusal must not record false provenance')
def test_other_worktree_mount_is_refused_before_preparation(self):
builder, inputs = self.existing_builder(source=self.project / 'other-worktree')
self.assert_prepare_refused(builder, inputs, 'source')
def test_other_cache_mount_is_refused_before_preparation(self):
builder, inputs = self.existing_builder(cache=self.project / 'other-cache')
self.assert_prepare_refused(builder, inputs, 'cache')
def test_changed_stage3_pin_refuses_old_container(self):
builder, inputs = self.existing_builder()
archive = lzma.compress(b'new stage layer')
(builder.cache / 'stage.tar.xz').write_bytes(archive)
inputs['stage3']['sha256'] = hashlib.sha256(archive).hexdigest()
self.assert_prepare_refused(builder, inputs, 'stage3')
def test_changed_snapshot_pin_refuses_existing_tree(self):
builder, inputs = self.existing_builder()
inputs['portage']['commit'] = 'new-snapshot'
self.assert_prepare_refused(builder, inputs, 'snapshot')
def test_matching_existing_builder_retains_compatible_provenance(self):
builder, inputs = self.existing_builder()
with patch.object(builder, 'fetch', return_value=inputs):
builder.prepare()
self.assertEqual(builder.manifest()['prepare'].get('stage3_sha256'), inputs['stage3']['sha256'])
self.assertEqual(builder.manifest()['prepare']['snapshot'], 'old-snapshot')
def test_status_invalidates_missing_package_inventory(self):
builder = self.builder()
with builder.locked():
builder.record('packages')
output = io.StringIO()
with contextlib.redirect_stdout(output):
builder.status()
self.assertFalse(json.loads(output.getvalue())['packages']['complete'])
def test_status_cli_runs_during_active_build_without_mutations(self):
builder = self.builder()
self.module.subprocess.run(['git', 'init', '--quiet', str(self.project)], check=True)
output = io.StringIO()
with builder.locked():
builder.record('packages')
manifest_before = (builder.work / 'manifest.json').read_bytes()
with patch.object(self.module, '__file__', str(self.project / 'hydra/build.py')):
with contextlib.redirect_stdout(output):
result = self.module.main(['status'])
self.assertEqual(result, 0, 'status must not acquire the exclusive build lock')
data = json.loads(output.getvalue())
self.assertTrue(data['running'])
self.assertFalse(data['packages']['complete'])
self.assertEqual((builder.work / 'manifest.json').read_bytes(), manifest_before)
def test_detached_emerge_refuses_preparation_before_fetch_or_writes(self):
builder, inputs = self.existing_builder()
fixture = self.project / 'podman-fixture.json'
data = json.loads(fixture.read_text())
data['active_job'] = True
fixture.write_text(json.dumps(data))
with patch.object(builder, 'fetch', return_value=inputs) as fetch:
with self.assertRaisesRegex(self.module.BuildError, 'package job'):
builder.prepare()
fetch.assert_not_called()
self.assertFalse((self.project / 'mutations.jsonl').exists())
def test_contained_job_lock_refuses_preparation_before_fetch(self):
builder, inputs = self.existing_builder()
fixture = self.project / 'podman-fixture.json'
data = json.loads(fixture.read_text())
data['job_lock'] = True
fixture.write_text(json.dumps(data))
with patch.object(builder, 'fetch', return_value=inputs) as fetch:
with self.assertRaisesRegex(self.module.BuildError, 'package job'):
builder.prepare()
fetch.assert_not_called()
def test_status_detects_detached_emerge_without_outer_lock(self):
builder, inputs = self.existing_builder()
fixture = self.project / 'podman-fixture.json'
data = json.loads(fixture.read_text())
data['active_job'] = True
fixture.write_text(json.dumps(data))
output = io.StringIO()
with contextlib.redirect_stdout(output):
builder.status()
status = json.loads(output.getvalue())
self.assertTrue(status['running'])
self.assertTrue(status['package_job_running'])
self.assertFalse(status['mutation_lock_held'])
def test_package_job_records_failure_and_log_without_host_collection(self):
self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
script = self.module.package_job_script(['/bin/sh', '-c', 'echo diagnostic; exit 7'],
self.project / 'job', self.project / 'pkgdb', 'inputs-one')
result = self.module.subprocess.run(['python3', '-c', script])
self.assertEqual(result.returncode, 7)
status = json.loads((self.project / 'job/packages.status.json').read_text())
self.assertFalse(status['running'])
self.assertEqual(status['exit_code'], 7)
self.assertIn('diagnostic', (self.project / 'job/packages.log').read_text())
self.assertFalse((self.project / 'job/packages.json').exists())
def test_package_job_persists_inventory_after_success_without_host_collection(self):
self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
package = self.project / 'pkgdb/category/example-1.0'
package.mkdir(parents=True)
(package / 'USE').write_text('wayland opengl')
(package / 'BUILD_TIME').write_text('123')
script = self.module.package_job_script(['/bin/sh', '-c', 'exit 0'], self.project / 'job',
self.project / 'pkgdb', 'inputs-one')
result = self.module.subprocess.run(['python3', '-c', script])
self.assertEqual(result.returncode, 0)
self.assertEqual(json.loads((self.project / 'job/packages.json').read_text()),
[{'package': 'category/example-1.0', 'use': ['wayland', 'opengl'], 'build_time': '123'}])
self.assertEqual(json.loads((self.project / 'job/packages.status.json').read_text())['exit_code'], 0)
def test_matching_stopped_builder_checks_provenance_before_start(self):
builder, inputs = self.existing_builder()
fixture = self.project / 'podman-fixture.json'
data = json.loads(fixture.read_text())
data['stopped'] = True
fixture.write_text(json.dumps(data))
with patch.object(builder, 'fetch', return_value=inputs):
try:
builder.prepare()
except self.module.BuildError as error:
self.fail(f'Compatible stopped builder must resume: {error}')
self.assertTrue(builder.manifest()['prepare']['complete'])
def test_unfinished_snapshot_is_not_reported_as_prepared(self):
builder, inputs = self.existing_builder(snapshot=None)
fixture = self.project / 'podman-fixture.json'
data = json.loads(fixture.read_text())
data['snapshot_pending'] = 'old-snapshot'
fixture.write_text(json.dumps(data))
builder.record('prepare')
self.assertFalse(builder.complete('prepare'))
if __name__ == '__main__':
unittest.main()
tests/test_image.py
-579
View File
@@ -1,579 +0,0 @@
import importlib.util
import bz2
import json
import os
import shlex
import sys
from pathlib import Path
import subprocess
import tempfile
import tomllib
import unittest
try:
from hydra import image
except ImportError:
image = None
class ImageTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(image, 'image pipeline is missing')
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name) / 'rootfs'
self.root.mkdir()
def file(self, relative, content='preserved'):
path = self.root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
return path
def kernel(self, version='6.12.58-gentoo-dist'):
(self.root / 'lib/modules' / version).mkdir(parents=True)
self.file('boot/vmlinuz-' + version, 'kernel')
return version
def test_kernel_from_payload_not_running_host(self):
self.kernel()
self.assertEqual(image.kernel_version(self.root), '6.12.58-gentoo-dist')
def test_real_gentoo_kernel_bin_layout(self):
version = '6.18.54-gentoo-dist-bin'
(self.root / 'lib/modules' / version).mkdir(parents=True)
kernel = self.file('boot/kernel-' + version, 'real Gentoo kernel')
self.assertEqual(image.kernel_version(self.root), version)
self.assertEqual(image.kernel_image(self.root), kernel)
def test_configured_real_kernel_layout_is_accepted_by_installer(self):
from hydra.installer import select_kernel
version = '6.18.54-gentoo-dist-bin'
(self.root / 'lib/modules' / version).mkdir(parents=True)
kernel = self.file('boot/kernel-' + version, 'real Gentoo kernel')
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
self.assertEqual(select_kernel(self.root, None), version)
alias = self.root / ('boot/vmlinuz-' + version)
self.assertTrue(alias.samefile(kernel))
self.assertEqual(image.kernel_image(self.root), alias)
def test_exported_repository_sync_works_with_extracted_tree(self):
import configparser
self.kernel()
self.file('etc/portage/repos.conf/gentoo.conf',
'[gentoo]\nlocation=/var/db/repos/gentoo\nsync-type=git\nauto-sync=no\n')
self.file('var/db/repos/gentoo/profiles/repo_name', 'gentoo')
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
config = configparser.ConfigParser()
config.read(self.root / 'etc/portage/repos.conf/gentoo.conf')
repo = config['gentoo']
self.assertEqual(repo['sync-type'], 'rsync')
self.assertEqual(repo['sync-uri'], 'rsync://rsync.gentoo.org/gentoo-portage')
self.assertEqual(repo['auto-sync'], 'yes')
self.assertEqual(repo['sync-rsync-verify-metamanifest'], 'yes')
self.assertEqual(repo['sync-openpgp-key-path'], '/usr/share/openpgp-keys/gentoo-release.asc')
self.assertFalse((self.root / 'var/db/repos/gentoo/.git').exists())
source = Path(__file__).resolve().parents[1]
builder_config = (source / 'image/portage/repos.conf/gentoo.conf').read_text()
self.assertIn('auto-sync = no', builder_config)
def test_conflicting_kernel_images_refused(self):
version = self.kernel()
self.file('boot/kernel-' + version, 'different kernel')
with self.assertRaises(image.ImageError):
image.kernel_version(self.root)
def test_ambiguous_or_missing_kernel_refused(self):
with self.assertRaises(image.ImageError):
image.kernel_version(self.root)
self.kernel('6.1-one')
self.kernel('6.1-two')
with self.assertRaises(image.ImageError):
image.kernel_version(self.root)
def test_module_directory_without_kernel_refused(self):
self.kernel()
(self.root / 'boot/vmlinuz-6.12.58-gentoo-dist').unlink()
with self.assertRaises(image.ImageError):
image.kernel_version(self.root)
def test_mount_faults_clean_only_successful_bindings(self):
source = Path(__file__).resolve().parents[1]
tools = Path(self.tmp.name) / 'tools'
tools.mkdir()
for name in ('python3', 'chroot', 'ln', 'systemctl'):
script = tools / name
script.write_text('#!/bin/sh\nexit 0\n')
script.chmod(0o755)
mount = tools / 'mount'
mount.write_text('#!/bin/sh\nn=$(cat "$STATE/count" 2>/dev/null || echo 0)\n'
'n=$((n+1)); echo "$n" > "$STATE/count"\n'
'test "$n" != "$FAIL_MOUNT"\n')
mount.chmod(0o755)
unmount = tools / 'umount'
unmount.write_text('#!/bin/sh\nprintf "%s\\n" "$@" >> "$STATE/unmounts"\n'
'test "$FAIL_UNMOUNT" != 1\n')
unmount.chmod(0o755)
for failed, cleanup_failed in ((1, 0), (2, 0), (3, 0), (0, 0), (0, 1)):
with self.subTest(failed=failed, cleanup_failed=cleanup_failed):
state = Path(self.tmp.name) / f'state-{failed}-{cleanup_failed}'
state.mkdir()
env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'],
STATE=str(state), FAIL_MOUNT=str(failed), FAIL_UNMOUNT=str(cleanup_failed))
result = subprocess.run(['bash', str(source / 'image/configure-rootfs.sh'), '/hydra-output/rootfs'],
env=env, capture_output=True, text=True)
expected = ['/hydra-output/rootfs/dev', '/hydra-output/rootfs/proc', '/hydra-output/rootfs/sys']
expected = list(reversed(expected[:failed - 1 if failed else 3]))
actual = (state / 'unmounts').read_text().splitlines() if (state / 'unmounts').exists() else []
self.assertEqual(actual, expected)
self.assertEqual(result.returncode == 0, not failed and not cleanup_failed)
def test_retry_guard_uses_valid_real_findmnt_options(self):
from hydra.build import rootfs_retry_guard_script
self.file('.hydra-export-owned')
script = 'set -eu\nroot=' + shlex.quote(str(self.root)) + '\n' + rootfs_retry_guard_script()
result = subprocess.run(['bash', '-c', script], capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse(self.root.exists())
def test_retry_refuses_removing_owned_rootfs_with_stale_child_mount(self):
from hydra.build import rootfs_retry_guard_script
marker = self.file('.hydra-export-owned')
preserved = self.file('dev/guest-data')
tools = Path(self.tmp.name) / 'guard-tools'
tools.mkdir()
findmnt = tools / 'findmnt'
findmnt.write_text('#!/bin/sh\nprintf "%s\\n" / "$STALE_MOUNT"\n')
findmnt.chmod(0o755)
env = dict(os.environ, PATH=str(tools) + ':' + os.environ['PATH'], STALE_MOUNT=str(self.root / 'dev'))
script = 'set -eu\nroot=' + shlex.quote(str(self.root)) + '\n' + rootfs_retry_guard_script()
result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
self.assertNotEqual(result.returncode, 0)
self.assertTrue(marker.is_file())
self.assertTrue(preserved.is_file())
self.assertIn('mounted', result.stderr)
env['STALE_MOUNT'] = str(Path(self.tmp.name) / 'different-rootfs')
result = subprocess.run(['bash', '-c', script], env=env, capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse(self.root.exists())
def test_export_sanitization_preserves_update_inputs(self):
preserved = ['var/db/pkg/sys-apps/portage-3/USE', 'var/lib/portage/world',
'var/db/repos/gentoo/profiles/repo_name', 'usr/share/openpgp-keys/gentoo-release.asc',
'etc/portage/patches/app-emulation/qemu/patch.patch', 'etc/portage/make.conf']
removed = ['etc/portage/gnupg/private-keys-v1.d/key', 'etc/portage/gnupg.hydra-untrusted/key',
'var/lib/hydra-build/packages.log', 'root/.ssh/id_ed25519', 'root/.bash_history',
'var/cache/binpkgs/package.tbz2', 'var/tmp/portage/build/log', 'run/container-state',
'var/log/emerge.log', 'var/lib/systemd/random-seed', 'hydra-source/private', 'hydra-cache/private']
for name in preserved + removed:
self.file(name)
self.file('etc/machine-id', 'builder-id')
self.file('etc/resolv.conf', 'nameserver 127.0.0.1')
image.sanitize_rootfs(self.root)
for name in preserved:
self.assertEqual((self.root / name).read_text(), 'preserved', name)
for name in removed:
self.assertFalse((self.root / name).exists(), name)
self.assertEqual((self.root / 'etc/machine-id').read_text(), '')
self.assertEqual((self.root / 'etc/hostname').read_text(), 'hydra-os\n')
self.assertTrue((self.root / 'etc/resolv.conf').is_symlink())
def test_export_filters_saved_environment_proxies_but_retains_package_metadata(self):
self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
# Portable stand-in for the external Portage filter API, not the scrubber.
helper = self.file('usr/lib/portage/python3.14/filter-bash-environment.py',
"import re\ndef filter_bash_environment(pattern, source, output):\n"
" for line in source:\n"
" match = re.match(rb'declare -x ([A-Za-z_][A-Za-z0-9_]*)=', line)\n"
" if not match or not pattern.fullmatch(match.group(1)):\n"
" output.write(line)\n")
path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
path.parent.mkdir(parents=True)
package_metadata = b'declare -x LICENSE="GPL-2"\ndeclare -x EAPI="8"\n'
path.write_bytes(bz2.compress(package_metadata + b'declare -x HTTPS_PROXY="https://user:secret@builder.invalid"\n'
b'declare -x HTTP_PROXY="http://127.0.0.1:1234"\n'
b'declare -x NO_PROXY="localhost"\n'))
result = image.sanitize_saved_environments(self.root)
self.assertEqual(bz2.decompress(path.read_bytes()), package_metadata)
self.assertEqual(result['filtered_files'], 1)
self.assertEqual(result['variable_names'], ['HTTPS_PROXY', 'HTTP_PROXY', 'NO_PROXY'])
def test_saved_environment_filter_refuses_missing_portage_helper(self):
self.assertTrue(callable(getattr(image, 'sanitize_saved_environments', None)), 'saved environment scrubber is missing')
path = self.root / 'var/db/pkg/app-emulation/qemu-9.2.4/environment.bz2'
path.parent.mkdir(parents=True)
original = bz2.compress(b'declare -x HTTPS_PROXY="private"\n')
path.write_bytes(original)
with self.assertRaises(image.ImageError):
image.sanitize_saved_environments(self.root)
self.assertEqual(path.read_bytes(), original)
def test_sanitization_cannot_follow_exported_symlink_to_host(self):
outside = Path(self.tmp.name) / 'host'
outside.mkdir()
secret = outside / 'machine-id'
secret.write_text('host-id')
(self.root / 'etc').symlink_to(outside)
with self.assertRaises(image.ImageError):
image.sanitize_rootfs(self.root)
self.assertEqual(secret.read_text(), 'host-id')
def test_installed_configuration_restores_package_sandbox_defaults(self):
self.kernel()
self.file('etc/portage/make.conf', 'FEATURES="getbinpkg binpkg-request-signature -usersandbox -sandbox -network-sandbox -ipc-sandbox -pid-sandbox"\n')
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
features = (self.root / 'etc/portage/make.conf').read_text()
for option in ('-usersandbox', '-sandbox', '-network-sandbox', '-ipc-sandbox', '-pid-sandbox'):
self.assertNotIn(option, features)
self.assertIn('binpkg-request-signature', features)
self.assertEqual((self.root / 'usr/share/hydra/build-manifest.json').is_file(), True)
def test_installed_helper_uses_relocated_payload_and_opens_empty_recovery(self):
self.kernel()
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
terminal = Path(self.tmp.name) / 'terminal'
terminal.write_text('#!/bin/sh\nshift\nexec "$@"\n')
terminal.chmod(0o755)
manifest = Path(self.tmp.name) / 'session.json'
manifest.write_text(json.dumps({'config': str(Path(self.tmp.name) / 'config'),
'state': str(Path(self.tmp.name) / 'state'),
'runtime': str(Path(self.tmp.name) / 'runtime'),
'terminal': str(terminal), 'names': []}))
script = r"""import json, pathlib, subprocess, sys
sys.path.insert(0, sys.argv[1])
from hydra.session import _module_argv
helper = _module_argv('clients', pathlib.Path(sys.argv[2]))
help_result = subprocess.run(helper[:2] + ['--help'], text=True, capture_output=True)
print(help_result.stdout)
print(help_result.stderr, file=sys.stderr)
if help_result.returncode:
sys.exit(help_result.returncode)
result = subprocess.run(helper, input=':quit\n', text=True, capture_output=True, timeout=10)
print(result.stdout)
print(result.stderr, file=sys.stderr)
sys.exit(result.returncode)
"""
result = subprocess.run([sys.executable, '-I', '-c', script, str(self.root / 'usr/lib/hydra'), str(manifest)],
cwd=self.tmp.name, text=True, capture_output=True, timeout=20)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('HydraOS local recovery', result.stdout)
self.assertIn('add', result.stdout)
def test_installed_updates_exclude_unadapted_qemu_and_viewer_binaries(self):
self.kernel()
source = Path(__file__).resolve().parents[1]
self.file('etc/portage/make.conf', (source / 'image/portage/make.conf').read_text())
image.configure_rootfs(self.root, source)
settings = subprocess.run(['bash', '-euc', 'source "$1"; printf "%s" "$EMERGE_DEFAULT_OPTS"',
'--', str(self.root / 'etc/portage/make.conf')], text=True, capture_output=True, check=True)
defaults = shlex.split(settings.stdout)
exclusions = [item.split('=', 1)[1].split() for item in defaults if item.startswith('--usepkg-exclude=')]
self.assertEqual(exclusions, [['app-emulation/qemu', 'app-emulation/virt-viewer']])
self.assertIn('--getbinpkg', defaults)
def test_export_enables_pulse_socket_without_starting_user_services(self):
source = Path(__file__).resolve().parents[1]
self.assertTrue(callable(getattr(image, 'enable_user_audio', None)), 'user audio enable is missing')
units = self.root / 'usr/lib/systemd/user'
units.mkdir(parents=True)
for name, section in [('pipewire.socket', 'Socket'), ('pipewire-pulse.socket', 'Socket'),
('wireplumber.service', 'Service')]:
target = 'sockets.target' if section == 'Socket' else 'default.target'
(units / name).write_text(f'[{section}]\n[Install]\nWantedBy={target}\n')
image.enable_user_audio(self.root)
for target, name in [('sockets.target', 'pipewire.socket'), ('sockets.target', 'pipewire-pulse.socket'),
('default.target', 'wireplumber.service')]:
link = self.root / 'etc/systemd/user' / (target + '.wants') / name
self.assertTrue(link.is_symlink(), name)
self.assertEqual(os.readlink(link), '/usr/lib/systemd/user/' + name)
def test_rootfs_configuration_separates_live_overrides_and_installed_auth(self):
source = Path(__file__).resolve().parents[1]
image.copy_overlay(self.root, source / 'image/rootfs')
installed = tomllib.loads((self.root / 'etc/greetd/config.toml').read_text())
live = tomllib.loads((self.root / 'etc/greetd/live.toml').read_text())
self.assertNotIn('initial_session', installed)
self.assertEqual(live['initial_session']['user'], 'hydra-live')
manifest = json.loads((self.root / 'etc/hydra/live-mode/remove-on-install.json').read_text())
for name in manifest:
self.assertTrue((self.root / name.lstrip('/')).is_file(), name)
self.assertIn('/etc/sudoers.d/90-hydra-live', manifest)
self.assertIn('/etc/greetd/live.toml', manifest)
self.assertNotIn('root=live:', (self.root / 'etc/dracut.conf.d/50-hydra.conf').read_text())
defaults = tomllib.loads((self.root / 'etc/hydra/defaults.toml').read_text())
self.assertEqual(defaults, dict(network='bridge', bridge='hydrabr0', width=1920, height=1080, refresh_millihz=144000))
def test_overlay_cannot_write_through_host_symlink(self):
outside = Path(self.tmp.name) / 'host'
outside.mkdir()
(self.root / 'etc').symlink_to(outside)
with self.assertRaises(image.ImageError):
image.copy_overlay(self.root, Path(__file__).resolve().parents[1] / 'image/rootfs')
self.assertEqual(list(outside.iterdir()), [])
def test_live_kernel_args_match_label_and_never_copy_payload_to_ram(self):
args = image.live_kernel_args()
self.assertIn('root=live:CDLABEL=HYDRA_OS', args)
self.assertIn('rd.live.overlay.overlayfs=1', args)
self.assertNotIn('rd.live.ram=1', args)
cfg = (Path(__file__).resolve().parents[1] / 'image/grub.cfg').read_text()
linux = next(line.strip() for line in cfg.splitlines() if line.strip().startswith('linux '))
self.assertEqual(linux.split()[2:], args)
def test_manifest_records_actual_kernel_packages_and_source_content(self):
self.kernel()
self.file('var/db/pkg/gui-wm/labwc-0.20.0/USE', 'X png')
self.file('var/db/pkg/gui-wm/labwc-0.20.0/BUILD_TIME', '123')
source = Path(__file__).resolve().parents[1]
result = image.payload_manifest(self.root, source)
self.assertEqual(result['kernel'], '6.12.58-gentoo-dist')
self.assertEqual(result['packages'], [{'package': 'gui-wm/labwc-0.20.0', 'use': ['X', 'png'], 'build_time': '123'}])
self.assertIn('hydra/session.py', result['source_files'])
self.assertIn('image/rootfs/etc/hydra/defaults.toml', result['source_files'])
def test_artifact_checkpoint_checks_bytes_not_flags(self):
path = self.file('artifacts/squashfs.img', 'actual-payload')
state = {'artifacts': {'squashfs.img': image.digest(path)}}
self.assertTrue(image.artifacts_valid(path.parent, state))
path.write_text('changed-payload')
self.assertFalse(image.artifacts_valid(path.parent, state))
path.unlink()
self.assertFalse(image.artifacts_valid(path.parent, state))
self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {}}))
self.assertFalse(image.artifacts_valid(path.parent, {'artifacts': {'../escape': '0' * 64}}))
class ImageBuildTests(unittest.TestCase):
def setUp(self):
from hydra.build import Builder
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.project = Path(self.tmp.name)
self.builder = Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
def inventory(self, records=None):
records = records if records is not None else [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]
self.builder.work.mkdir(exist_ok=True)
artifacts = self.builder.work / 'artifacts'
artifacts.mkdir(exist_ok=True)
path = artifacts / 'packages.json'
path.write_text(json.dumps(records))
original = self.builder.complete
self.builder.complete = lambda step: True if step == 'prepare' else original(step)
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
return path
def test_changed_inventory_invalidates_package_checkpoint(self):
path = self.inventory()
self.assertTrue(self.builder.complete('packages'))
path.write_text('[{"package":"app-emulation/qemu-8.0", "use": [], "build_time": null}]')
self.assertFalse(self.builder.complete('packages'))
def test_malformed_inventory_with_matching_digest_is_refused(self):
for data in ['{', '{}', '[]', '[{"package":"wrong"}]',
'[{"package":"app-emulation/qemu-9.2.4", "use":"spice", "build_time":"123"}]',
'[{"package":"app-emulation/qemu-9.2.4", "use":[], "build_time":42}]']:
with self.subTest(data=data):
path = self.inventory()
path.write_text(data)
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
self.assertFalse(self.builder.complete('packages'))
def test_changed_inventory_invalidates_completed_rootfs_and_iso(self):
path = self.inventory()
artifacts = path.parent
self.builder.active_package_job = lambda: False
self.builder.assembly = lambda snapshot=None: 'isolated-assembler'
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
def pod(*args, **kwargs):
if args[0] == 'export':
snapshot = artifacts / 'builder-rootfs.tar'
self.assertTrue(snapshot.is_file(), 'raw snapshot must be created privately before export')
self.assertEqual(snapshot.stat().st_mode & 0o777, 0o600)
snapshot.write_bytes(b'exported fixture')
elif args[0] == 'exec' and 'configure-rootfs.sh' in args[-1]:
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
(artifacts / name).write_bytes(b'artifact')
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
'packages': [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]}))
elif args[0] == 'exec' and 'grub-mkrescue' in args[-1]:
(artifacts / 'hydra-os.iso').write_bytes(b'ISO fixture')
(artifacts / 'iso-inspection.txt').write_text('HYDRA_OS\nEl Torito boot img : 1 UEFI\n')
else:
self.fail('Unexpected Podman operation: ' + repr(args))
self.builder.pod = pod
self.builder.rootfs()
self.builder.iso()
self.assertTrue(self.builder.complete('rootfs'))
self.assertTrue(self.builder.complete('iso'))
path.write_text('[{"package":"app-emulation/qemu-9.2.5", "use":[], "build_time":null}]')
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
self.assertTrue(self.builder.complete('packages'))
self.assertFalse(self.builder.complete('rootfs'))
self.assertFalse(self.builder.complete('iso'))
def test_export_inventory_must_match_validated_package_checkpoint(self):
from hydra.build import BuildError
path = self.inventory()
self.builder.complete = lambda step: step == 'packages'
self.builder.active_package_job = lambda: False
self.builder.assembly = lambda snapshot: 'isolated-assembler'
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
def pod(*args, **kwargs):
artifacts = self.builder.work / 'artifacts'
if args[0] == 'export':
(artifacts / 'builder-rootfs.tar').write_bytes(b'exported fixture')
elif args[0] == 'exec':
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
(artifacts / name).write_bytes(b'artifact')
# Actual exported package differs from the checkpoint inventory.
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
'packages': [{'package': 'app-emulation/qemu-9.2.5', 'use': ['spice'], 'build_time': '456'}]}))
else:
self.fail('Unexpected Podman operation: ' + repr(args))
self.builder.pod = pod
with self.assertRaisesRegex(BuildError, 'inventory'):
self.builder.rootfs()
self.assertFalse(self.builder.manifest()['rootfs']['complete'])
def test_corrupt_local_package_archive_invalidates_fetch_checkpoint(self):
import hashlib
source = self.project / 'image'
source.mkdir()
(source / 'sources.json').write_text(json.dumps({
'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'stage').hexdigest()},
'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()},
'local_packages': [{'url': 'https://example.test/brightnessctl.tar.gz',
'filename': 'brightnessctl.tar.gz', 'sha256': hashlib.sha256(b'original-source').hexdigest()}]}))
with self.builder.locked():
(self.builder.cache / 'stage.tar').write_bytes(b'stage')
(self.builder.cache / 'repo.tar').write_bytes(b'repo')
archive = self.builder.cache / 'brightnessctl.tar.gz'
archive.write_bytes(b'original-source')
self.builder.record('fetch')
self.assertTrue(self.builder.complete('fetch'))
archive.write_bytes(b'corrupted-source')
self.assertFalse(self.builder.complete('fetch'))
def test_changed_canonical_patch_invalidates_package_input_identity(self):
patches = self.project / 'patches'
patches.mkdir()
patch = patches / 'qemu-virtio-refresh-rate.patch'
patch.write_text('old patch')
before = self.builder.input_id()
patch.write_text('new patch')
self.assertNotEqual(self.builder.input_id(), before)
def test_running_package_job_blocks_export(self):
from hydra.build import BuildError
self.builder.active_package_job = lambda: True
self.builder.pod = lambda *args, **kw: self.fail('Running builder was modified or exported')
self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing')
with self.assertRaisesRegex(BuildError, 'running'):
self.builder.rootfs()
def test_unverified_packages_block_export(self):
from hydra.build import BuildError
self.builder.active_package_job = lambda: False
self.builder.pod = lambda *args, **kw: self.fail('Unverified builder was exported')
self.assertTrue(callable(getattr(self.builder, 'rootfs', None)), 'rootfs pipeline is missing')
with self.assertRaises(BuildError):
self.builder.rootfs()
def test_iso_requires_checked_rootfs_artifacts(self):
from hydra.build import BuildError
self.builder.pod = lambda *args, **kw: self.fail('ISO build started without checked rootfs')
self.assertTrue(callable(getattr(self.builder, 'iso', None)), 'iso pipeline is missing')
with self.assertRaises(BuildError):
self.builder.iso()
class NetworkTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(image, 'network preparation is missing')
def test_default_route_does_not_conflict(self):
image.check_network_conflicts([{'dst': 'default', 'gateway': '192.168.1.1', 'dev': 'eth0'},
{'dst': '192.168.1.0/24', 'dev': 'eth0'}], [])
def test_overlapping_lan_vpn_address_or_route_fails_closed(self):
for routes, addresses in [([{'dst': '10.203.0.0/16', 'dev': 'tun0'}], []),
([], [{'ifname': 'wlan0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.77', 'prefixlen': 24}]}]),
([{'dst': '10.203.0.8/32', 'dev': 'eth0'}], [])]:
with self.subTest(routes=routes, addresses=addresses), self.assertRaises(image.ImageError):
image.check_network_conflicts(routes, addresses)
def test_existing_owned_bridge_route_allowed(self):
image.check_network_conflicts([{'dst': '10.203.0.0/24', 'dev': 'hydrabr0'}],
[{'ifname': 'hydrabr0', 'addr_info': [{'family': 'inet', 'local': '10.203.0.1', 'prefixlen': 24}]}])
def test_malformed_data_not_silently_accepted(self):
for routes, addresses in [('bad', []), ([{'dst': 'nonsense', 'dev': 'eth0'}], []),
([], [{'ifname': 'eth0', 'addr_info': [{'family': 'inet', 'local': 'bad', 'prefixlen': 24}]}])]:
with self.subTest(routes=routes), self.assertRaises(image.ImageError):
image.check_network_conflicts(routes, addresses)
def test_network_conflict_stops_before_any_mutation(self):
calls = []
def run(argv, **kwargs):
calls.append(argv)
if argv == ['ip', '-json', '-4', 'route', 'show', 'table', 'all']:
return subprocess.CompletedProcess(argv, 0, json.dumps([{'dst': '10.203.0.0/24', 'dev': 'eth0'}]), '')
if argv == ['ip', '-json', '-4', 'address', 'show']:
return subprocess.CompletedProcess(argv, 0, '[]', '')
self.fail('Unexpected operation before conflict validation: ' + repr(argv))
with self.assertRaises(image.ImageError):
image.configure_network(run)
self.assertEqual(len(calls), 2)
def test_manual_bridge_has_nat_without_forward_filter(self):
changes = []
def run(argv, **kwargs):
responses = {
('ip', '-json', '-4', 'route', 'show', 'table', 'all'): '[]',
('ip', '-json', '-4', 'address', 'show'): '[]',
('ip', '-json', '-details', 'link', 'show'): '[]',
}
if tuple(argv) in responses:
return subprocess.CompletedProcess(argv, 0, responses[tuple(argv)], '')
if argv[0] == 'nmcli' and argv[1] == '--get-values':
return subprocess.CompletedProcess(argv, 10, '', 'not found')
if argv == ['nft', 'list', 'table', 'ip', 'hydra_nat']:
return subprocess.CompletedProcess(argv, 1, '', 'not found')
changes.append((argv, kwargs))
return subprocess.CompletedProcess(argv, 0, '', '')
image.configure_network(run)
addition = next(argv for argv, kwargs in changes if argv[:3] == ['nmcli', 'connection', 'add'])
self.assertEqual(addition[addition.index('ipv4.method') + 1], 'manual')
rules = next(kwargs['input'] for argv, kwargs in changes if argv == ['nft', '-f', '-'])
self.assertIn('masquerade', rules)
self.assertNotIn('hook forward', rules)
self.assertNotIn('reject', rules)
self.assertNotIn('flush ruleset', rules)
def test_foreign_bridge_name_refused(self):
calls = []
def run(argv, **kwargs):
calls.append(argv)
if argv[0] == 'ip':
output = '[{"ifname": "hydrabr0", "linkinfo": {"info_kind": "bridge"}}]' if 'link' in argv else '[]'
return subprocess.CompletedProcess(argv, 0, output, '')
if argv[0] == 'nmcli' and argv[1] == '--get-values':
return subprocess.CompletedProcess(argv, 10, '', 'not found')
self.fail('Foreign interface triggered a network mutation: ' + repr(argv))
with self.assertRaises(image.ImageError):
image.configure_network(run)
def test_changed_owned_connection_refused(self):
def run(argv, **kwargs):
if argv[0] == 'ip':
return subprocess.CompletedProcess(argv, 0, '[]', '')
if argv[0] == 'nmcli' and argv[1] == '--get-values':
return subprocess.CompletedProcess(argv, 0, image.CONNECTION_UUID + '\nbridge\nhydrabr0\nshared\n10.203.0.1/24\n', '')
self.fail('Changed connection triggered a network mutation: ' + repr(argv))
with self.assertRaises(image.ImageError):
image.configure_network(run)
if __name__ == '__main__':
unittest.main()
tests/test_installer.py
-715
View File
@@ -1,715 +0,0 @@
"""Installer tests never invoke a real partition, mount or bootloader command."""
import contextlib
import io
import json
import os
import pty
import stat
import struct
from types import SimpleNamespace
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch
try:
from hydra import installer as i
except ImportError:
i = None
def disk(path='/dev/vda', children=None, mounts=None, kind='disk', pkname=None):
return {'path': path, 'type': kind, 'pkname': pkname, 'mountpoints': [] if mounts is None else mounts,
'size': 32 * 1024 ** 3, 'maj:min': '252:' + str(sum((n + 1) * ord(c) for n, c in enumerate(path))), 'children': children or []}
def topology(*nodes):
return i.Topology.from_json(json.dumps({'blockdevices': list(nodes)}))
class FakeProbe:
def __init__(self, topo=None):
self.topo = topo or topology(disk())
self.calls = 0
def snapshot(self, payload):
self.calls += 1
return self.topo
def resolve(self, path):
return {'/dev/disk/by-id/test': '/dev/vda'}.get(path, path)
def physical_disk(self, path):
return path == '/dev/vda'
class FakeRunner:
def __init__(self, root, fail=None):
self.root, self.fail, self.calls, self.inputs = root, fail, [], []
def run(self, argv, *, input=None, sensitive=False):
self.calls.append(list(argv))
self.inputs.append((input, sensitive))
if self.fail == argv[0] or (argv[0] == 'chroot' and self.fail == argv[2]) or (self.fail == 'unsquashfs-extract' and argv[0] == 'unsquashfs' and '-d' in argv):
raise i.InstallError('command failed: ' + argv[0])
if argv[0] == 'unsquashfs' and '-d' in argv:
for directory in ('etc/hydra/live-mode', 'etc/systemd/system/getty@tty1.service.d',
'etc/sudoers.d', 'usr/share/zoneinfo/Europe', 'etc/locale.gen', 'lib/modules/6.12.1', 'boot'):
if directory != 'etc/locale.gen':
(self.root / directory).mkdir(parents=True, exist_ok=True)
(self.root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
(self.root / 'usr/share/zoneinfo/Europe/Moscow').touch()
(self.root / 'etc/locale.gen').write_text('# en_US.UTF-8 UTF-8\n')
(self.root / 'etc/hydra/live-mode/active').touch()
(self.root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps([
'/etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf',
'/etc/sudoers.d/90-hydra-live']))
(self.root / 'etc/systemd/system/getty@tty1.service.d/90-hydra-live.conf').touch()
(self.root / 'etc/sudoers.d/90-hydra-live').touch()
(self.root / 'boot/vmlinuz-6.12.1').touch()
if argv[0] == 'blkid':
return '11111111-2222-3333-4444-555555555555\n'
return ''
class GuardTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
def request(self, **kw):
return i.Request(**({'disk': '/dev/vda', 'user': 'emil'} | kw))
def test_confirmation_missing_and_mismatched(self):
for confirmation in (None, '/dev/vdb', '/dev/disk/by-id/test'):
with self.subTest(confirmation=confirmation), self.assertRaises(i.InstallError):
i.validate_request(self.request(execute=True, confirm_erase=confirmation), FakeProbe())
self.assertEqual(i.validate_request(self.request(execute=True, confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
def test_alias_resolved_but_confirmation_must_name_resolved_target(self):
self.assertEqual(i.validate_request(self.request(disk='/dev/disk/by-id/test', execute=True,
confirm_erase='/dev/vda'), FakeProbe()), '/dev/vda')
def test_disk_mounted_child_root_lvm_and_live_source_rejected(self):
examples = [
topology(disk(mounts=['/mnt'])),
topology(disk(children=[disk('/dev/vda1', kind='part', mounts=['/'])])),
topology(disk(children=[disk('/dev/vda2', kind='part', children=[disk('/dev/mapper/root', kind='lvm', mounts=['/'])])])),
]
live = topology(disk(children=[disk('/dev/vda1', kind='part')]))
live.protect('/dev/vda1')
examples.append(live)
for topo in examples:
with self.subTest(topo=topo), self.assertRaises(i.InstallError):
i.validate_request(self.request(), FakeProbe(topo))
def test_loop_backing_mount_protects_parent_disk(self):
topo = topology(disk(children=[disk('/dev/vda1', kind='part')]), disk('/dev/loop0', kind='loop', mounts=['/run/live']))
topo.link('/dev/loop0', '/dev/vda1')
with self.assertRaises(i.InstallError):
i.validate_request(self.request(), FakeProbe(topo))
def test_malformed_topology_fails_closed(self):
bad = ['{}', '{', '{"blockdevices":null}', '{"blockdevices":[]}',
json.dumps({'blockdevices': [disk(kind='mystery')]}),
json.dumps({'blockdevices': [disk(mounts='')] }),
json.dumps({'blockdevices': [disk(pkname='/dev/missing')]}),
json.dumps({'blockdevices': [disk(children=[disk('/dev/vda', kind='part')])]}),
json.dumps({'blockdevices': [disk('/dev/vda;reboot')]}),
json.dumps({'blockdevices': [disk(children='bad')]} )]
for data in bad:
with self.subTest(data=data), self.assertRaises(i.InstallError):
i.Topology.from_json(data)
def test_nonphysical_unknown_or_partition_target_rejected(self):
for path in ('/dev/vdb', '/tmp/fake', '/dev/vda1', '/dev/loop0', '/dev/vda;touch /tmp/x'):
with self.subTest(path=path), self.assertRaises(i.InstallError):
i.validate_request(self.request(disk=path), FakeProbe())
def test_username_timezone_and_payload_injection_rejected(self):
for username in ('root', '-u', 'a;reboot', 'a\nb', '../x', 'A', ''):
with self.subTest(username=username), self.assertRaises(i.InstallError):
i.validate_request(self.request(user=username), FakeProbe())
for timezone in ('../../etc/passwd', '/etc/passwd', 'a;reboot'):
with self.subTest(timezone=timezone), self.assertRaises(i.InstallError):
i.validate_request(self.request(timezone=timezone), FakeProbe())
for payload in ('-evil', '/tmp/a\nb'):
with self.subTest(payload=payload), self.assertRaises(i.InstallError):
i.validate_request(self.request(payload=payload), FakeProbe())
def test_plan_is_nonsecret_and_has_exact_partition_boot_commands(self):
plan = i.render_plan(self.request(), '/dev/vda', '/mnt/hydra-target')
self.assertIn('1MiB', plan)
self.assertIn('1025MiB', plan)
self.assertIn('--removable --no-nvram', plan)
self.assertIn('/run/initramfs/live/LiveOS/squashfs.img', plan)
self.assertIn('dry-run', plan)
self.assertNotIn('chpasswd:', plan)
class PipelineTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name) / 'target'
self.root.mkdir()
self.probe = FakeProbe()
self.log = io.StringIO()
self.request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
def install(self, runner, password='distinctive-private-password'):
i.install(self.request, runner, self.probe, self.root, password, self.log)
def test_full_fake_pipeline_sets_installed_defaults_and_cleans_live_files(self):
runner = FakeRunner(self.root)
self.install(runner)
self.assertIn('UUID=', (self.root / 'etc/fstab').read_text())
self.assertIn('ru_RU.UTF-8 UTF-8', (self.root / 'etc/locale.gen').read_text())
self.assertFalse((self.root / 'etc/hydra/live-mode').exists())
self.assertFalse((self.root / 'etc/sudoers.d/90-hydra-live').exists())
self.assertIn(['chroot', str(self.root), 'grub-install', '--target=x86_64-efi', '--efi-directory=/boot/efi', '--bootloader-id=HydraOS', '--removable', '--no-nvram'], runner.calls)
password_commands = [(args, data) for args, data in zip(runner.calls, runner.inputs) if 'chpasswd' in args]
self.assertEqual(len(password_commands), 1)
self.assertEqual(password_commands[0][1], ('emil:distinctive-private-password\n', True))
self.assertNotIn('distinctive-private-password', self.log.getvalue())
self.assertGreaterEqual(self.probe.calls, 2)
self.assertIn('hostonly_cmdline="no"', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
self.assertIn('omit_dracutmodules+=" dmsquash-live "', (self.root / 'etc/dracut.conf.d/90-hydra-installed.conf').read_text())
self.assertNotIn('root=live:', (self.root / 'etc/default/grub').read_text())
dracut = next(c for c in runner.calls if 'dracut' in c)
self.assertIn('6.12.1', dracut)
self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
def test_partition_format_and_extract_failures_abort_and_cleanup_only_mounted(self):
for failing in ('parted', 'mkfs.vfat', 'mkfs.ext4', 'unsquashfs-extract'):
with self.subTest(failing=failing):
runner = FakeRunner(self.root, failing)
with self.assertRaises(i.InstallError):
self.install(runner)
self.assertFalse(any('grub-install' in c for c in runner.calls))
index = next(n for n, c in enumerate(runner.calls) if c[0] == failing or (failing == 'unsquashfs-extract' and c[0] == 'unsquashfs' and '-d' in c))
self.assertTrue(all(c[0] == 'umount' for c in runner.calls[index + 1:]))
if failing == 'unsquashfs-extract':
self.assertEqual(runner.calls[-1], ['umount', str(self.root)])
self.assertIn(failing.split('-extract')[0], self.log.getvalue())
def test_unmount_failure_is_visible(self):
runner = FakeRunner(self.root, 'umount')
with self.assertRaises(i.InstallError):
self.install(runner)
self.assertIn('cleanup', self.log.getvalue())
def test_password_newlines_rejected_before_commands(self):
for password in ('x\ny', 'x\ry', '', 'x\x00y'):
runner = FakeRunner(self.root)
with self.subTest(password=password), self.assertRaises(i.InstallError):
self.install(runner, password)
self.assertEqual(runner.calls, [])
def test_dry_run_cannot_enter_install_pipeline(self):
runner = FakeRunner(self.root)
self.request.execute = False
with self.assertRaises(i.InstallError):
self.install(runner)
self.assertEqual(runner.calls, [])
def test_live_manifest_arbitrary_paths_and_symlink_parents_refused(self):
for path in ('/etc/passwd', '/home/emil/data', '/../etc/sudoers.d/90-hydra-live'):
directory = self.root / 'etc/hydra/live-mode'
directory.mkdir(parents=True, exist_ok=True)
(directory / 'active').touch()
(directory / 'remove-on-install.json').write_text(json.dumps([path]))
with self.subTest(path=path), self.assertRaises(i.InstallError):
i.remove_live_files(self.root)
outside = Path(self.tmp.name) / 'outside'
outside.mkdir()
(outside / '90-hydra-live').write_text('must remain')
(self.root / 'etc/sudoers.d').symlink_to(outside)
(directory / 'remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
with self.assertRaises(i.InstallError):
i.remove_live_files(self.root)
self.assertTrue((outside / '90-hydra-live').exists())
def test_missing_or_multiple_kernels_are_explicit_errors(self):
with self.assertRaises(i.InstallError):
i.select_kernel(self.root, None)
for version in ('6.12.1', '6.12.2'):
(self.root / 'lib/modules' / version).mkdir(parents=True)
(self.root / 'boot').mkdir(exist_ok=True)
(self.root / 'boot' / ('vmlinuz-' + version)).touch()
with self.assertRaises(i.InstallError):
i.select_kernel(self.root, None)
self.assertEqual(i.select_kernel(self.root, '6.12.2'), '6.12.2')
class RunnerTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(i, 'installer module must implement the guarded installer')
def test_sensitive_failure_never_includes_stderr_or_stdin(self):
with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 1, '', 'private-password')):
with self.assertRaises(i.InstallError) as caught:
i.Runner().run(['chroot', '/target', 'chpasswd'], input='user:private-password\n', sensitive=True)
self.assertNotIn('private-password', str(caught.exception))
def test_subprocess_uses_argv_without_shell_and_checked_returncode(self):
with patch('subprocess.run', return_value=subprocess.CompletedProcess([], 0, 'ok', '')) as call:
self.assertEqual(i.Runner().run(['tool', '/tmp/path;echo bad']), 'ok')
self.assertEqual(call.call_args.args[0], ['tool', '/tmp/path;echo bad'])
self.assertFalse(call.call_args.kwargs.get('shell', False))
class ProbeTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(i, 'installer module required')
def test_device_mapper_alias_and_dev_root_resolve_by_identity(self):
topo = topology(disk('/dev/vda', children=[disk('/dev/dm-0', kind='lvm')]))
mount = {'source': '/dev/root', 'target': '/', 'fstype': 'ext4', 'maj:min': topo.nodes['/dev/vda']['maj:min'], 'options': 'rw'}
probe = i.SystemProbe()
self.assertEqual(probe._block_source(mount, topo), '/dev/vda')
self.assertIn('/dev/vda', topo.protected)
def test_parent_reference_cycles_rejected(self):
data = json.dumps({'blockdevices': [disk('/dev/a', pkname='/dev/b'), disk('/dev/b', pkname='/dev/a')]})
with self.assertRaises(i.InstallError):
i.Topology.from_json(data)
def test_malformed_findmnt_rejected(self):
for data in ('{}', '{', '{"filesystems":[]}', '{"filesystems":[{"source":"overlay"}]}'):
with self.subTest(data=data), self.assertRaises(i.InstallError):
i.SystemProbe._mounts(data)
def test_unknown_mounted_block_source_refused(self):
mount = {'source': '/dev/missing', 'target': '/', 'fstype': 'ext4', 'maj:min': '100:1', 'options': 'rw'}
with self.assertRaises(i.InstallError):
i.SystemProbe()._block_source(mount, topology(disk()))
def test_payload_symlink_never_resolves_into_host_from_target(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / 'usr/lib/modules/6.12.1').mkdir(parents=True)
(root / 'lib').symlink_to('/usr/lib')
(root / 'boot').mkdir()
(root / 'boot/vmlinuz-6.12.1').touch()
self.assertEqual(i.select_kernel(root, None), '6.12.1')
with self.assertRaises(i.InstallError):
i.target_path(root, '/lib/modules')
def test_leaf_symlink_removal_does_not_follow_link(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / 'root'
outside = Path(tmp) / 'keep'
outside.write_text('keep')
(root / 'etc/hydra/live-mode').mkdir(parents=True)
(root / 'etc/sudoers.d').mkdir()
(root / 'etc/hydra/live-mode/active').touch()
(root / 'etc/hydra/live-mode/remove-on-install.json').write_text(json.dumps(['/etc/sudoers.d/90-hydra-live']))
(root / 'etc/sudoers.d/90-hydra-live').symlink_to(outside)
i.remove_live_files(root)
self.assertEqual(outside.read_text(), 'keep')
self.assertFalse((root / 'etc/sudoers.d/90-hydra-live').is_symlink())
def test_read_only_cli_invalid_target_and_confirmation_never_prompt_password(self):
with patch.object(i, 'getpass') as password, contextlib.redirect_stderr(io.StringIO()):
self.assertEqual(i.main(['--disk', '/dev/hydra-this-device-does-not-exist', '--user', 'emil']), 1)
self.assertEqual(i.main(['--execute']), 1)
password.getpass.assert_not_called()
def test_preflight_refuses_container_before_execution(self):
with patch('os.geteuid', return_value=0), patch.object(i.sys.stdin, 'isatty', return_value=True), \
patch.object(i.Path, 'exists', return_value=True):
with self.assertRaisesRegex(i.InstallError, 'container'):
i.preflight(i.Request('/dev/vda', 'emil'), FakeRunner(Path('/unused')))
class AccountTests(unittest.TestCase):
def setUp(self):
self.assertIsNotNone(i)
def test_live_account_removed_without_touching_other_home(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / 'etc').mkdir()
(root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\nhydra-live:x:1000:1000:Live:/home/hydra-live:/bin/bash\n')
(root / 'home/hydra-live').mkdir(parents=True)
(root / 'home/hydra-live/config').touch()
(root / 'home/keep').mkdir()
runner = FakeRunner(root)
i.remove_live_account(root, runner.run)
self.assertIn(['chroot', str(root), 'userdel', 'hydra-live'], runner.calls)
self.assertFalse((root / 'home/hydra-live').exists())
self.assertTrue((root / 'home/keep').exists())
def test_live_account_unexpected_home_or_root_uid_refused(self):
for row in ('hydra-live:x:1000:1000:Live:/home/other:/bin/bash',
'hydra-live:x:0:0:Live:/home/hydra-live:/bin/bash'):
with self.subTest(row=row), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / 'etc').mkdir()
(root / 'etc/passwd').write_text(row + '\n')
runner = FakeRunner(root)
with self.assertRaises(i.InstallError):
i.remove_live_account(root, runner.run)
self.assertEqual(runner.calls, [])
def test_live_manifest_nested_unit_paths_rejected(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
live = root / 'etc/hydra/live-mode'
live.mkdir(parents=True)
(live / 'active').touch()
(live / 'remove-on-install.json').write_text(json.dumps(['/etc/systemd/system/getty@foo/bar.service.d/90-hydra-live.conf']))
with self.assertRaises(i.InstallError):
i.remove_live_files(root)
class AdditionalFailureTests(unittest.TestCase):
setUp = PipelineTests.setUp
install = PipelineTests.install
def test_later_chroot_failure_aborts_bootloader_and_unmounts_all(self):
runner = FakeRunner(self.root, 'dracut')
with self.assertRaises(i.InstallError):
self.install(runner)
self.assertFalse(any('grub-install' in c for c in runner.calls))
self.assertEqual([c for c in runner.calls if c[0] == 'umount'], [
['umount', '--recursive', str(self.root / 'sys')],
['umount', str(self.root / 'proc')],
['umount', '--recursive', str(self.root / 'dev')],
['umount', str(self.root / 'boot/efi')],
['umount', str(self.root)],
])
def test_revalidation_detects_mounted_disk_before_partitioning(self):
class ChangedProbe(FakeProbe):
def snapshot(self, payload):
self.calls += 1
return topology(disk(mounts=['/mounted'])) if self.calls > 1 else self.topo
self.probe = ChangedProbe()
runner = FakeRunner(self.root)
with self.assertRaises(i.InstallError):
self.install(runner)
self.assertFalse(any(c[0] == 'parted' for c in runner.calls))
def test_unknown_multidevice_filesystem_refused(self):
topo = topology(disk())
for kind in ('zfs', 'btrfs'):
mount = {'source': 'pool' if kind == 'zfs' else '/dev/vda',
'target': '/', 'fstype': kind, 'maj:min': '0:22', 'options': 'rw'}
with self.subTest(kind=kind), self.assertRaises(i.InstallError):
i.SystemProbe()._block_source(mount, topo)
def test_prompt_requires_controlling_tty(self):
with patch('builtins.open', side_effect=OSError('no tty')):
with self.assertRaises(i.InstallError):
i.prompt_password()
class RelocatedLiveLoopTests(unittest.TestCase):
def probe(self, *, backing_device=None, backing_rdevice=0, ioctl_error=None,
opened_device=None, opened_mode=stat.S_IFBLK):
fixtures = Path(__file__).parent / 'fixtures'
data = json.loads((fixtures / 'installer-dracut-lsblk.json').read_text())
mounts = json.loads((fixtures / 'installer-dracut-findmnt.json').read_text())
media = next(m for m in i.SystemProbe._mounts(json.dumps(mounts)) if m['source'] == '/dev/sr0')
calls = []
class ReadOnlyRunner:
def run(self, args):
calls.append(args)
if args[0] == 'lsblk':
return json.dumps(data)
if '--target' not in args:
return json.dumps(mounts)
if args[args.index('--target') + 1] != i.DEFAULT_PAYLOAD:
raise i.InstallError('Real dracut stale backing path has no current mount')
return json.dumps({'filesystems': [media]})
probe = i.SystemProbe(ReadOnlyRunner())
probe.physical_disk = lambda path: path == '/dev/vda'
packet = struct.pack('=QQQ', os.makedev(11, 0) if backing_device is None else backing_device,
3462, backing_rdevice) + bytes(208)
with contextlib.ExitStack() as stack:
stack.enter_context(patch.object(Path, 'is_file', return_value=True))
stack.enter_context(patch.object(Path, 'read_text', return_value='/LiveOS/squashfs.img'))
opened = stack.enter_context(patch('os.open', return_value=91))
closed = stack.enter_context(patch('os.close'))
stack.enter_context(patch('os.fstat', return_value=SimpleNamespace(st_mode=opened_mode,
st_rdev=os.makedev(7, 0) if opened_device is None else opened_device)))
# Patch the syscall even before the installer imports fcntl.
syscall = stack.enter_context(patch('fcntl.ioctl', side_effect=ioctl_error,
return_value=packet))
topo = probe.snapshot(i.DEFAULT_PAYLOAD)
opened.assert_called_once_with('/dev/loop0', os.O_RDONLY | os.O_CLOEXEC)
closed.assert_called_once_with(91)
return probe, topo, calls, syscall
def test_actual_switched_root_topology_resolves_loop_by_kernel_identity(self):
probe, topo, calls, syscall = self.probe()
self.assertFalse(topo.component('/dev/vda') & topo.protected)
self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
self.assertTrue(topo.component('/dev/sr0') & topo.protected)
self.assertEqual(syscall.call_args.args, (91, 0x4C05, bytes(232)))
self.assertFalse(any('/LiveOS/squashfs.img' in args for args in calls))
probe.snapshot = lambda payload: topo
self.assertEqual(i.validate_request(i.Request('/dev/vda', 'hydratest'), probe), '/dev/vda')
probe.physical_disk = lambda path: path == '/dev/sr0'
with self.assertRaises(i.InstallError):
i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
# A mounted loop backed by a normal disk still forbids that whole disk.
topo.nodes['/dev/sr0']['type'] = 'disk'
topo.nodes['/dev/sr0']['size'] = 32 * 1024 ** 3
with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
i.validate_request(i.Request('/dev/sr0', 'hydratest'), probe)
def test_unknown_loop_filesystem_identity_fails_closed(self):
for identity in (os.makedev(8, 1), os.makedev(0, 99)):
with self.subTest(identity=identity), self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
self.probe(backing_device=identity)
def test_unreadable_or_changed_loop_device_fails_closed(self):
cases = [{'ioctl_error': OSError('cannot read loop status')},
{'opened_device': os.makedev(7, 1)}, {'opened_mode': stat.S_IFREG}]
for case in cases:
with self.subTest(case=case), self.assertRaises(i.InstallError):
self.probe(**case)
def test_truncated_loop_status_fails_closed_and_closes_descriptor(self):
with patch('os.open', return_value=91), patch('os.close') as closed, \
patch('os.fstat', return_value=SimpleNamespace(st_mode=stat.S_IFBLK, st_rdev=os.makedev(7, 0))), \
patch('fcntl.ioctl', return_value=bytes(16)):
with self.assertRaisesRegex(i.InstallError, 'Malformed loop status'):
i.SystemProbe._loop_backing_identity('/dev/loop0', '7:0')
closed.assert_called_once_with(91)
def test_direct_block_backing_uses_kernel_rdevice(self):
_, topo, _, _ = self.probe(backing_device=os.makedev(0, 6), backing_rdevice=os.makedev(11, 0))
self.assertIn('/dev/loop0', topo.component('/dev/sr0'))
class NestedLoopAncestryTests(unittest.TestCase):
def probe(self, backings=None, *, indirect=False, payload_loop=False, shared=False):
loop0 = disk('/dev/loop0', kind='loop', mounts=['/mnt/also'] if shared else [])
loop0['maj:min'] = '7:0'
loop1 = disk('/dev/loop1', kind='loop', mounts=[] if payload_loop else ['/mnt/valuable'])
loop1['maj:min'] = '7:1'
target, unrelated = disk('/dev/vda'), disk('/dev/vdb')
target['maj:min'], unrelated['maj:min'] = '253:0', '253:1'
media = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
media['maj:min'] = '11:0'
nodes = [loop0, loop1, target, unrelated, media]
if indirect:
partition = disk('/dev/loop0p1', kind='part', pkname='/dev/loop0')
partition['maj:min'] = '259:0'
nodes.insert(1, partition)
identities = backings or {'/dev/loop0': '253:0',
'/dev/loop1': '253:0' if shared else '259:0' if indirect else '7:0'}
mount = {'source': '/dev/loop1' if payload_loop else '/dev/sr0',
'target': '/payload' if payload_loop else '/run/initramfs/live',
'fstype': 'ext4' if payload_loop else 'iso9660',
'maj:min': '7:1' if payload_loop else '11:0', 'options': 'ro'}
class ReadOnlyRunner:
def run(self, args):
if args[0] == 'lsblk':
return json.dumps({'blockdevices': nodes})
full_mount = {'source': '/dev/sr0', 'target': '/run/initramfs/live',
'fstype': 'iso9660', 'maj:min': '11:0', 'options': 'ro'}
return json.dumps({'filesystems': [mount] if '--target' in args else [full_mount]})
probe = i.SystemProbe(ReadOnlyRunner())
probe.physical_disk = lambda path: path in {'/dev/vda', '/dev/vdb'}
with patch.object(Path, 'is_file', return_value=True), \
patch.object(i.SystemProbe, '_loop_backing_identity',
side_effect=lambda path, expected: identities[path]):
topo = probe.snapshot(i.DEFAULT_PAYLOAD)
probe.snapshot = lambda payload: topo
return probe, topo
def test_reverse_order_nested_loop_refuses_physical_backing(self):
probe, _ = self.probe()
with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
def test_reverse_order_nested_loop_allows_unrelated_disk(self):
probe, topo = self.probe()
self.assertIn('/dev/vda', topo.component('/dev/loop1'))
self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
def test_unknown_nested_backing_refuses_unrelated_installation(self):
with self.assertRaisesRegex(i.InstallError, 'Unknown physical backing'):
self.probe({'/dev/loop1': '7:0', '/dev/loop0': '0:99'})
def test_nested_backing_cycles_fail_closed(self):
for backings in ({'/dev/loop1': '7:0', '/dev/loop0': '7:1'},
{'/dev/loop1': '7:1', '/dev/loop0': '253:0'}):
with self.subTest(backings=backings), self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
self.probe(backings)
def test_nested_backing_through_loop_partition_refuses_disk(self):
probe, _ = self.probe(indirect=True)
with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
def test_cycle_through_loop_partition_fails_closed(self):
with self.assertRaisesRegex(i.InstallError, '[Cc]yclic'):
self.probe({'/dev/loop1': '259:0', '/dev/loop0': '7:1'}, indirect=True)
def test_unmounted_payload_loop_ancestors_refuse_disk(self):
probe, _ = self.probe(payload_loop=True)
with self.assertRaisesRegex(i.InstallError, 'backs mounted/root/live storage'):
i.validate_request(i.Request('/dev/vda', 'hydratest'), probe)
def test_two_loops_on_same_backing_are_not_a_directed_cycle(self):
probe, topo = self.probe(shared=True)
self.assertIn('/dev/loop0', topo.component('/dev/loop1'))
self.assertEqual(i.validate_request(i.Request('/dev/vdb', 'hydratest'), probe), '/dev/vdb')
class SnapshotTests(unittest.TestCase):
def test_snapshot_connects_live_loop_to_its_physical_backing(self):
with tempfile.TemporaryDirectory() as tmp:
payload = Path(tmp) / 'squashfs.img'
payload.touch()
data = {'blockdevices': [disk(), disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', mounts=['/run/initramfs/live'])]), disk('/dev/loop0', kind='loop', mounts=['/'])]}
media = {'source': '/dev/vdb1', 'target': '/run/initramfs/live', 'fstype': 'iso9660', 'maj:min': data['blockdevices'][1]['children'][0]['maj:min'], 'options': 'ro'}
loop = {'source': '/dev/loop0', 'target': '/', 'fstype': 'squashfs', 'maj:min': data['blockdevices'][2]['maj:min'], 'options': 'ro'}
class ReadOnlyRunner:
def run(self, args):
if args[0] == 'lsblk':
return json.dumps(data)
return json.dumps({'filesystems': [media] if '--target' in args else [media, loop]})
with patch.object(i.SystemProbe, '_loop_backing_identity',
return_value=media['maj:min']):
topo = i.SystemProbe(ReadOnlyRunner()).snapshot(str(payload))
self.assertFalse(topo.component('/dev/vda') & topo.protected)
self.assertIn('/dev/loop0', topo.component('/dev/vdb'))
self.assertTrue(topo.component('/dev/vdb') & topo.protected)
def test_failed_extraction_retains_diagnostic_on_target(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / 'target'
root.mkdir()
request = i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
runner = FakeRunner(root, 'unsquashfs-extract')
with self.assertRaises(i.InstallError):
i.install(request, runner, FakeProbe(), root, 'secret', io.StringIO())
diagnostic = root / 'var/log/hydra-install.log'
self.assertTrue(diagnostic.is_file())
self.assertIn('unsquashfs', diagnostic.read_text())
self.assertNotIn('secret', diagnostic.read_text())
class TargetPathTests(unittest.TestCase):
def test_live_mail_parent_symlink_is_resolved_within_target(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / 'etc').mkdir()
(root / 'etc/passwd').write_text('root:x:0:0:root:/root:/bin/bash\n')
(root / 'var/spool').mkdir(parents=True)
(root / 'var/mail').mkdir()
(root / 'var/spool/mail').symlink_to('/var/mail')
i.remove_live_account(root, FakeRunner(root).run)
def test_kernel_version_symlink_is_not_followed_into_host(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / 'root'
external = Path(tmp) / 'external'
external.mkdir()
(root / 'lib/modules').mkdir(parents=True)
(root / 'lib/modules/6.12.1').symlink_to(external)
(root / 'boot').mkdir()
(root / 'boot/vmlinuz-6.12.1').touch()
with self.assertRaises(i.InstallError):
i.select_kernel(root, None)
class IdentityTests(unittest.TestCase):
def test_conflicting_device_identities_rejected(self):
a, b = disk(), disk('/dev/vdb')
b['maj:min'] = a['maj:min']
with self.assertRaises(i.InstallError):
i.Topology.from_json(json.dumps({'blockdevices': [a, b]}))
class TtyTests(unittest.TestCase):
def test_password_stream_opens_real_nonseekable_tty(self):
master, slave = pty.openpty()
try:
slave_name = os.ttyname(slave)
real_open = open
def tty_open(path, mode):
self.assertEqual(path, '/dev/tty')
return real_open(slave_name, mode, opener=lambda path, flags: os.open(path, flags | os.O_NOCTTY))
with patch('builtins.open', tty_open), patch.object(i.getpass, 'getpass', return_value='test-only-secret'):
self.assertEqual(i.prompt_password(), 'test-only-secret')
finally:
os.close(slave)
os.close(master)
class ReviewRegressionTests(unittest.TestCase):
def request(self):
return i.Request('/dev/vda', 'emil', execute=True, confirm_erase='/dev/vda')
def test_parentless_protected_stacked_root_refuses_all_targets(self):
for kind in ('lvm', 'crypt', 'dm', 'mpath', 'md', 'raid0', 'raid1', 'raid4', 'raid5', 'raid6', 'raid10'):
with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
topo = topology(disk(), disk('/dev/dm-0', kind=kind, mounts=['/']))
i.validate_request(self.request(), FakeProbe(topo))
def test_child_relationship_is_not_evidence_of_backing_parent(self):
for kind in ('part', 'lvm', 'crypt', 'raid1'):
node = disk('/dev/missing-backing', kind=kind,
children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
with self.subTest(kind=kind), self.assertRaisesRegex(i.InstallError, 'backing parent'):
topo = topology(disk(), node)
i.validate_request(self.request(), FakeProbe(topo))
def test_complete_stack_on_other_disk_preserves_eligible_target(self):
stack = disk('/dev/vdb', children=[disk('/dev/vdb1', kind='part', children=[
disk('/dev/dm-crypt', kind='crypt', children=[disk('/dev/dm-lvm', kind='lvm', mounts=['/'])])])])
topo = topology(disk(), stack)
self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
self.assertTrue(topo.component('/dev/vdb') & topo.protected)
def test_empty_optical_drive_does_not_block_eligible_target(self):
optical = disk('/dev/sr0', kind='rom')
optical['size'] = 0
topo = topology(disk(), optical)
self.assertEqual(i.validate_request(self.request(), FakeProbe(topo)), '/dev/vda')
self.assertFalse(topo.component('/dev/sr0') & topo.protected)
def test_zero_size_disk_cannot_be_selected(self):
target = disk()
target['size'] = 0
with self.assertRaises(i.InstallError):
topo = topology(target)
i.validate_request(self.request(), FakeProbe(topo))
def test_zero_size_mounted_optical_drive_is_not_treated_as_empty(self):
optical = disk('/dev/sr0', kind='rom', mounts=['/run/initramfs/live'])
optical['size'] = 0
with self.assertRaises(i.InstallError):
topology(disk(), optical)
def test_zero_size_optical_with_separately_listed_dependent_refused(self):
optical = disk('/dev/sr0', kind='rom')
optical['size'] = 0
dependent = disk('/dev/dm-0', kind='crypt', pkname='/dev/sr0', mounts=['/'])
with self.assertRaises(i.InstallError):
topology(disk(), optical, dependent)
def test_zero_size_optical_cannot_be_protected_by_findmnt(self):
optical = disk('/dev/sr0', kind='rom')
optical['size'] = 0
topo = topology(disk(), optical)
with self.assertRaises(i.InstallError):
topo.protect('/dev/sr0')
def test_zero_size_optical_with_dependents_is_not_irrelevant(self):
optical = disk('/dev/sr0', kind='rom', children=[disk('/dev/dm-0', kind='crypt', mounts=['/'])])
optical['size'] = 0
with self.assertRaises(i.InstallError):
topology(disk(), optical)
if __name__ == '__main__':
unittest.main()
tests/test_iso_smoke.py
-634
View File
@@ -1,634 +0,0 @@
"""Regression tests for disposable-VM isolation and trustworthy serial evidence."""
import importlib.util
from pathlib import Path
import socket
import select
import subprocess
import sys
import time
import tempfile
import threading
import unittest
SCRIPT = Path(__file__).resolve().parents[1] / 'dev/iso-smoke.py'
class SmokeTests(unittest.TestCase):
def setUp(self):
self.assertTrue(SCRIPT.is_file(), 'disposable ISO acceptance harness is missing')
spec = importlib.util.spec_from_file_location('iso_smoke', SCRIPT)
self.mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(self.mod)
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.code = self.root / 'code.fd'
self.variables = self.root / 'vars.fd'
self.code.write_bytes(b'firmware')
self.variables.write_bytes(b'fresh vars')
def test_existing_disk_or_symlink_cannot_be_a_test_drive(self):
disk = self.root / 'disk.qcow2'
disk.write_bytes(b'original')
with self.assertRaises(self.mod.SmokeError):
self.mod.reserve_disk(disk)
link = self.root / 'link.qcow2'
link.symlink_to(disk)
with self.assertRaises(self.mod.SmokeError):
self.mod.reserve_disk(link)
self.assertEqual(disk.read_bytes(), b'original')
def test_qemu_only_accepts_owned_regular_disk_and_readonly_iso(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
iso = self.root / 'trusted.iso'
iso.write_bytes(b'ISO')
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, iso, 2048)
drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
self.assertTrue(any('format=qcow2' in item and str(disk) in item for item in drives))
self.assertTrue(any('media=cdrom' in item and 'readonly=on' in item for item in drives))
self.assertIn('accel=kvm', ' '.join(argv))
disk.unlink()
disk.symlink_to(iso)
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, iso, 2048)
def test_installed_reboot_has_no_optical_medium(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, None, 2048)
self.assertNotIn('media=cdrom', ' '.join(argv))
self.assertNotIn('-cdrom', argv)
def test_hardlinked_disk_is_refused(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
(self.root / 'alias').hardlink_to(disk)
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, None, 2048)
def test_result_frame_ignores_command_echo_and_preserves_failure(self):
command = self.mod.frame_command('printf sentinel; false', 'testtoken')
result = subprocess.run(['/bin/sh', '-c', command], capture_output=True)
echo = command.encode() + b'\r\n'
self.assertIsNone(self.mod.parse_frame(echo, 'testtoken'))
self.assertEqual(self.mod.parse_frame(echo + result.stdout, 'testtoken'),
(1, b'sentinel'))
def test_console_redacts_password_even_when_split_across_reads(self):
one, two = socket.socketpair()
self.addCleanup(one.close)
self.addCleanup(two.close)
log = self.root / 'serial.log'
console = self.mod.Console(one, log, ['split-secret'])
self.addCleanup(console.close)
two.sendall(b'hello split-')
console.read(0.1)
two.sendall(b'secret goodbye')
console.read(0.1)
console.close()
self.assertNotIn(b'split-secret', log.read_bytes())
self.assertIn(b'[REDACTED]', log.read_bytes())
def test_console_timeout_is_bounded_and_partial_frame_not_success(self):
one, two = socket.socketpair()
self.addCleanup(one.close)
self.addCleanup(two.close)
console = self.mod.Console(one, self.root / 'serial.log', [])
self.addCleanup(console.close)
two.sendall(b'\\036token:END:0\\037')
with self.assertRaises(self.mod.SmokeError):
console.command('true', timeout=0.05)
def installer_console(self):
one, two = socket.socketpair()
self.addCleanup(two.close)
console = self.mod.Console(one, self.root / 'installer.log', ['test-secret'])
self.addCleanup(console.close)
return console, two
def test_captured_installer_refusal_fails_before_password_timeout(self):
console, guest = self.installer_console()
guest.sendall(b'\x1b]3008;type=command\x1b\\hydra-install: Command failed (1): '
b'findmnt --json --target /LiveOS/squashfs.img '
b'--output SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS: \r\n')
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*findmnt'):
console.expect_installer(rb'Installed account password: ', 0.05, 'testtoken')
def test_installer_shell_return_without_completion_is_failure_even_zero(self):
for status in (0, 1, 127):
with self.subTest(status=status):
console, guest = self.installer_console()
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:' + str(status).encode() + b'\x1f')
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*INSTALL_EXIT'):
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
0.05, 'testtoken')
def test_installer_echo_cannot_fake_exit_and_complete_prompt_still_required(self):
console, guest = self.installer_console()
guest.sendall(self.mod.installer_command('testtoken').encode() + b'\r\n'
b'Installed account password: ')
self.assertEqual(console.expect_installer(rb'Installed account password: ',
0.05, 'testtoken').group(), b'Installed account password: ')
guest.sendall(b'Installation completed. Diagnostic')
with self.assertRaisesRegex(self.mod.SmokeError, 'Timed out'):
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
0.05, 'testtoken')
guest.sendall(b' log: /var/log/hydra-install.log\r\n')
self.assertIn(b'/var/log/hydra-install.log', console.expect_installer(
rb'Installation completed\. Diagnostic log: [^\r\n]+', 0.05, 'testtoken').group())
def test_installer_wrapper_marks_actual_shell_return_and_redacts_error(self):
console, guest = self.installer_console()
fake_sudo = self.root / 'sudo'
fake_sudo.write_text('#!/bin/sh\nprintf "hydra-install: test-secret refused\\n"\nexit 1\n')
fake_sudo.chmod(0o700)
result = subprocess.run(['/bin/sh', '-c', self.mod.installer_command('testtoken')],
env={'PATH': str(self.root)}, capture_output=True, check=True)
self.assertIn(b'\x1etesttoken:INSTALL_EXIT:1\x1f', result.stdout)
guest.sendall(result.stdout)
with self.assertRaisesRegex(self.mod.SmokeError, r'Installer.*\[REDACTED\] refused') as caught:
console.expect_installer(rb'Repeat password: ', 0.05, 'testtoken')
self.assertNotIn('test-secret', str(caught.exception))
def test_installer_followup_waits_for_exit_and_new_shell_prompt(self):
console, guest = self.installer_console()
premature = []
def respond():
guest.sendall(b'Installation completed. Diagnostic log: /run/installer.log\r\n')
premature.append(bool(select.select([guest], [], [], 0.05)[0]))
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:0\x1f')
premature.append(bool(select.select([guest], [], [], 0.05)[0]))
guest.sendall(b'\x1b[?2004hhydra-live@hydra-os ~ $ ')
data = b''
while not data.endswith(b'\n'):
data += guest.recv(65536)
result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
guest.sendall(result.stdout)
worker = threading.Thread(target=respond, daemon=True)
worker.start()
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
1, 'testtoken')
console.finish_installer('testtoken', timeout=1)
self.assertEqual(console.command('printf ready', timeout=1), b'ready')
worker.join(timeout=1)
self.assertEqual(premature, [False, False], 'command reached installer before shell was ready')
def test_installer_completion_does_not_override_nonzero_return(self):
console, guest = self.installer_console()
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:1\x1fhydra-live@hydra-os ~ $ ')
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*status 1'):
console.finish_installer('testtoken', timeout=0.05)
def shutdown_vm(self, program):
one, two = socket.socketpair()
vm = self.mod.VM.__new__(self.mod.VM)
vm.console = self.mod.Console(one, self.root / 'shutdown.log', ['split-secret'])
self.addCleanup(vm.console.close)
code = ('import socket,sys,time; s=socket.socket(fileno=int(sys.argv[1])); ' + program)
vm.process = subprocess.Popen([sys.executable, '-c', code, str(two.fileno())],
pass_fds=(two.fileno(),), stdin=subprocess.DEVNULL)
two.close()
def reap():
if vm.process.poll() is None:
vm.process.terminate()
vm.process.wait(timeout=2)
self.addCleanup(reap)
return vm
def test_poweroff_drains_and_redacts_serial_until_owned_process_exits(self):
vm = self.shutdown_vm("s.sendall(b'Stopping split-'); time.sleep(.02); "
"s.sendall(b'secret service\\nPower down\\n'); s.close()")
vm.wait_poweroff(1)
output = (self.root / 'shutdown.log').read_bytes()
self.assertIn(b'Stopping [REDACTED] service\nPower down\n', output)
self.assertNotIn(b'split-secret', output)
def test_poweroff_preserves_failure_cause_and_original_timeout(self):
for program, error in (("s.sendall(b'Stop job stuck\\n'); s.close(); time.sleep(1)",
'Guest failed to power off in time'),
("s.sendall(b'QEMU shutdown failed\\n'); s.close(); sys.exit(7)",
'Test QEMU failed during poweroff')):
with self.subTest(program=program):
vm = self.shutdown_vm(program)
started = time.monotonic()
with self.assertRaisesRegex(self.mod.SmokeError, error):
vm.wait_poweroff(.15)
self.assertLess(time.monotonic() - started, .5)
output = (self.root / 'shutdown.log').read_bytes()
self.assertIn(b'Stop job stuck' if 'time.sleep' in program else b'QEMU shutdown failed', output)
if 'time.sleep' in program:
self.assertIsNone(vm.process.poll(), 'timeout must not kill or falsely pass the owned process')
def test_frame_real_shell_roundtrip_does_not_accept_sentinel_in_echo(self):
one, two = socket.socketpair()
self.addCleanup(one.close)
self.addCleanup(two.close)
console = self.mod.Console(one, self.root / 'serial.log', [])
self.addCleanup(console.close)
def guest():
data = b''
while not data.endswith(b'\n'):
data += two.recv(65536)
two.sendall(data)
result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
two.sendall(result.stdout)
thread = threading.Thread(target=guest)
thread.start()
self.assertEqual(console.command("printf 'persisted-data'", timeout=2), b'persisted-data')
thread.join(timeout=2)
def test_firmware_vars_must_be_private_copy_in_scratch(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
self.variables.unlink()
self.variables.symlink_to(self.code)
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, None, 2048)
def test_drive_outside_scratch_and_ambiguous_qemu_paths_are_refused(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
other = self.root / 'other'
other.mkdir()
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', other, disk, self.code,
self.variables, None, 2048)
comma = self.root / 'disk,cache=unsafe.qcow2'
self.mod.reserve_disk(comma)
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', self.root, comma, self.code,
self.variables, None, 2048)
def probe_fixture(self, case, live=False):
# Run the production shell against ordinary scratch files. Commands that
# query guest state are boundary fixtures; grep/find/test are real tools.
import os
root = self.root / 'guest'
for directory in ('etc/hydra', 'etc/greetd', 'etc/sudoers.d',
'etc/systemd/system/getty@tty2.service.d',
'etc/systemd/system/serial-getty@ttyS0.service.d',
'usr/share/firmware', 'proc', 'sys/firmware/efi',
'run/initramfs/live/LiveOS'):
(root / directory).mkdir(parents=True, exist_ok=True)
(root / 'etc/greetd/config.toml').write_text('normal authentication')
(root / 'usr/share/firmware/OVMF_CODE.fd').write_text('firmware')
(root / 'proc/cmdline').write_text('root=UUID=disk ro' if case != 'live-cmdline' else 'root=live:CDLABEL=HYDRA_OS rd.live.image')
if live or case == 'live-marker':
(root / 'etc/hydra/live-mode').mkdir(exist_ok=True)
(root / 'etc/hydra/live-mode/active').touch()
if live:
(root / 'run/initramfs/live/LiveOS/squashfs.img').touch()
files = {
'live-dropin': ('etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf', 'live'),
'live-sudoers': ('etc/sudoers.d/90-hydra-live', 'live'),
'autologin': ('etc/systemd/system/serial-getty@ttyS0.service.d/other.conf', 'ExecStart=agetty --autologin hydratest'),
'nopasswd': ('etc/sudoers.d/other', 'hydratest ALL=(ALL) NOPASSWD: ALL'),
'live-greetd': ('etc/greetd/live.toml', 'live'),
}
if case in files:
path, text = files[case]
(root / path).write_text(text)
if case == 'config-read-error':
unreadable = root / 'etc/sudoers.d/unreadable'
unreadable.write_text('configuration')
unreadable.chmod(0)
if case == 'cmdline-read-error':
(root / 'proc/cmdline').unlink()
commands = self.root / 'fixture-bin'
commands.mkdir(exist_ok=True)
dispatcher = commands / 'fixture'
dispatcher.write_text('''#!/usr/bin/env python3
import os, pathlib, sys
name = pathlib.Path(sys.argv[0]).name
args = sys.argv[1:]
case = os.environ['SMOKE_CASE']
if name == 'id': print('hydra-live')
elif name == 'findmnt': print('/dev/vda2' if 'SOURCE' in args else ('overlay' if os.environ['SMOKE_LIVE']=='1' else 'ext4'))
elif name == 'systemctl':
if args[0] == 'is-active':
states = [unit != case for unit in args[1:]]
print('\\n'.join('active' if state else 'failed' for state in states))
sys.exit(0 if any(states) else 3)
print('success' if 'Result' in args else '0')
elif name == 'getent':
if args[:2] == ['passwd','hydra-live']:
if case == 'account-lookup-error': sys.exit(3)
if case != 'live-account': sys.exit(2)
print('hydra-live:x:1000:1000::/home/hydra-live:/bin/bash')
elif args[0] == 'shadow': print(args[1] + ':' + ('!' if args[1]=='root' else '$6$hash') + ':1::::::')
else: print('127.0.0.1 STREAM example')
elif name == 'ip': print('10.203.0.1/24' if 'addr' in args else 'default via 10.0.2.2')
elif name == 'nmcli': print('connected')
else: print('fixture tool')
''')
dispatcher.chmod(0o755)
for command in ('id', 'findmnt', 'systemctl', 'getent', 'ip', 'nft',
'nmcli', 'sudo', 'hydra', 'hydra-install',
'qemu-system-x86_64', 'qemu-img', 'remote-viewer', 'foot', 'labwc'):
path = commands / command
if not path.exists():
path.symlink_to(dispatcher)
probe = self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK
for path in ('/etc/', '/usr/share', '/proc/cmdline', '/sys/firmware/efi', '/run/initramfs/live/LiveOS/squashfs.img'):
probe = probe.replace(path, str(root) + path)
import shutil
for command in ('python3', 'sh', 'grep', 'find', 'cut', 'readlink', 'base64'):
target = shutil.which(command)
self.assertIsNotNone(target)
(commands / command).symlink_to(target) if not (commands / command).exists() else None
env = dict(os.environ, PATH=str(commands),
SMOKE_CASE=case, SMOKE_LIVE='1' if live else '0')
result = subprocess.run(['/bin/sh', '-c', self.mod.frame_command(probe, 'probecheck')],
env=env, capture_output=True, timeout=10)
parsed = self.mod.parse_frame(result.stdout, 'probecheck')
self.assertIsNotNone(parsed, result.stderr)
return parsed[0]
def test_installed_probe_rejects_each_retained_live_artifact(self):
self.assertEqual(self.probe_fixture('clean'), 0)
for case in ('live-cmdline', 'live-account', 'live-dropin', 'live-sudoers',
'autologin', 'nopasswd', 'live-marker', 'live-greetd'):
with self.subTest(case=case):
# Give each negative case an independent clean filesystem.
import shutil
shutil.rmtree(self.root / 'guest')
self.assertNotEqual(self.probe_fixture(case), 0)
def test_installed_probe_does_not_treat_account_query_error_as_absence(self):
self.assertNotEqual(self.probe_fixture('account-lookup-error'), 0)
def test_installed_probe_rejects_unreadable_configuration(self):
self.assertNotEqual(self.probe_fixture('config-read-error'), 0)
def test_installed_probe_rejects_unreadable_boot_arguments(self):
self.assertNotEqual(self.probe_fixture('cmdline-read-error'), 0)
def test_installed_probe_rejects_each_failed_required_service(self):
self.assertEqual(self.probe_fixture('clean'), 0)
for service in ('NetworkManager', 'greetd', 'hydra-network', 'hydra-dnsmasq'):
with self.subTest(service=service):
self.assertNotEqual(self.probe_fixture(service), 0)
def test_live_probe_rejects_each_failed_required_service(self):
self.assertEqual(self.probe_fixture('clean', live=True), 0)
for service in ('NetworkManager', 'hydra-network', 'hydra-dnsmasq'):
with self.subTest(service=service):
self.assertNotEqual(self.probe_fixture(service, live=True), 0)
def tcp_probe(self, fragments):
# Execute the exact Python client embedded in the guest network probe,
# redirecting only its address to a fresh loopback socket.
import ast
import time
source = self.mod.NETWORK_CHECK.split("<<'NETWORK'\n", 1)[1].rsplit('\nNETWORK\n', 1)[0]
assignments = (node for node in ast.walk(ast.parse(source)) if isinstance(node, ast.Assign))
value = next(node.value for node in assignments
if any(isinstance(target, ast.Name) and target.id == 'client_code' for target in node.targets))
client = ast.parse(ast.literal_eval(value))
listener = socket.socket()
self.addCleanup(listener.close)
listener.bind(('127.0.0.1', 0))
listener.listen(1)
listener.settimeout(3)
host, port = listener.getsockname()
class Redirect(ast.NodeTransformer):
def visit_Tuple(self, node):
if isinstance(node.elts[0], ast.Constant) and node.elts[0].value == '10.203.0.250':
return ast.Tuple(elts=[ast.Constant(host), ast.Constant(port)], ctx=ast.Load())
return self.generic_visit(node)
client = ast.fix_missing_locations(Redirect().visit(client))
def server():
connection, _ = listener.accept()
with connection:
for fragment in fragments:
try:
connection.sendall(fragment)
except BrokenPipeError:
break
time.sleep(.05)
thread = threading.Thread(target=server)
thread.start()
result = subprocess.run(['python3', '-c', ast.unparse(client)], capture_output=True, timeout=5)
thread.join(timeout=3)
self.assertFalse(thread.is_alive())
return result
def test_tcp_probe_accepts_complete_marker_across_fragmented_reads(self):
result = self.tcp_probe([b'HYDRA_', b'LAN_TO_', b'VM_OK'])
self.assertEqual(result.returncode, 0, result.stderr)
def test_tcp_probe_rejects_truncated_marker(self):
self.assertNotEqual(self.tcp_probe([b'HYDRA_LAN_TO_']).returncode, 0)
def lifecycle_fixture(self):
import json
import os
import uuid
name = 'logout-test'
runtime = self.root / 'private'
run = runtime / name
run.mkdir(parents=True, mode=0o700)
runtime.chmod(0o700)
audio = self.root / 'user-runtime'
(audio / 'pulse').mkdir(parents=True)
context = {'name': name, 'uuid': str(uuid.uuid4()), 'runtime': str(runtime),
'user_runtime': str(audio), 'uid': os.getuid()}
process = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
'-name', name, '-uuid', context['uuid']])
self.addCleanup(lambda: process.wait(timeout=3))
self.addCleanup(lambda: process.poll() is None and process.kill())
(run / 'qemu.pid').write_text(str(process.pid))
(run / 'identity.json').write_text(json.dumps({'name': name, 'uuid': context['uuid']}))
listeners = []
for path in (run / 'qmp.sock', run / 'spice.sock', audio / 'pipewire-0', audio / 'pulse/native'):
listener = socket.socket(socket.AF_UNIX)
listener.bind(str(path))
listener.listen(5)
listener.settimeout(.1)
listeners.append(listener)
self.addCleanup(listener.close)
stopped = threading.Event()
def qmp_server():
while not stopped.is_set():
try:
connection, _ = listeners[0].accept()
except socket.timeout:
continue
except OSError:
break
with connection, connection.makefile('rwb', buffering=0) as stream:
stream.write(b'{"QMP": {}}\n')
for line in stream:
request = json.loads(line)
replies = {'query-name': {'name': context.get('qmp_name', name)},
'query-uuid': {'UUID': context['uuid']},
'query-status': {'status': 'running'},
'query-kvm': {'enabled': True, 'present': True}}
if request['execute'] == 'query-kvm' and 'on_snapshot' in context:
context.pop('on_snapshot')()
stream.write(json.dumps({'return': replies.get(request['execute'], {}),
'id': request['id']}).encode() + b'\n')
thread = threading.Thread(target=qmp_server)
thread.start()
self.addCleanup(lambda: thread.join(timeout=3))
self.addCleanup(stopped.set)
return context, process, run
def test_logout_probe_detects_lost_private_socket(self):
context, process, run = self.lifecycle_fixture()
before = self.mod.lifecycle_snapshot(context)
self.assertEqual(self.mod.lifecycle_snapshot(context), before)
(run / 'spice.sock').unlink()
with self.assertRaises(self.mod.SmokeError):
self.mod.lifecycle_snapshot(context)
def test_logout_probe_rejects_stale_metadata_and_foreign_qmp_identity(self):
import json
context, process, run = self.lifecycle_fixture()
self.mod.lifecycle_snapshot(context)
(run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': 'stale'}))
with self.assertRaises(self.mod.SmokeError):
self.mod.lifecycle_snapshot(context)
(run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': context['uuid']}))
context['qmp_name'] = 'foreign-process'
with self.assertRaises(self.mod.SmokeError):
self.mod.lifecycle_snapshot(context)
def test_logout_probe_rejects_dead_process_and_changed_runtime_socket(self):
context, process, run = self.lifecycle_fixture()
before = self.mod.lifecycle_snapshot(context)
audio = Path(context['user_runtime']) / 'pipewire-0'
audio.unlink()
replacement = socket.socket(socket.AF_UNIX)
self.addCleanup(replacement.close)
replacement.bind(str(audio))
replacement.listen(1)
after = self.mod.lifecycle_snapshot(context)
with self.assertRaises(self.mod.SmokeError):
self.mod.verify_lifecycle_survival(before, after)
process.terminate()
process.wait(timeout=3)
with self.assertRaises(self.mod.SmokeError):
self.mod.lifecycle_snapshot(context)
def test_logout_probe_accepts_verified_same_runtime_and_rejects_process_restart(self):
context, process, run = self.lifecycle_fixture()
before = self.mod.lifecycle_snapshot(context)
self.mod.verify_lifecycle_survival(before, self.mod.lifecycle_snapshot(context))
process.terminate()
process.wait(timeout=3)
replacement = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
'-name', context['name'], '-uuid', context['uuid']])
self.addCleanup(lambda: replacement.wait(timeout=3))
self.addCleanup(lambda: replacement.poll() is None and replacement.kill())
(run / 'qemu.pid').write_text(str(replacement.pid))
changed = self.mod.lifecycle_snapshot(context)
with self.assertRaises(self.mod.SmokeError):
self.mod.verify_lifecycle_survival(before, changed)
def test_large_guest_probe_survives_serial_tty_canonical_line_limit(self):
import os
import pty
import termios
master, slave = pty.openpty()
self.addCleanup(os.close, master)
self.addCleanup(os.close, slave)
options = termios.tcgetattr(slave)
options[3] &= ~termios.ECHO
termios.tcsetattr(slave, termios.TCSANOW, options)
process = subprocess.Popen(['/bin/sh'], stdin=slave, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
self.addCleanup(process.stdout.close)
self.addCleanup(process.stderr.close)
self.addCleanup(lambda: process.wait(timeout=3))
self.addCleanup(lambda: process.poll() is None and process.kill())
payload = 'guest-proof-' * 1200
command = self.mod.frame_command('printf %s ' + payload, 'longprobe')
os.write(master, (command + '\nexit\n').encode())
output, error = process.communicate(timeout=3)
self.assertEqual(self.mod.parse_frame(output, 'longprobe'), (0, payload.encode()), error)
def test_generated_guest_lifecycle_probe_compiles(self):
compile(self.mod.lifecycle_guest_script(), '<guest-lifecycle-probe>', 'exec')
def session_fixture(self, records):
import json
import os
from unittest.mock import patch
source = self.root / 'session-records.json'
source.write_text(json.dumps(records))
commands = self.root / 'session-fixture-bin'
commands.mkdir()
command = commands / 'loginctl'
command.write_text('''#!/usr/bin/env python3
import json, os, sys
records = json.load(open(os.environ['SMOKE_SESSION_RECORDS']))
args = sys.argv[1:]
if args[0] == 'show-user':
print(' '.join(record['Id'] for record in records if str(record['User']) == args[1]))
elif args[0] == 'show-session':
record = next((record for record in records if record['Id'] == args[1]), None)
if record is None: sys.exit(1)
for key, value in record.items(): print(key + '=' + str(value))
else: sys.exit(1)
''')
command.chmod(0o755)
environment = patch.dict(os.environ, {'PATH': str(commands) + ':' + os.environ['PATH'],
'SMOKE_SESSION_RECORDS': str(source)})
environment.start()
self.addCleanup(environment.stop)
return source
def session_record(self, ident='serial1', state='active', cls='user', leader=1234):
import os
return {'Id': ident, 'User': os.getuid(), 'State': state, 'Class': cls,
'Leader': leader, 'TTY': 'ttyS0' if cls == 'user' else ''}
def test_logout_sessions_accept_closing_scope_and_manager_with_surviving_vm(self):
context, process, run = self.lifecycle_fixture()
original = self.session_record()
context['login_session'] = original
context['before'] = self.mod.lifecycle_snapshot(context)
self.session_fixture([dict(original, State='closing'),
self.session_record('manager1', 'active', 'manager')])
self.assertTrue(self.mod.no_open_logins(context))
self.assertEqual(self.mod.capture_logout_snapshot(context), context['before'])
def test_logout_sessions_initial_login_ignores_manager_but_rejects_open_replacement(self):
import json
import os
original = self.session_record()
context = {'uid': os.getuid(), 'login_session': original}
source = self.session_fixture([original, self.session_record('manager1', 'active', 'manager')])
self.assertEqual(self.mod.require_initial_login(os.getuid(), 'serial1'), original)
for state in ('online', 'active'):
with self.subTest(state=state):
source.write_text(json.dumps([dict(original, State='closing'),
self.session_record('new-login', state)]))
self.assertFalse(self.mod.no_open_logins(context))
source.write_text(json.dumps([dict(original, State='closing', Leader=9999)]))
with self.assertRaises(self.mod.SmokeError):
self.mod.no_open_logins(context)
def test_logout_sessions_reject_login_appearing_during_snapshot(self):
import json
context, process, run = self.lifecycle_fixture()
original = self.session_record()
context['login_session'] = original
context['before'] = self.mod.lifecycle_snapshot(context)
source = self.session_fixture([dict(original, State='closing')])
context['on_snapshot'] = lambda: source.write_text(json.dumps([
dict(original, State='closing'), self.session_record('relogin')]))
self.assertTrue(self.mod.no_open_logins(context))
with self.assertRaises(self.mod.SmokeError):
self.mod.capture_logout_snapshot(context)
def test_guest_tools_accept_gentoo_remote_viewer_without_virt_viewer(self):
self.assertEqual(self.probe_fixture('clean'), 0)
self.assertEqual(self.probe_fixture('clean', live=True), 0)