#!/usr/bin/env python3 # Copyright (c) 2026 Petr Balvín (https://petrbalvin.org) # SPDX-License-Identifier: MIT """Idempotent server setup for Fedora Server and openEuler. No external dependencies — stdlib and system tools only. Every operation checks current state before acting; running the script repeatedly produces the same result with no errors and no duplicate work. Usage: server-setup.py # full setup server-setup.py --dry-run # preview without changes server-setup.py --skip-update # skip system update server-setup.py --skip-packages # skip package installation server-setup.py --skip-firewall # skip firewall setup server-setup.py --skip-selinux # skip SELinux server-setup.py --skip-podman # skip Podman server-setup.py --skip-auto-updates # skip automatic updates server-setup.py --version Exit status is 0 when every enabled section succeeds, 1 otherwise. """ from __future__ import annotations import argparse import os import pwd import subprocess import sys import time from typing import Any BOLD = "\033[1m" RED = "\033[31m" GREEN = "\033[32m" YELLOW = "\033[33m" DIM = "\033[2m" RESET = "\033[0m" VERSION = "1.2.0" # Timeout in seconds for dnf operations — metadata downloads and package # transactions on a fresh or slow system routinely exceed the 60 s default # used for cheap probes. DNF_TIMEOUT = 1800 # Map os-release ID to display name — only these two are supported. SUPPORTED_OS: dict[str, str] = { "fedora": "Fedora", "openeuler": "openEuler", } # Base packages installed on both Fedora and openEuler. BASE_PACKAGES: list[str] = [ "nano", "curl", "wget", "htop", "tmux", "rsync", "python3", "python3-devel", "nginx", "openssl", "jq", "fastfetch", ] # Services to open in firewalld by default — ssh is mandatory, losing it # means locking out remote administration. FIREWALL_SERVICES: list[str] = ["ssh"] # Opened in firewalld only when nginx is installed (it is in BASE_PACKAGES). NGINX_FIREWALL_SERVICES: list[str] = ["http", "https"] # dnf-automatic configuration file. dnf4 ships it with defaults; dnf5 reads # host overrides from this path (defaults: /usr/share/dnf5/dnf5-plugins/). DNF_AUTOMATIC_CONF = "/etc/dnf/automatic.conf" # Timer units to probe, in order of preference. dnf4 ships the -install # variant; dnf5 (Fedora 41+) ships only the single dnf5-automatic.timer. AUTO_UPDATE_TIMER_CANDIDATES: tuple[str, ...] = ( "dnf-automatic-install.timer", "dnf5-automatic.timer", "dnf-automatic.timer", ) # --------------------------------------------------------------------------- # Progress helpers — write to stderr so stdout stays clean for machine output # --------------------------------------------------------------------------- def _status(msg: str) -> None: """Print a progress message to stderr — stays on the same line until ``_status_done``.""" print(f" {msg}...", end="", flush=True, file=sys.stderr) def _status_done(msg: str = "done") -> None: """Complete the current progress line on stderr.""" print(f" {msg}", file=sys.stderr) def _info(msg: str) -> None: """Print an informational line to stderr.""" print(f" {DIM}{msg}{RESET}", file=sys.stderr) def _warn(msg: str) -> None: """Print a warning to stderr.""" print(f" {YELLOW}⚠ {msg}{RESET}", file=sys.stderr) def _ok(msg: str) -> None: """Print a success message to stderr.""" print(f" {GREEN}✓ {msg}{RESET}", file=sys.stderr) def _fail(msg: str) -> None: """Print a failure message to stderr.""" print(f" {RED}✗ {msg}{RESET}", file=sys.stderr) # --------------------------------------------------------------------------- # Low-level command runner # --------------------------------------------------------------------------- def run( cmd: list[str], timeout: int = 60, check: bool = False, env: dict[str, str] | None = None, ) -> subprocess.CompletedProcess[str]: """Execute a command and return the CompletedProcess. Sets LC_ALL=C for reliable English-language output parsing. Raises subprocess.CalledProcessError when *check* is True and the command exits non-zero. A missing executable or an expired timeout is reported and returned as a synthetic failure (rc 127 / 124) instead of raising, so callers always get a CompletedProcess to inspect. """ merged_env = {**os.environ, "LC_ALL": "C"} if env: merged_env.update(env) try: return subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, env=merged_env, check=check, ) except FileNotFoundError: _fail(f"Command not found: {cmd[0]}") return subprocess.CompletedProcess(cmd, 127, "", f"command not found: {cmd[0]}") except subprocess.TimeoutExpired: _fail(f"Command timed out after {timeout}s: {' '.join(cmd)}") return subprocess.CompletedProcess(cmd, 124, "", f"timed out after {timeout}s") # --------------------------------------------------------------------------- # OS detection # --------------------------------------------------------------------------- def parse_os_release() -> dict[str, str]: """Parse /etc/os-release into a dict of KEY=VALUE pairs.""" result: dict[str, str] = {} try: with open("/etc/os-release", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#"): continue if "=" not in line: continue key, _, value = line.partition("=") # Strip surrounding quotes if present. value = value.strip().strip('"').strip("'") result[key] = value except OSError: pass return result def detect_os() -> tuple[str, str, str]: """Detect the operating system. Returns a tuple of (os_id, display_name, version_id). Exits with an error message when the OS is not supported. """ release = parse_os_release() os_id = release.get("ID", "").lower() version_id = release.get("VERSION_ID", "unknown") if os_id not in SUPPORTED_OS: print( f"{RED}{BOLD}Error:{RESET} Unsupported operating system: " f"'{release.get('ID', 'unknown')}' (detected from /etc/os-release).", file=sys.stderr, ) print( f" Supported systems: {', '.join(SUPPORTED_OS.values())}", file=sys.stderr, ) sys.exit(1) return os_id, SUPPORTED_OS[os_id], version_id # --------------------------------------------------------------------------- # Root check # --------------------------------------------------------------------------- def check_root() -> None: """Exit with a clear message if not running as root (UID 0).""" if os.geteuid() != 0: print( f"{RED}{BOLD}Error:{RESET} This script must be run as root (UID 0).", file=sys.stderr, ) print( f" Current UID: {os.geteuid()}. Try: sudo python3 server-setup.py", file=sys.stderr, ) sys.exit(1) # --------------------------------------------------------------------------- # Section helpers # --------------------------------------------------------------------------- def _rpm_installed(pkg: str) -> bool: """Return True if the RPM package is already installed.""" result = run(["rpm", "-q", pkg]) return result.returncode == 0 def _dnf_install(packages: list[str]) -> bool: """Install packages via dnf in a single transaction. Returns True on success.""" if not packages: return True result = run(["dnf", "install", "-y", *packages], timeout=DNF_TIMEOUT) return result.returncode == 0 # --------------------------------------------------------------------------- # 1. System update # --------------------------------------------------------------------------- def system_update(dry_run: bool = False) -> dict[str, Any]: """Update the system via dnf. Returns a dict with summary info: {'updated': bool, 'skipped': bool}. """ info: dict[str, Any] = { "updated": False, "skipped": False, "failed": False, "reboot_required": False, } _status("Checking for system updates") check_result = run(["dnf", "check-update"], timeout=DNF_TIMEOUT) # dnf check-update: 0 = no updates, 100 = updates available, 1 = error. if check_result.returncode == 0: _status_done("already up to date") info["skipped"] = True return info if check_result.returncode != 100: _status_done("check failed") _fail("dnf check-update failed — cannot determine update state") info["failed"] = True return info _status_done("updates available") if dry_run: _info("Would run: dnf update -y") return info _status("Applying system updates") update_result = run(["dnf", "update", "-y"], timeout=DNF_TIMEOUT) if update_result.returncode != 0: _status_done("failed") _fail("System update returned non-zero exit code") info["failed"] = True return info _status_done() info["updated"] = True # dnf needs-restarting -r: rc 1 plus the message means a reboot is # required. The plugin (dnf-utils) may be absent; anything unrecognised # is treated as "unknown" rather than a false positive. reboot_check = run(["dnf", "needs-restarting", "-r"]) if reboot_check.returncode == 1 and "Reboot is required" in reboot_check.stdout: info["reboot_required"] = True _warn("Reboot required to fully apply updates") return info # --------------------------------------------------------------------------- # 2. Base packages # --------------------------------------------------------------------------- def install_packages(dry_run: bool = False) -> dict[str, Any]: """Install base packages, skipping any that are already present. Returns a dict with 'installed' (list), 'skipped' (list), 'failed' (list). """ installed: list[str] = [] skipped: list[str] = [] failed: list[str] = [] to_install: list[str] = [] _status("Checking base packages") for pkg in BASE_PACKAGES: if _rpm_installed(pkg): skipped.append(pkg) else: to_install.append(pkg) already = len(skipped) missing = len(to_install) total = len(BASE_PACKAGES) _status_done(f"{already}/{total} already installed") if not to_install: _ok("All base packages already present") return {"installed": installed, "skipped": skipped, "failed": failed} _info(f"Installing {missing} package(s): {' '.join(to_install)}") if dry_run: for pkg in to_install: _info(f" Would install: {pkg}") installed.append(pkg) return {"installed": installed, "skipped": skipped, "failed": failed} # Single transaction first — faster and atomic. If it fails (e.g. one # package is unavailable on this distro), fall back to per-package # installs so the rest still succeed. _status(f"Installing {missing} package(s) in one transaction") if _dnf_install(to_install): _status_done() installed.extend(to_install) return {"installed": installed, "skipped": skipped, "failed": failed} _status_done("transaction failed — falling back to per-package") for pkg in to_install: _status(f"Installing {pkg}") if _dnf_install([pkg]): _status_done() installed.append(pkg) else: _status_done("failed") _fail(f"Failed to install {pkg}") failed.append(pkg) return {"installed": installed, "skipped": skipped, "failed": failed} # --------------------------------------------------------------------------- # 3. Firewall # --------------------------------------------------------------------------- def setup_firewall(dry_run: bool = False) -> dict[str, Any]: """Set up firewalld with default rules. SSH is written to the permanent configuration and verified BEFORE the service is enabled/started (via the offline client, which needs no running daemon), so a misconfiguration can never lock out a remote admin. Returns a dict with summary of what was configured. """ info: dict[str, Any] = { "installed": False, "enabled": False, "zone_set": False, "services_added": [], "services_skipped": [], "failed": False, } # --- Install firewalld if missing --- _status("Checking firewalld") firewalld_present = _rpm_installed("firewalld") if not firewalld_present: _status_done("not installed") if dry_run: _info("Would install: firewalld") info["installed"] = True else: if _dnf_install(["firewalld"]): _ok("Installed firewalld") info["installed"] = True firewalld_present = True else: _fail("Failed to install firewalld") info["failed"] = True return info else: _status_done("installed") info["installed"] = True # ssh is mandatory; http/https only when nginx is present. services = list(FIREWALL_SERVICES) if _rpm_installed("nginx"): services.extend(NGINX_FIREWALL_SERVICES) # --- Permanent rules first, via the offline client (no daemon needed) --- permanent_changed = False for service in services: _status(f"Checking firewall service '{service}'") if not firewalld_present: # Dry run on a host without firewalld: nothing to query yet. _status_done("would add") info["services_added"].append(service) permanent_changed = True continue query = run(["firewall-offline-cmd", "--zone=public", f"--query-service={service}"]) if query.returncode == 0: _status_done("already present") info["services_skipped"].append(service) continue if dry_run: _status_done("would add") info["services_added"].append(service) permanent_changed = True continue add = run(["firewall-offline-cmd", "--zone=public", f"--add-service={service}"]) if add.returncode != 0: _status_done("failed") _fail(f"Failed to add service '{service}' to zone 'public'") info["failed"] = True if service == "ssh": _warn("Aborting before firewalld is enabled to prevent lockout") return info continue _status_done("added") info["services_added"].append(service) permanent_changed = True # --- Lockout gate: never enable firewalld without confirmed SSH access --- if not dry_run: verify = run(["firewall-offline-cmd", "--zone=public", "--query-service=ssh"]) if verify.returncode != 0: _fail("Cannot confirm that SSH is allowed in zone 'public'") _warn( "Aborting firewall setup BEFORE enabling firewalld — remote access would be at risk" ) info["failed"] = True return info # --- Enable and start (only reached with SSH confirmed) --- was_active = run(["systemctl", "is-active", "firewalld.service"]).returncode == 0 _status("Enabling firewalld service") enabled = run(["systemctl", "is-enabled", "firewalld.service"]) if enabled.returncode != 0: if dry_run: _status_done("would enable") info["enabled"] = True else: enable = run(["systemctl", "enable", "firewalld.service"]) if enable.returncode != 0: _status_done("failed") _fail("Failed to enable firewalld.service") info["failed"] = True return info _status_done("enabled") info["enabled"] = True else: _status_done("already enabled") info["enabled"] = True _status("Starting firewalld service") if not was_active: if dry_run: _status_done("would start") else: start = run(["systemctl", "start", "firewalld.service"]) if start.returncode != 0: _status_done("failed") _fail("Failed to start firewalld.service") info["failed"] = True return info _status_done("started") else: _status_done("already running") # --- Default zone (applies to runtime immediately) --- _status("Checking default zone") zone_result = run(["firewall-offline-cmd", "--get-default-zone"]) current_zone = zone_result.stdout.strip() if current_zone != "public": if dry_run: _status_done(f"would change from '{current_zone or '?'}' to 'public'") info["zone_set"] = True else: set_zone = run(["firewall-cmd", "--set-default-zone=public"]) if set_zone.returncode != 0: _status_done("failed") _fail("Failed to set default zone to 'public'") info["failed"] = True return info _status_done(f"changed to 'public' (was '{current_zone}')") info["zone_set"] = True else: _status_done("already 'public'") # --- Reload only when a running daemon has drifted from permanent --- if permanent_changed and was_active: if dry_run: _info("Would run: firewall-cmd --reload") else: _status("Reloading firewall") reload_result = run(["firewall-cmd", "--reload"]) if reload_result.returncode != 0: _status_done("failed") _fail("Failed to reload firewalld") info["failed"] = True return info _status_done() return info # --------------------------------------------------------------------------- # 4. SELinux # --------------------------------------------------------------------------- def setup_selinux(dry_run: bool = False) -> dict[str, Any]: """Ensure SELinux is enforcing and policycoreutils-python-utils is installed. A Disabled -> enforcing transition cannot be performed at runtime (setenforce fails when the kernel has SELinux off): only the config is switched and /.autorelabel is created so the next boot relabels the filesystem before entering enforcing mode. Returns a dict with summary of changes. """ info: dict[str, Any] = { "mode_changed": False, "config_changed": False, "utils_installed": False, "current_mode": "unknown", "reboot_required": False, "failed": False, } # --- Check current mode --- _status("Checking SELinux mode") mode_result = run(["getenforce"]) if mode_result.returncode != 0: _status_done("failed") _fail("Cannot determine SELinux mode (getenforce failed or is missing)") _info("On systems without SELinux, re-run with --skip-selinux") info["failed"] = True return info current_mode = mode_result.stdout.strip() info["current_mode"] = current_mode _status_done(current_mode) if current_mode == "Permissive": if dry_run: _info("Would set SELinux to enforcing mode") info["mode_changed"] = True else: _status("Setting SELinux to enforcing") enforce = run(["setenforce", "1"]) if enforce.returncode != 0: _status_done("failed") _fail("setenforce 1 failed") info["failed"] = True else: _status_done() info["mode_changed"] = True elif current_mode == "Disabled": # setenforce cannot work here — config + autorelabel below instead. info["reboot_required"] = True # --- Ensure SELINUX=enforcing in config --- _status("Checking /etc/selinux/config") config_changed = False try: with open("/etc/selinux/config", encoding="utf-8") as fh: config_content = fh.read() has_enforcing = False lines = config_content.splitlines() for line in lines: stripped = line.strip() if stripped.startswith("SELINUX=") and not stripped.startswith("#"): value = stripped.split("=", 1)[1].strip() if value.lower() == "enforcing": has_enforcing = True break if not has_enforcing: if dry_run: _status_done("would update to SELINUX=enforcing") else: new_lines: list[str] = [] found = False for line in lines: stripped = line.strip() if stripped.startswith("SELINUX=") and not stripped.startswith("#"): new_lines.append("SELINUX=enforcing") found = True else: new_lines.append(line) if not found: new_lines.append("SELINUX=enforcing") with open("/etc/selinux/config", "w", encoding="utf-8") as fh: fh.write("\n".join(new_lines) + "\n") _status_done("updated to enforcing") config_changed = True else: _status_done("already enforcing") except OSError as exc: _status_done("error") _fail(f"Cannot read/write /etc/selinux/config: {exc}") info["failed"] = True info["config_changed"] = config_changed # --- Disabled -> enforcing requires a reboot with filesystem relabel --- if current_mode == "Disabled": if dry_run: _info("Would create /.autorelabel (relabel on next boot)") else: _status("Scheduling filesystem relabel on next boot") try: if not os.path.exists("/.autorelabel"): with open("/.autorelabel", "w", encoding="utf-8"): pass _status_done("/.autorelabel created") except OSError as exc: _status_done("failed") _fail(f"Cannot create /.autorelabel: {exc}") info["failed"] = True _warn( "SELinux is Disabled — enforcing mode requires a REBOOT; " "the filesystem will be relabelled on next boot" ) # --- Install policycoreutils-python-utils --- _status("Checking policycoreutils-python-utils") if not _rpm_installed("policycoreutils-python-utils"): _status_done("not installed") if dry_run: _info("Would install: policycoreutils-python-utils") info["utils_installed"] = True else: if _dnf_install(["policycoreutils-python-utils"]): _ok("Installed policycoreutils-python-utils") info["utils_installed"] = True else: _fail("Failed to install policycoreutils-python-utils") info["failed"] = True else: _status_done("already installed") return info # --------------------------------------------------------------------------- # 5. Podman # --------------------------------------------------------------------------- def _subid_entry_exists(path: str, user: str) -> bool: """Return True if *user* already has an entry in a subordinate ID file.""" try: with open(path, encoding="utf-8") as fh: for line in fh: if line.split(":", 1)[0] == user: return True except OSError: pass return False def ensure_rootless_subids(dry_run: bool = False) -> None: """Allocate subordinate UID/GID ranges for the sudo-invoking user. Rootless Podman needs a 65536-wide subuid/subgid range per user. The range is derived deterministically from the user's UID so re-runs are stable, and existing entries are never modified. Failures are warnings only — rootless use is optional. """ user = os.environ.get("SUDO_USER", "") if not user or user == "root": _info("No non-root invoking user — skipping rootless subuid/subgid setup") return try: uid = pwd.getpwnam(user).pw_uid except KeyError: _warn(f"Cannot look up user '{user}' — skipping subuid/subgid setup") return start = 100000 + max(uid - 1000, 0) * 65536 id_range = f"{start}-{start + 65535}" for path, flag in (("/etc/subuid", "--add-subuids"), ("/etc/subgid", "--add-subgids")): if _subid_entry_exists(path, user): _info(f"{path}: entry for '{user}' already present") continue if dry_run: _info(f"Would run: usermod {flag} {id_range} {user}") continue _status(f"Allocating subordinate IDs for '{user}' in {path}") result = run(["usermod", flag, id_range, user]) if result.returncode == 0: _status_done(id_range) else: _status_done("failed") _warn(f"usermod {flag} failed — rootless Podman may not work for '{user}'") def setup_podman(dry_run: bool = False) -> dict[str, Any]: """Install Podman if not present and prepare rootless use. Returns a dict with 'installed' (bool), 'version' (str), 'failed' (bool). """ info: dict[str, Any] = {"installed": False, "version": "", "failed": False} _status("Checking Podman") already = _rpm_installed("podman") if already: # Verify it actually works. ver = run(["podman", "--version"]) if ver.returncode == 0: info["version"] = ver.stdout.strip() _status_done(info["version"]) else: _status_done("installed but not working") _warn("podman package is installed but 'podman --version' failed") info["installed"] = True ensure_rootless_subids(dry_run) return info _status_done("not installed") if dry_run: _info("Would install: podman") ensure_rootless_subids(dry_run) return info _status("Installing Podman") if _dnf_install(["podman"]): ver = run(["podman", "--version"]) info["version"] = ver.stdout.strip() _status_done(info["version"]) info["installed"] = True else: _status_done("failed") _fail("Failed to install Podman") info["failed"] = True return info ensure_rootless_subids(dry_run) return info # --------------------------------------------------------------------------- # 6. Automatic updates (dnf-automatic) # --------------------------------------------------------------------------- def _render_dnf_automatic_conf(lines: list[str], desired: dict[str, str]) -> tuple[list[str], bool]: """Apply *desired* keys to the [commands] section of automatic.conf lines. Returns (new_lines, changed). Existing keys are rewritten only when the value actually differs; missing keys are appended at the end of the [commands] section; a missing [commands] section is created. Keys in other sections and comment lines are left untouched. """ new_lines: list[str] = [] seen: set[str] = set() in_commands = False found_commands = False changed = False def flush_missing() -> None: nonlocal changed for key, val in desired.items(): if key not in seen: new_lines.append(f"{key} = {val}\n") changed = True for line in lines: stripped = line.strip() if stripped.startswith("[") and stripped.endswith("]"): if in_commands: flush_missing() seen.clear() in_commands = stripped.lower() == "[commands]" found_commands = found_commands or in_commands new_lines.append(line) continue if in_commands and "=" in stripped and not stripped.startswith("#"): key, _, raw_value = stripped.partition("=") key = key.strip() if key in desired: seen.add(key) if raw_value.strip().lower() != desired[key].lower(): new_lines.append(f"{key} = {desired[key]}\n") changed = True continue new_lines.append(line) if in_commands: flush_missing() if not found_commands: if new_lines and new_lines[-1].strip(): new_lines.append("\n") new_lines.append("[commands]\n") flush_missing() return new_lines, changed def setup_auto_updates(os_id: str, dry_run: bool = False) -> dict[str, Any]: """Configure and enable automatic system updates. Fedora: dnf-automatic (security updates via daily timer). The package, config path and timer unit names differ between dnf4 and dnf5 — all are probed at runtime. openEuler: dnf-hotpatch-plugin + custom systemd service/timer applying all updates plus kernel hot patches. Returns a dict with summary of what was configured. """ info: dict[str, Any] = { "installed": False, "configured": False, "timer_enabled": False, "method": os_id, "failed": False, } if os_id == "fedora": # --- Install dnf-automatic (dnf4 name; provided virtually on dnf5) --- _status("Checking dnf-automatic (Fedora)") pkg_present = _rpm_installed("dnf-automatic") or _rpm_installed("dnf5-plugin-automatic") if not pkg_present: _status_done("not installed") if dry_run: _info("Would install: dnf-automatic") info["installed"] = True else: if _dnf_install(["dnf-automatic"]) or _dnf_install(["dnf5-plugin-automatic"]): _ok("Installed dnf-automatic") info["installed"] = True pkg_present = True else: _fail("Failed to install dnf-automatic") info["failed"] = True return info else: _status_done("installed") info["installed"] = True # --- Configure /etc/dnf/automatic.conf --- _status(f"Configuring {DNF_AUTOMATIC_CONF}") desired = { "apply_updates": "yes", "download_updates": "yes", "upgrade_type": "security", } try: with open(DNF_AUTOMATIC_CONF, encoding="utf-8") as fh: config_lines = fh.readlines() except FileNotFoundError: # dnf5: the host override file is created from scratch. config_lines = [] except OSError as exc: _status_done("cannot read config") _fail(f"Cannot read {DNF_AUTOMATIC_CONF}: {exc}") info["failed"] = True return info rendered, changed = _render_dnf_automatic_conf(config_lines, desired) if not changed: _status_done("already configured") info["configured"] = True elif dry_run: _status_done("would update") info["configured"] = True else: try: with open(DNF_AUTOMATIC_CONF, "w", encoding="utf-8") as fh: fh.writelines(rendered) except OSError as exc: _status_done("failed") _fail(f"Cannot write {DNF_AUTOMATIC_CONF}: {exc}") info["failed"] = True return info _status_done("updated") info["configured"] = True # --- Pick the available timer unit (names differ between dnf4/dnf5) --- _status("Detecting automatic update timer") timer_name = "" if pkg_present: for candidate in AUTO_UPDATE_TIMER_CANDIDATES: if run(["systemctl", "cat", candidate]).returncode == 0: timer_name = candidate break if not timer_name: if dry_run and not pkg_present: _status_done("would detect after installation") _info("Would enable and start the automatic update timer") info["timer_enabled"] = True return info _status_done("none found") _fail( "No automatic update timer found (tried: " + ", ".join(AUTO_UPDATE_TIMER_CANDIDATES) + ")" ) info["failed"] = True return info _status_done(timer_name) # --- Enable and start the timer --- _status(f"Checking {timer_name}") timer_enabled = run(["systemctl", "is-enabled", timer_name]) if timer_enabled.returncode != 0: if dry_run: _status_done("would enable") info["timer_enabled"] = True else: enable = run(["systemctl", "enable", timer_name]) if enable.returncode != 0: _status_done("failed") _fail(f"Failed to enable {timer_name}") info["failed"] = True return info _status_done("enabled") info["timer_enabled"] = True else: _status_done("already enabled") info["timer_enabled"] = True _status(f"Starting {timer_name}") timer_active = run(["systemctl", "is-active", timer_name]) if timer_active.returncode != 0: if dry_run: _status_done("would start") else: start = run(["systemctl", "start", timer_name]) if start.returncode != 0: _status_done("failed") _fail(f"Failed to start {timer_name}") info["failed"] = True return info _status_done("started") else: _status_done("already running") elif os_id == "openeuler": # --- Install dnf-hotpatch-plugin --- _status("Checking dnf-hotpatch-plugin (openEuler)") if not _rpm_installed("dnf-hotpatch-plugin"): _status_done("not installed") if dry_run: _info("Would install: dnf-hotpatch-plugin") info["installed"] = True else: if _dnf_install(["dnf-hotpatch-plugin"]): _ok("Installed dnf-hotpatch-plugin") info["installed"] = True else: _fail("Failed to install dnf-hotpatch-plugin") info["failed"] = True return info else: _status_done("installed") info["installed"] = True # --- Create or refresh auto-update.service / auto-update.timer --- # Policy: openEuler applies ALL available updates (the Fedora # counterpart applies security-only updates via dnf-automatic). service_path = "/etc/systemd/system/auto-update.service" service_content = """\ [Unit] Description=Automatic system updates After=network-online.target Wants=network-online.target [Service] Type=oneshot # Apply all updates, then activate any kernel hot patches. The leading # '-' keeps the unit successful when no hot patches are available. ExecStart=/usr/bin/dnf update -y ExecStart=-/usr/bin/dnf hotupgrade -y """ timer_path = "/etc/systemd/system/auto-update.timer" timer_content = """\ [Unit] Description=Automatic system updates timer [Timer] OnCalendar=daily RandomizedDelaySec=3600 Persistent=true [Install] WantedBy=timers.target """ needs_reload = False for unit_path, unit_content in ( (service_path, service_content), (timer_path, timer_content), ): unit_name = unit_path.rsplit("/", 1)[-1] _status(f"Checking {unit_name} (openEuler)") try: with open(unit_path, encoding="utf-8") as fh: existing: str | None = fh.read() except OSError: existing = None if existing == unit_content: _status_done("up to date") continue if dry_run: _status_done("would create" if existing is None else "would update (drift)") continue try: with open(unit_path, "w", encoding="utf-8") as fh: fh.write(unit_content) except OSError as exc: _status_done("failed") _fail(f"Cannot write {unit_path}: {exc}") info["failed"] = True continue _status_done("created" if existing is None else "updated (content drift)") needs_reload = True info["configured"] = True # Reload systemd if we created or refreshed unit files if needs_reload and not dry_run: _status("Reloading systemd daemon") daemon_reload = run(["systemctl", "daemon-reload"]) if daemon_reload.returncode != 0: _status_done("failed") _fail("systemctl daemon-reload failed") info["failed"] = True return info _status_done("reloaded") # --- Enable auto-update.timer --- _status("Checking auto-update.timer (openEuler)") timer_enabled = run(["systemctl", "is-enabled", "auto-update.timer"]) if timer_enabled.returncode != 0: if dry_run: _status_done("would enable") info["timer_enabled"] = True else: enable = run(["systemctl", "enable", "auto-update.timer"]) if enable.returncode != 0: _status_done("failed") _fail("Failed to enable auto-update.timer") info["failed"] = True return info _status_done("enabled") info["timer_enabled"] = True else: _status_done("already enabled") info["timer_enabled"] = True # --- Start auto-update.timer --- _status("Starting auto-update.timer") timer_active = run(["systemctl", "is-active", "auto-update.timer"]) if timer_active.returncode != 0: if dry_run: _status_done("would start") else: start = run(["systemctl", "start", "auto-update.timer"]) if start.returncode != 0: _status_done("failed") _fail("Failed to start auto-update.timer") info["failed"] = True return info _status_done("started") else: _status_done("already running") # --- Verify the hotpatch plugin works (tri-state: None = unchecked) --- info["hotpatch_verified"] = None _status("Verifying dnf hotpatch plugin (openEuler)") if dry_run: _status_done("skipped (dry run)") else: hotpatch = run( ["dnf", "hot-updateinfo", "list", "cves", "--installed"], timeout=DNF_TIMEOUT, ) if hotpatch.returncode == 0: _status_done("verified") info["hotpatch_verified"] = True else: _status_done("failed") _warn("dnf hot-updateinfo returned non-zero — hotpatch may not be functional") info["hotpatch_verified"] = False return info # --------------------------------------------------------------------------- # Summary # --------------------------------------------------------------------------- def print_summary( os_display: str, version_id: str, results: dict[str, Any], warnings: list[str], failures: list[str], elapsed: float, ) -> None: """Print a final summary of all sections to stderr.""" print(f"\n{BOLD}── Setup Summary ──{RESET}", file=sys.stderr) print(f" OS: {os_display} {version_id}", file=sys.stderr) # System update ru = results.get("update", {}) if ru.get("failed"): _fail("System update failed") elif ru.get("updated"): _ok("System updated") elif ru.get("skipped"): _info("System already up to date") else: _info("System update skipped") if ru.get("reboot_required"): _warn("Reboot required to fully apply updates") # Packages rp = results.get("packages", {}) installed = len(rp.get("installed", [])) skipped = len(rp.get("skipped", [])) failed_pkgs = len(rp.get("failed", [])) _ok(f"Packages: {skipped} already present, {installed} installed") if failed_pkgs: _fail(f"{failed_pkgs} package(s) failed to install") # Firewall rf = results.get("firewall", {}) if rf.get("failed"): _fail("Firewall setup failed") elif rf: svc = len(rf.get("services_added", [])) svc_skip = len(rf.get("services_skipped", [])) _ok(f"Firewall: {svc} service(s) added, {svc_skip} already present") # SELinux rs = results.get("selinux", {}) if rs.get("failed"): _fail("SELinux setup failed") elif rs: _ok(f"SELinux: mode={rs.get('current_mode', '?')}") if rs.get("reboot_required"): _warn("SELinux: reboot required — filesystem relabel scheduled (/.autorelabel)") # Podman rpod = results.get("podman", {}) if rpod.get("failed"): _fail("Podman installation failed") else: pod_ver = rpod.get("version", "") if pod_ver: _ok(f"Podman: {pod_ver}") else: _info("Podman: not installed") # Auto-updates rau = results.get("auto_updates", {}) if rau.get("failed"): _fail("Auto-updates setup failed") elif rau: timer = "enabled" if rau.get("timer_enabled") else "not enabled" _ok(f"Auto-updates: timer {timer}") if rau.get("hotpatch_verified") is not None: hp = "verified" if rau["hotpatch_verified"] else "not verified" _info(f" dnf hotpatch: {hp}") if warnings: print(file=sys.stderr) for w in warnings: _warn(w) print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr) if failures: _fail(f"Completed with failures in: {', '.join(failures)}") else: _ok("All sections completed successfully") print(f" {BOLD}Total time: {elapsed:.1f}s{RESET}", file=sys.stderr) print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr) print(file=sys.stderr) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args(argv: list[str] | None = None) -> argparse.Namespace: """Parse command-line arguments.""" parser = argparse.ArgumentParser( description="Idempotent server setup for Fedora Server and openEuler", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--dry-run", action="store_true", help="Print what would be done without making changes", ) parser.add_argument( "--skip-update", action="store_true", help="Skip system update", ) parser.add_argument( "--skip-packages", action="store_true", help="Skip base package installation", ) parser.add_argument( "--skip-firewall", action="store_true", help="Skip firewall setup", ) parser.add_argument( "--skip-selinux", action="store_true", help="Skip SELinux configuration", ) parser.add_argument( "--skip-podman", action="store_true", help="Skip Podman installation", ) parser.add_argument( "--skip-auto-updates", action="store_true", help="Skip automatic updates configuration", ) parser.add_argument( "--version", action="version", version=f"%(prog)s {VERSION}", ) return parser.parse_args(argv) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: """Entry point — parse args, check root, detect OS, run enabled sections.""" start_time = time.monotonic() warnings: list[str] = [] results: dict[str, Any] = {} # Parse arguments first so --help/--version work for any user on any OS. args = parse_args() check_root() os_id, os_display, version_id = detect_os() # Header print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr) print( f"{BOLD} Server Setup v{VERSION} — {os_display} {version_id}{RESET}", file=sys.stderr, ) print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr) if args.dry_run: print( f"\n {YELLOW}{BOLD}DRY RUN — no changes will be made{RESET}", file=sys.stderr, ) print(file=sys.stderr) # 1. System update print(f"\n{BOLD}── System Update ──{RESET}", file=sys.stderr) if not args.skip_update: results["update"] = system_update(dry_run=args.dry_run) if results["update"].get("reboot_required"): warnings.append("System updates require a reboot to take full effect") else: _info("System update: skipped (--skip-update)") # 2. Base packages print(f"\n{BOLD}── Base Packages ──{RESET}", file=sys.stderr) if not args.skip_packages: results["packages"] = install_packages(dry_run=args.dry_run) if results["packages"].get("failed"): warnings.append( f"Some packages failed to install: {', '.join(results['packages']['failed'])}" ) else: _info("Base packages: skipped (--skip-packages)") # 3. Firewall print(f"\n{BOLD}── Firewall ──{RESET}", file=sys.stderr) if not args.skip_firewall: results["firewall"] = setup_firewall(dry_run=args.dry_run) else: _info("Firewall: skipped (--skip-firewall)") # 4. SELinux print(f"\n{BOLD}── SELinux ──{RESET}", file=sys.stderr) if not args.skip_selinux: results["selinux"] = setup_selinux(dry_run=args.dry_run) else: _info("SELinux: skipped (--skip-selinux)") # 5. Podman print(f"\n{BOLD}── Podman ──{RESET}", file=sys.stderr) if not args.skip_podman: results["podman"] = setup_podman(dry_run=args.dry_run) else: _info("Podman: skipped (--skip-podman)") # 6. Automatic updates print(f"\n{BOLD}── Automatic Updates ──{RESET}", file=sys.stderr) if not args.skip_auto_updates: results["auto_updates"] = setup_auto_updates(os_id, dry_run=args.dry_run) else: _info("Auto-updates: skipped (--skip-auto-updates)") # Any truthy "failed" value (bool flag or non-empty package list) counts. failures = [name for name, res in results.items() if res.get("failed")] elapsed = time.monotonic() - start_time print_summary(os_display, version_id, results, warnings, failures, elapsed) if failures: sys.exit(1) if __name__ == "__main__": main()