Files
scripts/server-setup.py

1437 lines
50 KiB
Python

#!/usr/bin/env python3
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
# SPDX-License-Identifier: MIT
"""Idempotent server setup for Fedora Server, CentOS Stream 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-epel # skip EPEL setup on CentOS
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.4.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",
"centos": "CentOS Stream",
"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
def ensure_epel(dry_run: bool = False) -> dict[str, Any]:
"""Ensure the EPEL repository is enabled on CentOS Stream.
EPEL (Extra Packages for Enterprise Linux) is required for several
packages on CentOS. This function is idempotent — if EPEL is already
enabled it returns immediately.
"""
info: dict[str, Any] = {
"installed": False,
"already_enabled": False,
"failed": False,
}
_status("Checking EPEL repository")
# Quick check: is epel-release already installed and repo enabled?
repo_check = run(["dnf", "repolist", "--enabled"], timeout=60)
if "epel" in repo_check.stdout.lower():
_status_done("already enabled")
info["already_enabled"] = True
return info
if _rpm_installed("epel-release"):
_status_done("package installed but repo disabled")
else:
_status_done("not installed")
if dry_run:
_info("Would enable CRB repository")
_info("Would install: epel-release")
info["installed"] = True
return info
# Enable CodeReady Linux Builder (CRB) — some EPEL packages depend on it.
# This repo ships with CentOS but is disabled by default. Non-fatal if
# it's already enabled or not available.
_status("Enabling CRB repository")
crb_result = run(["dnf", "config-manager", "--set-enabled", "crb"], timeout=60)
if crb_result.returncode == 0:
_status_done()
else:
_status_done("not available (non-fatal)")
_status("Installing epel-release")
if _dnf_install(["epel-release"]):
_status_done()
_ok("EPEL repository enabled")
info["installed"] = True
else:
_status_done("failed")
_fail("Failed to install epel-release")
info["failed"] = True
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 in ("fedora", "centos"):
# --- Install dnf-automatic (dnf4 name; provided virtually on dnf5) ---
_status("Checking dnf-automatic (Fedora / CentOS Stream)")
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, CentOS Stream 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-epel",
action="store_true",
help="skip EPEL repository setup on CentOS",
)
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. EPEL repository (CentOS Stream only)
if os_id == "centos":
print(f"\n{BOLD}── EPEL Repository ──{RESET}", file=sys.stderr)
if not args.skip_epel:
results["epel"] = ensure_epel(dry_run=args.dry_run)
if results["epel"].get("failed"):
warnings.append(
"EPEL repository setup failed — some packages may be unavailable"
)
else:
_info("EPEL repository: skipped (--skip-epel)")
# 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()