From 05ee077c036777d954d048835a8196ac91478f89 Mon Sep 17 00:00:00 2001
From: Codex <codex@localhost>
Date: Tue, 13 Oct 2026 04:17:55 +0300
Subject: [PATCH] Add guarded Gentoo whole-disk installer and failure tests

---
 bin/hydra-install       |  11 +
 hydra/installer.py      | 709 ++++++++++++++++++++++++++++++++++++++++
 tests/test_installer.py | 493 ++++++++++++++++++++++++++++
 3 files changed, 1213 insertions(+)
 create mode 100755 bin/hydra-install
 create mode 100644 hydra/installer.py
 create mode 100644 tests/test_installer.py

diff --git a/bin/hydra-install b/bin/hydra-install
new file mode 100755
index 0000000..66a8dff
--- /dev/null
+++ b/bin/hydra-install
@@ -0,0 +1,11 @@
+#!/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())
diff --git a/hydra/installer.py b/hydra/installer.py
new file mode 100644
index 0000000..e439e85
--- /dev/null
+++ b/hydra/installer.py
@@ -0,0 +1,709 @@
+"""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 json
+import os
+from pathlib import Path, PurePosixPath
+import re
+import shlex
+import shutil
+import stat
+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.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')
+            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())
+            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['type'] == 'part' and not result.edges[path]:
+                raise InstallError('Partition has no known parent')
+        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')
+        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}')
+        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)
+        # Mounted loops may hide the physical filesystem that stores their file.
+        for path, node in topo.nodes.items():
+            if node['type'] == 'loop' and path in topo.protected:
+                backing_file = Path('/sys/dev/block') / node['maj:min'] / 'loop/backing_file'
+                try:
+                    backing = backing_file.read_text().strip()
+                except OSError as exc:
+                    raise InstallError(f'Cannot inspect mounted loop backing file: {path}') from exc
+                backing = re.sub(r'\\([0-7]{3})', lambda m: chr(int(m[1], 8)), backing)
+                if not backing.startswith('/'):
+                    backing = '/' + backing
+                source = self._source_for_path(backing, topo)
+                if not source:
+                    raise InstallError(f'Unknown physical backing for mounted loop: {path}')
+                topo.link(path, 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)
+        return topo
+
+    @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())
diff --git a/tests/test_installer.py b/tests/test_installer.py
new file mode 100644
index 0000000..6f6b969
--- /dev/null
+++ b/tests/test_installer.py
@@ -0,0 +1,493 @@
+"""Installer tests never invoke a real partition, mount or bootloader command."""
+import contextlib
+import io
+import json
+import os
+import pty
+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 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]})
+            real_read = Path.read_text
+            def read(path, *args, **kw):
+                if str(path).startswith('/sys/dev/block/'):
+                    return str(payload)
+                return real_read(path, *args, **kw)
+            with patch.object(Path, 'read_text', read):
+                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)
+
+
+if __name__ == '__main__':
+    unittest.main()