"""Read-only validation and explicit storage plans for the installer. Only plain GPT partitions on physical disks are supported. This module never selects a target or runs a partition/format command while constructing a plan. """ from dataclasses import dataclass import hashlib import os from pathlib import Path, PurePosixPath import re ESP_TYPE = 'c12a7328-f81f-11d2-ba4b-00a0c93ec93b' UUID = re.compile(r'[A-Fa-f0-9]+(?:-[A-Fa-f0-9]+)+\Z') PARTUUID = re.compile(r'[A-Fa-f0-9]{8}(?:-[A-Fa-f0-9]{4}){3}-[A-Fa-f0-9]{12}\Z') ESP_RESERVE = 64 * 1024**2 MIN_ROOT = 4 * 1024**3 def fail(message): from .installer import InstallError raise InstallError(message) def ancestors(topo, path): seen, pending = set(), list(topo.parents[path]) while pending: current = pending.pop() if current not in seen: seen.add(current) pending.extend(topo.parents[current] - seen) return seen def fingerprint(node): return tuple(node.get(key) for key in ('path', 'type', 'size', 'maj:min', 'start', 'partuuid', 'parttype', 'fstype', 'uuid')) @dataclass(frozen=True) class DiskPlan: target: str size: int identity: tuple @dataclass(frozen=True) class PartitionPlan: root: str esp: str root_parent: str esp_parent: str esp_number: int root_size: int root_partuuid: str esp_partuuid: str esp_uuid: str identity: tuple def partition_plan(request, probe, topo): from .installer import DEVICE paths = [] for chosen in (request.root_partition, request.esp_partition): if not isinstance(chosen, str) or not DEVICE.fullmatch(chosen) or '..' in PurePosixPath(chosen).parts: fail('Choose explicit /dev root and ESP partitions') canonical = probe.resolve(chosen) if not DEVICE.fullmatch(canonical): fail('Invalid canonical partition path') paths.append(canonical) root, esp = paths if root == esp: fail('Root and ESP must be distinct partitions') if request.execute and request.confirm_format != root: fail(f'Execution requires --confirm-format {root} exactly') parents, numbers, identities = [], [], [] for path in paths: node = topo.nodes.get(path) if not node or node['type'] != 'part' or len(topo.parents[path]) != 1: fail('Target must be a plain partition with exactly one physical backing disk') parent = next(iter(topo.parents[path])) if topo.nodes[parent]['type'] != 'disk' or topo.parents[parent] or not probe.physical_disk(parent): fail('Nested or virtual partition backing is unsupported') if topo.edges[path] - {parent}: fail('Selected partition has active holders or nested devices') if any(source == path or source in ancestors(topo, path) or path in ancestors(topo, source) for source in topo.protected): fail(f'Selected partition backs mounted/root/live storage: {path}') if not PARTUUID.fullmatch(node.get('partuuid') or ''): fail('Missing or ambiguous GPT partition identity') # Check every sibling extent; never infer bounds from partition names. siblings = [p for p in topo.nodes if parent in topo.parents[p]] extents = [] for sibling in siblings: child = topo.nodes[sibling] start = child.get('start') if child['type'] != 'part' or topo.parents[sibling] != {parent} or isinstance(start, bool) or not isinstance(start, int) or start <= 0: fail('Ambiguous partition bounds or nested topology') end = start * 512 + child['size'] if child['size'] <= 0 or end > topo.nodes[parent]['size']: fail('Partition extends beyond its backing disk') extents.append((start*512, end, sibling)) extents.sort() if any(left[1] > right[0] for left, right in zip(extents, extents[1:])): fail('Overlapping partition bounds') # The partition number is provided by the kernel, not derived from an alias. number = probe.partition_number(path, node) if number <= 0: fail('Missing kernel partition number') probe.verify_identity(path, node) probe.verify_identity(parent, topo.nodes[parent]) parents.append(parent); numbers.append(number) identities.append((fingerprint(node), fingerprint(topo.nodes[parent]), tuple(sorted((sibling, fingerprint(topo.nodes[sibling])) for sibling in siblings)))) if topo.nodes[root]['size'] < MIN_ROOT: fail('Root partition is smaller than the minimum 4 GiB') esp_node = topo.nodes[esp] if esp_node.get('fstype') != 'vfat' or (esp_node.get('parttype') or '').lower() != ESP_TYPE: fail('ESP must be an existing FAT GPT EFI System Partition') if not UUID.fullmatch(esp_node.get('uuid') or ''): fail('ESP filesystem UUID missing') # Equal GPT IDs anywhere would make later PARTUUID recovery ambiguous. ids = [n.get('partuuid', '').lower() for n in topo.nodes.values() if n.get('partuuid')] if len(ids) != len(set(ids)): fail('Duplicate GPT partition identities') return PartitionPlan(root, esp, *parents, numbers[1], topo.nodes[root]['size'], topo.nodes[root]['partuuid'], esp_node['partuuid'], esp_node['uuid'], tuple(identities)) def format_root(root, filesystem): return (['mkfs.btrfs', '-f', '-L', 'HYDRA_ROOT', root] if filesystem == 'btrfs' else ['mkfs.ext4', '-F', '-L', 'HYDRA_ROOT', root]) def payload_capacity(listing): """Sum uncompressed inode sizes from unsquashfs numeric long listing.""" allocated, inodes = 0, 0 for line in listing.splitlines(): match = re.match(r'^[-dlcbps][rwxstST-]{9}\s+\d+/\d+\s+(\d+)\s+\d{4}-\d\d-\d\d\s+\d\d:\d\d\s+', line) if match: size = int(match[1]) allocated += ((size + 4095) // 4096) * 4096 + 4096 inodes += 1 elif re.match(r'^[cb][rwxstST-]{9}\s+\d+/\d+\s+\d+,\s*\d+\s+\d{4}-\d\d-\d\d\s+\d\d:\d\d\s+', line): allocated += 4096 inodes += 1 elif line and not line.startswith(('Parallel unsquashfs:', 'Number of processors', 'squashfs-root')): fail('Cannot determine uncompressed payload capacity') if not inodes: fail('Cannot determine uncompressed payload capacity') # Metadata, allocated blocks, generated initramfs, and initial user state. return max(MIN_ROOT, allocated * 12 // 10 + 1024**3) def foreign_efi_hashes(directory, *, ignore_owned=False): """Fingerprint all foreign regular files; reject ambiguous/special paths.""" root = Path(directory) result = {} for path in sorted(root.rglob('*')): relative = path.relative_to(root).as_posix() if ignore_owned and (relative.casefold() == 'efi/hydraos' or relative.casefold().startswith('efi/hydraos/')): continue if path.is_symlink() or not (path.is_file() or path.is_dir()): fail('ESP contains unsupported special files') if path.is_file(): with path.open('rb') as source: digest = hashlib.file_digest(source, 'sha256').hexdigest() result[relative] = digest return result def inspect_esp(directory): root = Path(directory) for entry in root.iterdir(): if entry.name.casefold() == 'efi': if not entry.is_dir() or entry.is_symlink(): fail('ESP EFI directory is not an ordinary directory') if any(p.name.casefold() == 'hydraos' for p in entry.iterdir()): fail('ESP already contains EFI/HydraOS; refusing loader collision') info = os.statvfs(root) if info.f_bavail * info.f_frsize < ESP_RESERVE: fail('ESP needs at least 64 MiB free for the Hydra EFI loader') return foreign_efi_hashes(root)