import logging import os import subprocess import sys from contextlib import contextmanager logger = logging.getLogger(__name__) # Import network logging for critical operations from .network_logging import log_to_discord def log_os_install(operation, status="start", details=""): log_to_discord( "INFO", f"OSINSTALL_{operation}_{status}: {details}", module="os_install" ) class CommandResult: def __init__(self, stdout, stderr, returncode): self.stdout = stdout self.stderr = stderr self.returncode = returncode def run_command(cmd, check=True): logger.info(f"Running command: {' '.join(cmd)}") process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, # Line buffered ) stdout_lines = [] stderr_lines = [] # Helper to read stream def read_stream(stream, line_list, log_level): for line in stream: line_clean = line.strip() if line_clean: log_level(line_clean) line_list.append(line) import threading t1 = threading.Thread( target=read_stream, args=(process.stdout, stdout_lines, logger.info) ) t2 = threading.Thread( target=read_stream, args=(process.stderr, stderr_lines, logger.error) ) t1.start() t2.start() t1.join() t2.join() returncode = process.wait() stdout_str = "".join(stdout_lines) stderr_str = "".join(stderr_lines) if check and returncode != 0: raise subprocess.CalledProcessError( returncode, cmd, output=stdout_str, stderr=stderr_str ) return CommandResult(stdout_str, stderr_str, returncode) @contextmanager def mount_pseudo_fs(mount_root): """ Context manager to bind mount /dev, /proc, /sys, and efivarfs into mount_root. """ logger.info(f"Mounting pseudo-filesystems to {mount_root}...") mounts = ["dev", "proc", "sys"] mounted_paths = [] try: for fs in mounts: target = os.path.join(mount_root, fs) os.makedirs(target, exist_ok=True) run_command(["mount", "--bind", f"/{fs}", target]) mounted_paths.append(target) # Mount efivarfs if it exists on the host efivars_path = "/sys/firmware/efi/efivars" if os.path.exists(efivars_path): target = os.path.join(mount_root, "sys/firmware/efi/efivars") os.makedirs(target, exist_ok=True) try: run_command(["mount", "-t", "efivarfs", "efivarfs", target]) mounted_paths.append(target) except Exception as e: logger.warning(f"Failed to mount efivarfs: {e}") yield finally: logger.info(f"Unmounting pseudo-filesystems from {mount_root}...") for path in reversed(mounted_paths): try: run_command(["umount", "-l", path]) except Exception as e: logger.warning(f"Failed to unmount {path}: {e}") def is_uefi(): return os.path.exists("/sys/firmware/efi") def install_minimal_os(mount_root, releasever="43"): """ Installs minimal Fedora packages to mount_root. """ logger.info(f"Installing minimal Fedora {releasever} to {mount_root}...") log_os_install("INSTALL", "start", f"Target: {mount_root}, Release: {releasever}") uefi = is_uefi() packages = [ "basesystem", "bash", "coreutils", "kernel", "systemd", "dnf", "shadow-utils", "util-linux", "passwd", "rootfiles", "vim-minimal", "grub2-tools", "grubby", ] if uefi: packages += ["grub2-efi-x64", "shim-x64", "efibootmgr"] else: packages += ["grub2-pc"] # Offline installation logic possible_repos = [ "/run/install/repo", "/run/install/source", "/mnt/install/repo", "/run/initramfs/live", "/run/initramfs/isoscan", ] iso_repo = None for path in possible_repos: if os.path.exists(os.path.join(path, "repodata")): iso_repo = path break elif os.path.exists(os.path.join(path, "Packages")): iso_repo = path break # Try searching in /run/media if not found if not iso_repo and os.path.exists("/run/media"): try: for user in os.listdir("/run/media"): user_path = os.path.join("/run/media", user) if os.path.isdir(user_path): for label in os.listdir(user_path): label_path = os.path.join(user_path, label) if os.path.exists(os.path.join(label_path, "repodata")): iso_repo = label_path break if iso_repo: break except Exception: pass dnf_args = [] if iso_repo: logger.info(f"Found ISO repository at {iso_repo}. Using strictly offline mode.") dnf_args = [ "--disablerepo=*", f"--repofrompath=iridium-iso,{iso_repo}", "--enablerepo=iridium-iso", "--cacheonly", ] else: logger.warning("ISO repository not found. DNF might try to use network.") dnf_args = [] cmd = [ "dnf", "install", "-y", f"--installroot={mount_root}", f"--releasever={releasever}", "--setopt=install_weak_deps=False", "--nodocs", ] if not iso_repo: cmd.append("--use-host-config") cmd += dnf_args + packages with mount_pseudo_fs(mount_root): run_command(cmd) logger.info("Base system installation complete.") log_os_install("INSTALL", "complete", f"Installed to {mount_root}") def configure_system(mount_root, partition_info, user_info=None, disk_device=None): """ Basic configuration: fstab, grub2, and user creation. """ logger.info("Configuring system...") log_os_install("CONFIGURE", "start", f"Configuring system in {mount_root}") uefi = is_uefi() # 1. Generate fstab def get_uuid(dev): res = run_command(["blkid", "-s", "UUID", "-o", "value", dev]) return res.stdout.strip() root_uuid = get_uuid(partition_info["root"]) fstab_lines = [f"UUID={root_uuid} / ext4 defaults 1 1"] if uefi and partition_info.get("efi"): efi_uuid = get_uuid(partition_info["efi"]) fstab_lines.append(f"UUID={efi_uuid} /boot/efi vfat defaults 0 2") if partition_info.get("swap"): swap_uuid = get_uuid(partition_info["swap"]) fstab_lines.append(f"UUID={swap_uuid} none swap defaults 0 0") os.makedirs(os.path.join(mount_root, "etc"), exist_ok=True) with open(os.path.join(mount_root, "etc/fstab"), "w") as f: f.write("\n".join(fstab_lines) + "\n") with mount_pseudo_fs(mount_root): # 2. Configure User if user_info: logger.info(f"Creating user {user_info['username']}...") run_command(["chroot", mount_root, "useradd", "-m", "-G", "wheel", user_info["username"]]) # Set hostname with open(os.path.join(mount_root, "etc/hostname"), "w") as f: f.write(user_info["hostname"] + "\n") # Set passwords try: res = subprocess.run( ["openssl", "passwd", "-6", user_info["password"]], capture_output=True, text=True, check=True ) hashed_pass = res.stdout.strip() run_command(["chroot", mount_root, "usermod", "-p", hashed_pass, user_info["username"]]) run_command(["chroot", mount_root, "usermod", "-p", hashed_pass, "root"]) except Exception as e: logger.error(f"Failed to set passwords: {e}") # 3. Configure GRUB2 logger.info("Configuring GRUB2...") # Ensure /etc/default/grub exists grub_default = os.path.join(mount_root, "etc/default/grub") if not os.path.exists(grub_default): with open(grub_default, "w") as f: f.write('GRUB_TIMEOUT=5\nGRUB_DISTRIBUTOR="$(sed \'s, release .*$,,g\' /etc/system-release)"\nGRUB_DEFAULT=saved\nGRUB_DISABLE_SUBMENU=true\nGRUB_TERMINAL_OUTPUT="console"\nGRUB_CMDLINE_LINUX="rhgb quiet"\nGRUB_DISABLE_RECOVERY="true"\nGRUB_ENABLE_BLSCFG=true\n') if uefi: run_command(["chroot", mount_root, "grub2-install", "--target=x86_64-efi", "--efi-directory=/boot/efi", "--bootloader-id=iridium", "--recheck"]) run_command(["chroot", mount_root, "grub2-mkconfig", "-o", "/boot/efi/EFI/iridium/grub.cfg"]) # Fedora compatibility link fedora_dir = os.path.join(mount_root, "boot/efi/EFI/fedora") os.makedirs(fedora_dir, exist_ok=True) with open(os.path.join(fedora_dir, "grub.cfg"), "w") as f: f.write(f"configfile /EFI/iridium/grub.cfg\n") else: if not disk_device: # Try to guess disk device from root partition disk_device = partition_info["root"].rstrip("0123456789") if disk_device.endswith("p"): disk_device = disk_device[:-1] logger.info(f"Installing GRUB to {disk_device} (BIOS)") run_command(["chroot", mount_root, "grub2-install", "--target=i386-pc", disk_device]) run_command(["chroot", mount_root, "grub2-mkconfig", "-o", "/boot/grub2/grub.cfg"]) run_command(["sync"]) logger.info("System configuration complete.") log_os_install("CONFIGURE", "complete", "systemd-boot and user configured successfully")