Files
bootstrap_dev_env/bootstrap_environment.py
T
2026-05-17 11:21:01 -05:00

1294 lines
48 KiB
Python
Executable File

#!/usr/bin/env python3
"""
Bootstrap packages listed in formatted_packages.txt.
Sections handled:
=== System Packages === — installed via dnf or apt-get
=== Flatpak Packages === — installed via flatpak from Flathub
=== Custom Installed Packages === — downloaded, verified, extracted
Usage:
sudo python3 bootstrap_environment.py [--only system|flatpak|custom] [--no-gui]
"""
import argparse
import datetime
import getpass
import hashlib
import json
import os
import platform
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
SCRIPT_DIR = Path(__file__).parent
PACKAGES_FILE = SCRIPT_DIR / "formatted_packages.txt"
RUN_LOG = SCRIPT_DIR / "bootstrap_run.log"
# ── issue log ─────────────────────────────────────────────────────────────────
_issues: list[str] = []
_issues_lock = threading.Lock()
def _log_issue(level: str, msg: str) -> None:
with _issues_lock:
print(f" [{level}] {msg}")
_issues.append(f"[{level}] {msg}")
def warn(msg: str) -> None:
_log_issue("WARN", msg)
def err(msg: str) -> None:
_log_issue("ERROR", msg)
def write_run_log() -> None:
if not _issues:
print("\nNo issues — log file not written.")
return
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
lines = [f"# Bootstrap run — {timestamp}", ""] + _issues
RUN_LOG.write_text("\n".join(lines) + "\n")
print(f"\n{len(_issues)} issue(s) logged to: {RUN_LOG}")
# ── subprocess helpers ────────────────────────────────────────────────────────
def run(
cmd: list,
*,
as_sudo: bool = False,
check: bool = True,
input: Optional[bytes] = None,
capture_output: bool = False,
cwd: Optional[str] = None,
) -> subprocess.CompletedProcess:
if as_sudo and os.geteuid() != 0:
cmd = ["sudo"] + cmd
print(f" $ {' '.join(str(c) for c in cmd)}")
return subprocess.run(
cmd, check=check, input=input,
capture_output=capture_output, cwd=cwd,
)
def shell(
cmd: str,
*,
check: bool = True,
capture_output: bool = False,
text: bool = False,
) -> subprocess.CompletedProcess:
print(f" $ {cmd}")
return subprocess.run(
cmd, shell=True, check=check,
capture_output=capture_output, text=text,
)
def has_cmd(name: str) -> bool:
return shutil.which(name) is not None
# ── architecture detection ────────────────────────────────────────────────────
# Tokens commonly seen in download URLs / asset names per architecture.
_ARCH_TOKENS: dict[str, tuple[str, ...]] = {
"x86_64": ("x86_64", "amd64", "x64"),
"aarch64": ("aarch64", "arm64"),
}
def detect_arch() -> str:
m = platform.machine().lower()
if m in ("x86_64", "amd64"):
return "x86_64"
if m in ("aarch64", "arm64"):
return "aarch64"
sys.exit(f"Unsupported architecture: {platform.machine()} (supports x86_64, aarch64)")
ARCH = detect_arch()
# Per-arch substitutions used by repo / download URL construction.
_ARCH_GO = {"x86_64": "amd64", "aarch64": "arm64"}
_ARCH_MINIKUBE = {"x86_64": "amd64", "aarch64": "arm64"}
_ARCH_DEB = {"x86_64": "amd64", "aarch64": "arm64"}
def _arch_matches(name: str, arch: str = ARCH) -> bool:
n = name.lower()
return any(tok in n for tok in _ARCH_TOKENS[arch])
def _other_arch() -> str:
return "aarch64" if ARCH == "x86_64" else "x86_64"
def _has_other_arch_token(name: str) -> bool:
n = name.lower()
return any(tok in n for tok in _ARCH_TOKENS[_other_arch()])
# ── sudo prereq ───────────────────────────────────────────────────────────────
def check_sudo() -> None:
if os.geteuid() == 0:
return
if not has_cmd("sudo"):
sys.exit("sudo is required but not installed.")
print("Validating sudo access ...")
result = subprocess.run(["sudo", "-v"], check=False)
if result.returncode != 0:
sys.exit("sudo authentication failed.")
# ── package manager detection ─────────────────────────────────────────────────
def detect_pkg_mgr() -> str:
for mgr in ("dnf", "apt-get"):
if has_cmd(mgr):
return mgr
sys.exit("No supported package manager found (expected dnf or apt-get).")
PKG_MGR = detect_pkg_mgr()
# ── network helpers ───────────────────────────────────────────────────────────
def _download(url: str, dest: Path) -> bool:
"""Stream URL → dest. Returns True on success, False on failure (logged)."""
print(f" Downloading {Path(url).name} ...")
try:
with urllib.request.urlopen(url) as resp, open(dest, "wb") as f:
shutil.copyfileobj(resp, f, length=1 << 20)
except (urllib.error.URLError, OSError) as e:
err(f"Download failed for {url}: {e}")
return False
return True
def _fetch_json(url: str) -> Optional[dict]:
req = urllib.request.Request(url, headers={"Accept": "application/vnd.github+json"})
try:
with urllib.request.urlopen(req) as resp:
return json.load(resp)
except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
err(f"API request failed for {url}: {e}")
return None
# ── installation checks ───────────────────────────────────────────────────────
def is_system_pkg_installed(pkg: str) -> bool:
if PKG_MGR == "dnf":
return subprocess.run(["rpm", "-q", pkg], capture_output=True).returncode == 0
elif PKG_MGR == "apt-get":
result = subprocess.run(
["dpkg-query", "-W", "-f=${Status}", pkg],
capture_output=True, text=True, check=False,
)
return "install ok installed" in result.stdout
return False
def is_flatpak_installed(app_id: str) -> bool:
if not has_cmd("flatpak"):
return False
return subprocess.run(["flatpak", "info", app_id], capture_output=True).returncode == 0
def is_special_pkg_installed(pkg: str) -> bool:
if pkg == "obsidian":
return Path("/usr/local/bin/obsidian").exists()
if pkg == "minikube":
return Path("/usr/local/bin/minikube").exists() or has_cmd("minikube")
if pkg == "bashtop":
return Path("/usr/local/bin/bashtop").exists() or (Path.home() / "bashtop").exists()
if pkg == "pipx":
return has_cmd("pipx")
if pkg == "poetry":
return has_cmd("poetry")
return is_system_pkg_installed(pkg)
# ── package name overrides ────────────────────────────────────────────────────
# Maps distro-specific or unavailable names to their real equivalents.
# None = skip with a warning.
_OVERRIDES: dict[str, dict[str, Optional[list[str]]]] = {
"dnf": {
"build-essential": ["gcc", "gcc-c++", "make"], # Debian meta-package
"rg": ["ripgrep"], # binary name ≠ package name
"docker-compose": None, # conflicts with docker-compose-plugin from Docker CE; v2 covers this
"webcamoid": None, # not in Fedora repos; installed via Flatpak instead
},
"apt-get": {
"ffmpeg-free": ["ffmpeg"], # Fedora-specific name
"rg": ["ripgrep"],
},
}
def resolve_system_pkgs(names: list[str]) -> tuple[list[str], list[str]]:
"""Return (resolved_names, skipped_names) after applying distro overrides."""
overrides = _OVERRIDES.get(PKG_MGR, {})
resolved, skipped = [], []
for pkg in names:
if pkg in overrides:
replacement = overrides[pkg]
if replacement is None:
skipped.append(pkg)
else:
resolved.extend(replacement)
else:
resolved.append(pkg)
return resolved, skipped
# ── repo setup ────────────────────────────────────────────────────────────────
def _repo_file_exists(*paths: str) -> bool:
return any(Path(p).exists() for p in paths)
def _write_dnf_repo(name: str, display_name: str, baseurl: str, gpgkey: str) -> None:
content = (
f"[{name}]\n"
f"name={display_name}\n"
f"baseurl={baseurl}\n"
f"enabled=1\ngpgcheck=1\n"
f"gpgkey={gpgkey}\n"
)
path = f"/etc/yum.repos.d/{name}.repo"
run(["tee", path], as_sudo=True, input=content.encode(),
capture_output=True, check=True)
def setup_docker_repo() -> None:
if PKG_MGR == "dnf":
if _repo_file_exists("/etc/yum.repos.d/docker-ce.repo"):
return
run(["dnf", "config-manager", "addrepo", "--from-repofile",
"https://download.docker.com/linux/fedora/docker-ce.repo"], as_sudo=True)
elif PKG_MGR == "apt-get":
if _repo_file_exists("/etc/apt/sources.list.d/docker.list"):
return
run(["apt-get", "install", "-y", "ca-certificates", "curl", "gnupg"], as_sudo=True)
shell(
"install -m 0755 -d /etc/apt/keyrings && "
"curl -fsSL https://download.docker.com/linux/ubuntu/gpg | "
"sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg && "
"sudo chmod a+r /etc/apt/keyrings/docker.gpg"
)
codename = shell(
". /etc/os-release && echo $VERSION_CODENAME",
capture_output=True, text=True,
).stdout.strip()
deb_arch = _ARCH_DEB[ARCH]
run(
["tee", "/etc/apt/sources.list.d/docker.list"],
as_sudo=True,
input=(
f"deb [arch={deb_arch} signed-by=/etc/apt/keyrings/docker.gpg] "
f"https://download.docker.com/linux/ubuntu {codename} stable\n"
).encode(),
capture_output=True, check=True,
)
run(["apt-get", "update"], as_sudo=True)
def setup_gh_repo() -> None:
if PKG_MGR == "dnf":
if _repo_file_exists("/etc/yum.repos.d/gh-cli.repo"):
return
run(["dnf", "config-manager", "addrepo", "--from-repofile",
"https://cli.github.com/packages/rpm/gh-cli.repo"], as_sudo=True)
elif PKG_MGR == "apt-get":
if _repo_file_exists("/etc/apt/sources.list.d/github-cli.list"):
return
deb_arch = _ARCH_DEB[ARCH]
shell(
"curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | "
"sudo dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && "
"sudo chmod go+r /usr/share/keyrings/githubcli-archive-keyring.gpg && "
f"echo 'deb [arch={deb_arch} signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] "
"https://cli.github.com/packages stable main' | "
"sudo tee /etc/apt/sources.list.d/github-cli.list > /dev/null"
)
run(["apt-get", "update"], as_sudo=True)
def setup_chrome_repo() -> None:
if ARCH != "x86_64":
warn("Google Chrome has no Linux build for this arch — skipping repo")
return
if PKG_MGR == "dnf":
if _repo_file_exists("/etc/yum.repos.d/google-chrome.repo"):
return
_write_dnf_repo(
"google-chrome", "Google Chrome",
"https://dl.google.com/linux/chrome/rpm/stable/x86_64",
"https://dl.google.com/linux/linux_signing_key.pub",
)
elif PKG_MGR == "apt-get":
if _repo_file_exists("/etc/apt/sources.list.d/google-chrome.list"):
return
shell(
"curl -fsSL https://dl.google.com/linux/linux_signing_key.pub | "
"sudo gpg --dearmor -o /etc/apt/keyrings/google-chrome.gpg && "
"echo 'deb [arch=amd64 signed-by=/etc/apt/keyrings/google-chrome.gpg] "
"https://dl.google.com/linux/chrome/deb/ stable main' | "
"sudo tee /etc/apt/sources.list.d/google-chrome.list > /dev/null && "
"sudo apt-get update"
)
def setup_vivaldi_repo() -> None:
if ARCH != "x86_64":
warn("Vivaldi repo on this arch is not supported by this script — skipping")
return
if PKG_MGR == "dnf":
if _repo_file_exists("/etc/yum.repos.d/vivaldi.repo"):
return
_write_dnf_repo(
"vivaldi", "Vivaldi",
"https://repo.vivaldi.com/archive/rpm/x86_64",
"https://repo.vivaldi.com/archive/linux_signing_key.pub",
)
elif PKG_MGR == "apt-get":
if _repo_file_exists("/etc/apt/sources.list.d/vivaldi.list"):
return
shell(
"curl -fsSL https://repo.vivaldi.com/archive/linux_signing_key.pub | "
"sudo gpg --dearmor -o /etc/apt/keyrings/vivaldi.gpg && "
"echo 'deb [arch=amd64 signed-by=/etc/apt/keyrings/vivaldi.gpg] "
"https://repo.vivaldi.com/archive/deb/ stable main' | "
"sudo tee /etc/apt/sources.list.d/vivaldi.list > /dev/null && "
"sudo apt-get update"
)
def setup_temurin_repo() -> None:
# Adoptium uses $basearch in baseurl → multi-arch.
if PKG_MGR == "dnf":
if _repo_file_exists("/etc/yum.repos.d/adoptium.repo"):
return
_write_dnf_repo(
"Adoptium", "Adoptium",
"https://packages.adoptium.net/artifactory/rpm/fedora/$releasever/$basearch",
"https://packages.adoptium.net/artifactory/api/gpg/key/public",
)
elif PKG_MGR == "apt-get":
if _repo_file_exists("/etc/apt/sources.list.d/adoptium.list"):
return
shell(
"wget -qO - https://packages.adoptium.net/artifactory/api/gpg/key/public | "
"sudo gpg --dearmor | sudo tee /etc/apt/keyrings/adoptium.gpg > /dev/null && "
"echo \"deb [signed-by=/etc/apt/keyrings/adoptium.gpg] "
"https://packages.adoptium.net/artifactory/deb/ "
"$(awk -F= '/^VERSION_CODENAME/{print$2}' /etc/os-release) main\" | "
"sudo tee /etc/apt/sources.list.d/adoptium.list > /dev/null && "
"sudo apt-get update"
)
_REPO_GROUPS: list[tuple[set[str], callable]] = [
(
{"containerd.io", "docker-buildx-plugin", "docker-ce-cli",
"docker-ce-rootless-extras", "docker-ce", "docker-compose-plugin"},
setup_docker_repo,
),
({"gh"}, setup_gh_repo),
({"google-chrome-stable"}, setup_chrome_repo),
({"vivaldi-stable"}, setup_vivaldi_repo),
({"temurin-25-jdk"}, setup_temurin_repo),
]
# ── special package installers ────────────────────────────────────────────────
_SPECIAL_PKGS = {"github-desktop", "zoom", "obsidian", "minikube", "bashtop", "pipx", "poetry"}
# GUI apps — skipped when --no-gui is passed (headless environments).
_GUI_SYSTEM_PKGS = {
"github-desktop",
"google-chrome-stable",
"obs-studio",
"obsidian",
"shutter",
"virt-manager",
"vivaldi-stable",
"webcamoid",
"wireshark",
"zoom",
}
def _install_github_desktop(tmp: Path) -> None:
data = _fetch_json("https://api.github.com/repos/shiftkey/desktop/releases/latest")
if data is None:
return
suffix, host_tokens, exclude_tokens = (
(".rpm", _ARCH_TOKENS[ARCH], _ARCH_TOKENS[_other_arch()])
if PKG_MGR == "dnf"
else (".deb", _ARCH_TOKENS[ARCH], _ARCH_TOKENS[_other_arch()])
)
def matches(name: str) -> bool:
n = name.lower()
if not n.endswith(suffix):
return False
if not any(t in n for t in host_tokens):
return False
if any(t in n for t in exclude_tokens if t not in host_tokens):
return False
return True
asset = next((a for a in data["assets"] if matches(a["name"])), None)
if asset is None:
err(f"No GitHub Desktop {suffix} asset found for {ARCH}")
return
dest = tmp / asset["name"]
if not _download(asset["browser_download_url"], dest):
return
installer = "dnf" if PKG_MGR == "dnf" else "apt-get"
run([installer, "install", "-y", str(dest)], as_sudo=True, check=False)
def _install_zoom(tmp: Path) -> None:
if ARCH != "x86_64":
warn("Zoom has no aarch64 Linux client — skipping")
return
if PKG_MGR == "dnf":
dest = tmp / "zoom.rpm"
if not _download("https://zoom.us/client/latest/zoom_x86_64.rpm", dest):
return
run(["dnf", "install", "-y", str(dest)], as_sudo=True, check=False)
elif PKG_MGR == "apt-get":
dest = tmp / "zoom.deb"
if not _download("https://zoom.us/client/latest/zoom_amd64.deb", dest):
return
run(["apt-get", "install", "-y", str(dest)], as_sudo=True, check=False)
def _install_obsidian(tmp: Path) -> None:
data = _fetch_json("https://api.github.com/repos/obsidianmd/obsidian-releases/releases/latest")
if data is None:
return
host_tokens = _ARCH_TOKENS[ARCH]
other_tokens = _ARCH_TOKENS[_other_arch()]
def matches(name: str) -> bool:
n = name.lower()
if not n.endswith(".appimage"):
return False
if not any(t in n for t in host_tokens):
return False
if any(t in n for t in other_tokens if t not in host_tokens):
return False
return True
asset = next((a for a in data["assets"] if matches(a["name"])), None)
if asset is None:
err(f"No Obsidian AppImage found for {ARCH}")
return
dest = tmp / asset["name"]
if not _download(asset["browser_download_url"], dest):
return
install_path = Path("/usr/local/bin/obsidian")
run(["cp", str(dest), str(install_path)], as_sudo=True)
run(["chmod", "755", str(install_path)], as_sudo=True)
print(f" Obsidian AppImage installed at {install_path}")
def _install_minikube(tmp: Path) -> None:
arch_token = _ARCH_MINIKUBE[ARCH]
base_url = f"https://storage.googleapis.com/minikube/releases/latest/minikube-linux-{arch_token}"
dest = tmp / "minikube"
if not _download(base_url, dest):
return
print(" Fetching SHA256 ...")
try:
with urllib.request.urlopen(base_url + ".sha256") as resp:
expected = resp.read().decode().strip().split()[0]
except (urllib.error.URLError, OSError) as e:
err(f"minikube SHA256 fetch failed: {e}")
return
actual = _sha256_of(dest)
if actual != expected:
err(f"minikube SHA256 mismatch: expected {expected}, got {actual}")
return
print(" SHA256 OK")
install_path = Path("/usr/local/bin/minikube")
run(["cp", str(dest), str(install_path)], as_sudo=True)
run(["chmod", "755", str(install_path)], as_sudo=True)
print(f" minikube installed to {install_path}")
def _install_bashtop(_tmp: Path) -> None:
clone_dir = Path.home() / "bashtop"
if clone_dir.exists():
print(f" Updating existing clone at {clone_dir} ...")
if run(["git", "-C", str(clone_dir), "pull"], check=False).returncode != 0:
err("bashtop git pull failed")
return
else:
print(f" Cloning bashtop to {clone_dir} ...")
if run(["git", "clone", "https://github.com/aristocratos/bashtop.git",
str(clone_dir)], check=False).returncode != 0:
err("bashtop git clone failed")
return
if run(["make", "install"], as_sudo=True, cwd=str(clone_dir), check=False).returncode != 0:
err("bashtop 'make install' failed")
return
# Also expose the clone dir on PATH so `bashtop` from source works.
profile_line = f"export PATH=$PATH:{clone_dir}"
profile_script = "/etc/profile.d/bashtop.sh"
run(["bash", "-c",
f"grep -qxF {profile_line!r} {profile_script} 2>/dev/null || "
f"echo {profile_line!r} >> {profile_script}"],
as_sudo=True, check=False)
print(f" bashtop installed. Clone at {clone_dir}, binary at /usr/local/bin/bashtop")
def _install_pipx(_tmp: Path) -> None:
if not has_cmd("python3"):
err("Python 3 is not installed — cannot install pipx")
return
run([PKG_MGR, "install", "-y", "pipx"], as_sudo=True, check=False)
if has_cmd("pipx"):
run(["pipx", "ensurepath"], check=False)
else:
err("pipx command not found after install")
def _install_poetry(_tmp: Path) -> None:
if not has_cmd("pipx"):
err("pipx is not installed — cannot install poetry")
return
run(["pipx", "install", "poetry"], check=False)
def install_special_pkg(pkg: str, tmp: Path) -> None:
if pkg == "github-desktop":
_install_github_desktop(tmp)
elif pkg == "zoom":
_install_zoom(tmp)
elif pkg == "obsidian":
_install_obsidian(tmp)
elif pkg == "minikube":
_install_minikube(tmp)
elif pkg == "bashtop":
_install_bashtop(tmp)
elif pkg == "pipx":
_install_pipx(tmp)
elif pkg == "poetry":
_install_poetry(tmp)
# ── system package installation ───────────────────────────────────────────────
def install_system_packages(to_install_regular: list[str], to_install_special: list[str]) -> None:
print("\n=== System Packages ===")
seen_repos: set[int] = set()
for pkg in to_install_regular:
for idx, (members, setup_fn) in enumerate(_REPO_GROUPS):
if pkg in members and idx not in seen_repos:
print(f" [REPO] Setting up repository for {pkg} ...")
setup_fn()
seen_repos.add(idx)
for pkg in to_install_regular:
result = run([PKG_MGR, "install", "-y", pkg], as_sudo=True, check=False)
if result.returncode != 0:
err(f"System package failed to install: {pkg}")
if to_install_special:
with tempfile.TemporaryDirectory() as tmp:
for pkg in to_install_special:
print(f"\n [SPECIAL] Installing {pkg} ...")
install_special_pkg(pkg, Path(tmp))
# ── flatpak package installation ──────────────────────────────────────────────
def install_flatpak_packages(to_install: list[str]) -> None:
print("\n=== Flatpak Packages ===")
if not has_cmd("flatpak"):
print(" flatpak is not installed.")
if not _yn(" Install flatpak now? [y/N] "):
warn("flatpak not installed — skipping Flatpak section")
return
result = run([PKG_MGR, "install", "-y", "flatpak"], as_sudo=True, check=False)
if result.returncode != 0 or not has_cmd("flatpak"):
err("flatpak installation failed — skipping Flatpak section")
return
run(
["flatpak", "remote-add", "--if-not-exists", "flathub",
"https://dl.flathub.org/repo/flathub.flatpakrepo"],
as_sudo=True, check=False,
)
for pkg_id in to_install:
print(f"\n Installing {pkg_id} ...")
result = run(["flatpak", "install", "--noninteractive", "flathub", pkg_id], check=False)
if result.returncode != 0:
err(f"Flatpak failed to install: {pkg_id}")
# ── custom package installation ───────────────────────────────────────────────
@dataclass
class CustomPackage:
name: str
url: Optional[str] = None # archive download URL (not required for all install methods)
sha256: Optional[str] = None # hex digest to compare against
sha256_url: Optional[str] = None # URL to a minisig file
minisign_key: Optional[str] = None # base64 public key for minisign verification
install_path: Optional[str] = None # override the default install-check path
def _default_install_path(pkg: CustomPackage) -> Optional[Path]:
n = pkg.name.lower()
if n.startswith("go-"):
return Path("/usr/local/go")
if "firecracker" in n:
return Path("/usr/local/bin/firecracker")
if "zig" in n:
return Path("/usr/local/bin/zig")
if n == "nvm":
return Path("~/.nvm")
if n == "pyenv":
return Path("~/.pyenv")
if n == "neovim":
return Path("/opt/nvim-linux-x86_64")
return None
def is_custom_pkg_installed(pkg: CustomPackage) -> tuple[bool, Optional[Path]]:
"""Return (is_installed, check_path)."""
raw = Path(pkg.install_path) if pkg.install_path else _default_install_path(pkg)
if raw is None:
return False, None
check = raw.expanduser()
return check.exists(), check
def _sha256_of(path: Path) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
def _verify(archive: Path, pkg: CustomPackage) -> bool:
"""Returns True if verification passed (or nothing to verify), False on failure."""
if pkg.sha256:
actual = _sha256_of(archive)
if actual != pkg.sha256:
err(f"SHA256 mismatch for {pkg.name}: expected {pkg.sha256}, got {actual}")
return False
print(" SHA256 OK")
elif pkg.sha256_url:
sig_path = archive.parent / Path(pkg.sha256_url).name
if not _download(pkg.sha256_url, sig_path):
return False
if has_cmd("minisign"):
cmd = ["minisign", "-Vm", str(archive), "-x", str(sig_path)]
if pkg.minisign_key:
cmd += ["-P", pkg.minisign_key]
result = run(cmd, check=False)
if result.returncode != 0:
err(f"minisign verification failed for {pkg.name}")
return False
print(" minisign OK")
else:
warn(f"minisign not installed — skipping signature verification for {pkg.name}")
return True
def _url_arch_ok(pkg: CustomPackage) -> bool:
"""If the URL clearly targets a different arch than the host, warn and return False."""
if not pkg.url:
return True
if _arch_matches(pkg.url):
return True
if _has_other_arch_token(pkg.url):
warn(
f"{pkg.name}: URL targets {_other_arch()} but host is {ARCH}. "
f"Update formatted_packages.txt with a matching URL/SHA256."
)
return False
return True # ambiguous — let it proceed
def _install_go(archive: Path) -> None:
go_root = Path("/usr/local/go")
if go_root.exists():
print(f" Removing existing Go at {go_root} ...")
run(["rm", "-rf", str(go_root)], as_sudo=True)
run(["tar", "-C", "/usr/local", "-xzf", str(archive)], as_sudo=True)
profile_line = "export PATH=$PATH:/usr/local/go/bin"
profile_script = "/etc/profile.d/local_go.sh"
run(["bash", "-c",
f"grep -qxF {profile_line!r} {profile_script} 2>/dev/null || "
f"echo {profile_line!r} >> {profile_script}"],
as_sudo=True, check=False)
print(f" Go installed to {go_root}")
def _install_firecracker(archive: Path, tmp: Path) -> None:
with tarfile.open(archive) as tf:
tf.extractall(tmp, filter="data")
binary = next(
(p for p in tmp.rglob("firecracker*")
if p.is_file() and not p.suffix == ".debug" and "debug" not in p.name),
None,
)
if binary is None:
err("firecracker binary not found in archive")
return
dest = Path("/usr/local/bin/firecracker")
run(["cp", str(binary), str(dest)], as_sudo=True)
run(["chmod", "755", str(dest)], as_sudo=True)
print(f" firecracker installed to {dest}")
def _install_zig(pkg: CustomPackage, archive: Path, tmp: Path) -> None:
parent = Path("/usr/local")
version = pkg.name.split("-", 1)[1] # "zig-0.16.0" → "0.16.0"
zig_dir = parent / f"zig-{version}"
if zig_dir.exists():
run(["rm", "-rf", str(zig_dir)], as_sudo=True)
run(["tar", "-C", str(parent), "-xJf", str(archive)], as_sudo=True)
extracted = next(parent.glob(f"zig-{ARCH}-linux*"), None)
if extracted and extracted != zig_dir:
run(["mv", str(extracted), str(zig_dir)], as_sudo=True)
symlink = Path("/usr/local/bin/zig")
run(["ln", "-sf", str(zig_dir / "zig"), str(symlink)], as_sudo=True)
print(f" Zig installed to {zig_dir}, symlinked at {symlink}")
def _install_pyenv() -> None:
print(" Installing pyenv via curl ...")
if shell("curl https://pyenv.run | bash", check=False).returncode != 0:
err("pyenv installation failed")
return
print(" pyenv installed to ~/.pyenv")
def _install_nvm() -> None:
data = _fetch_json("https://api.github.com/repos/nvm-sh/nvm/releases/latest")
if data is None:
return
version = data["tag_name"]
install_url = f"https://raw.githubusercontent.com/nvm-sh/nvm/{version}/install.sh"
print(f" Installing NVM {version} via curl ...")
if shell(f"curl -o- {install_url} | bash", check=False).returncode != 0:
err("NVM installation failed")
return
print(f" NVM {version} installed to ~/.nvm")
def _install_neovim(pkg: CustomPackage, tmp: Path) -> None:
data = _fetch_json("https://api.github.com/repos/neovim/neovim/releases/latest")
if data is None:
return
asset_name = "nvim-linux-x86_64.tar.gz"
asset = next((a for a in data["assets"] if a["name"] == asset_name), None)
if asset is None:
err(f"Neovim asset {asset_name} not found")
return
expected_digest = asset.get("digest")
if not expected_digest or not expected_digest.startswith("sha256:"):
err("Neovim asset digest missing or invalid")
return
expected_hash = expected_digest.split(":", 1)[1]
dest = tmp / asset_name
if not _download(asset["browser_download_url"], dest):
return
actual_hash = _sha256_of(dest)
if actual_hash != expected_hash:
err(f"Neovim SHA256 mismatch: expected {expected_hash}, got {actual_hash}")
return
print(" SHA256 OK")
print(" Extracting Neovim to /opt ...")
run(["rm", "-rf", "/opt/nvim-linux-x86_64"], as_sudo=True)
run(["tar", "-C", "/opt", "-xzf", str(dest)], as_sudo=True)
profile_line = 'export PATH="$PATH:/opt/nvim-linux-x86_64/bin"'
profile_script = "/etc/profile.d/neovim.sh"
run(["bash", "-c",
f"grep -qxF {profile_line!r} {profile_script} 2>/dev/null || "
f"echo {profile_line!r} >> {profile_script}"],
as_sudo=True, check=False)
print(f" Neovim installed to /opt/nvim-linux-x86_64")
def _clone_nvim_config() -> None:
config_dir = Path.home() / ".config" / "nvim"
repo_url = "git@github.com:JMR-dev/nvim-config.git"
print(f"\n[Neovim] Setting up configuration from {repo_url} ...")
if config_dir.exists():
print(f" Removing existing configuration at {config_dir} ...")
shutil.rmtree(config_dir)
config_dir.parent.mkdir(parents=True, exist_ok=True)
print(f" Cloning to {config_dir} ...")
# We use a temp clone and then move to ensure we handle the "rename" part of the request
# although cloning directly to 'nvim' is effectively the same.
# The user asked: "clones my nvim config to $HOME/.config/$REPO and renames the repo root folder to just nvim"
repo_name = repo_url.split("/")[-1].replace(".git", "")
temp_clone = config_dir.parent / repo_name
if temp_clone.exists():
shutil.rmtree(temp_clone)
result = run(["git", "clone", repo_url, str(temp_clone)], check=False)
if result.returncode != 0:
err("Neovim configuration clone failed")
return
print(f" Renaming {temp_clone.name} to {config_dir.name} ...")
temp_clone.rename(config_dir)
print(f" Neovim configuration ready at {config_dir}")
def ensure_node_lts() -> None:
"""If nvm is present, ensure Node LTS is installed and set as the default."""
nvm_dir = Path.home() / ".nvm"
if not nvm_dir.exists():
return
# nvm version lts/* prints the installed LTS version, or "N/A" if not installed
check = shell(
'bash -c "source ~/.nvm/nvm.sh 2>/dev/null && nvm version lts/* 2>/dev/null"',
capture_output=True, text=True, check=False,
)
installed = check.stdout.strip()
if installed and installed != "N/A":
print(f"\n[NVM] Node LTS ({installed}) already installed.")
else:
print("\n[NVM] Installing Node.js LTS ...")
result = shell('bash -c "source ~/.nvm/nvm.sh && nvm install --lts"', check=False)
if result.returncode != 0:
err("Node.js LTS install via nvm failed")
return
print(" Node.js LTS installed.")
print("[NVM] Setting Node LTS as default ...")
result = shell(
"bash -c \"source ~/.nvm/nvm.sh && nvm alias default 'lts/*' && nvm use --lts\"",
check=False,
)
if result.returncode != 0:
err("Setting nvm default to LTS failed")
def _latest_stable_python(pyenv_bin: Path) -> Optional[str]:
"""Return the latest stable CPython 3.x version string from `pyenv install --list`."""
result = subprocess.run(
[str(pyenv_bin), "install", "--list"],
capture_output=True, text=True, check=False,
)
if result.returncode != 0:
err("pyenv install --list failed")
return None
# Match only pure X.Y.Z lines — excludes a1/b1/rc1/dev suffixes and PyPy/Anaconda/etc.
stable_re = re.compile(r"^\s*(\d+)\.(\d+)\.(\d+)\s*$")
versions: list[tuple[int, int, int]] = []
for line in result.stdout.splitlines():
m = stable_re.match(line)
if m:
major, minor, patch = int(m.group(1)), int(m.group(2)), int(m.group(3))
if major >= 3:
versions.append((major, minor, patch))
if not versions:
return None
versions.sort()
return ".".join(str(p) for p in versions[-1])
def ensure_python_latest() -> Optional[threading.Thread]:
"""If pyenv is present, ensure the latest stable Python is installed and set as global.
When a compile is required, runs it in a background thread and returns the
thread handle. The caller must `join()` it before writing the run log so any
install/global failure is captured. Returns None if no compile was needed.
"""
pyenv_dir = Path.home() / ".pyenv"
if not pyenv_dir.exists():
return None
pyenv_bin = pyenv_dir / "bin" / "pyenv"
if not pyenv_bin.exists():
warn(f"pyenv binary not found at {pyenv_bin}")
return None
latest = _latest_stable_python(pyenv_bin)
if latest is None:
err("Could not determine latest stable Python from pyenv")
return None
installed = subprocess.run(
[str(pyenv_bin), "versions", "--bare"],
capture_output=True, text=True, check=False,
).stdout.split()
if latest in installed:
# Fast path — no compile needed, just set global synchronously.
print(f"\n[pyenv] Python {latest} already installed.")
print(f"[pyenv] Setting Python {latest} as global default ...")
if run([str(pyenv_bin), "global", latest], check=False).returncode != 0:
err(f"pyenv global {latest} failed")
return None
print(f"\n[pyenv] Backgrounding install of Python {latest} "
f"(compile may take several minutes; output captured) ...")
start = time.monotonic()
def _worker():
install_cmd = [str(pyenv_bin), "install", "--skip-existing", latest]
p1 = subprocess.run(install_cmd, capture_output=True, text=True, check=False)
elapsed = int(time.monotonic() - start)
if p1.returncode != 0:
err(f"pyenv install {latest} failed after {elapsed}s")
tail = "\n".join(p1.stderr.splitlines()[-20:]) if p1.stderr else ""
if tail:
print(f"\n[pyenv stderr tail]\n{tail}")
return
p2 = subprocess.run(
[str(pyenv_bin), "global", latest],
capture_output=True, text=True, check=False,
)
if p2.returncode != 0:
err(f"pyenv global {latest} failed")
return
print(f"\n[pyenv] Python {latest} installed and set as global default ({elapsed}s).")
t = threading.Thread(target=_worker, daemon=True, name="pyenv-install")
t.start()
return t
def install_custom_packages(to_install: list[CustomPackage]) -> None:
print("\n=== Custom Packages ===")
for pkg in to_install:
_, check_path = is_custom_pkg_installed(pkg)
print(f"\n Installing {pkg.name} ..."
+ (f" (install path: {check_path})" if check_path else ""))
if check_path is None:
warn(f"{pkg.name}: no known install path — script will not detect future installs")
name_lower = pkg.name.lower()
# Handlers that manage their own download/install
if name_lower == "nvm":
_install_nvm()
continue
if name_lower == "pyenv":
_install_pyenv()
continue
if name_lower == "neovim":
with tempfile.TemporaryDirectory() as tmp_str:
_install_neovim(pkg, Path(tmp_str))
continue
if not pkg.url:
warn(f"No URL or install handler for '{pkg.name}' — skipping")
continue
if not _url_arch_ok(pkg):
continue
with tempfile.TemporaryDirectory() as tmp_str:
tmp = Path(tmp_str)
archive = tmp / Path(pkg.url).name
if not _download(pkg.url, archive):
continue
if not _verify(archive, pkg):
continue
if name_lower.startswith("go-"):
_install_go(archive)
elif "firecracker" in name_lower:
_install_firecracker(archive, tmp)
elif "zig" in name_lower:
_install_zig(pkg, archive, tmp)
else:
warn(f"No install handler for '{pkg.name}' — skipping")
# ── pre-install checks ───────────────────────────────────────────────────────
def check_system_packages(names: list[str]) -> dict:
overrides = _OVERRIDES.get(PKG_MGR, {})
resolved, skipped = resolve_system_pkgs(names)
# Packages remapped to different name(s) (not skipped entirely)
remapped = [(n, overrides[n]) for n in names if n in overrides and overrides[n] is not None]
special = [p for p in resolved if p in _SPECIAL_PKGS]
regular = [p for p in resolved if p not in _SPECIAL_PKGS]
to_install_r, already_r = [], []
for p in regular:
(already_r if is_system_pkg_installed(p) else to_install_r).append(p)
to_install_s, already_s = [], []
for p in special:
(already_s if is_special_pkg_installed(p) else to_install_s).append(p)
return {
"to_install_regular": to_install_r,
"to_install_special": to_install_s,
"already_installed": already_r + already_s,
"skipped": skipped,
"remapped": remapped,
}
def check_flatpak_packages(pkg_ids: list[str]) -> dict:
to_install, already = [], []
for p in pkg_ids:
(already if is_flatpak_installed(p) else to_install).append(p)
return {"to_install": to_install, "already_installed": already}
def check_custom_packages(packages: list[CustomPackage]) -> dict:
to_install, already = [], []
for pkg in packages:
installed, path = is_custom_pkg_installed(pkg)
(already if installed else to_install).append((pkg, path))
return {"to_install": [p for p, _ in to_install],
"already_installed": already}
def _fmt(items: list, limit: int = 6) -> str:
names = [str(i) for i in items]
shown = " ".join(names[:limit])
return shown + (f" … +{len(names)-limit} more" if len(names) > limit else "")
def print_check_summary(sys_c: dict, flat_c: dict, cust_c: dict, only: Optional[str]) -> int:
"""Print pre-install summary. Returns total count to install."""
total = 0
if only in (None, "system"):
to_r = sys_c["to_install_regular"]
to_s = sys_c["to_install_special"]
ok = sys_c["already_installed"]
skip = sys_c["skipped"]
remap = sys_c["remapped"]
print("\nSystem packages:")
if ok:
print(f" [OK] {len(ok):3d} already installed")
n = len(to_r) + len(to_s)
if n:
print(f" [INSTALL] {n:3d} to install: {_fmt(to_r + to_s)}")
if skip:
print(f" [SKIP] {len(skip):3d} overridden (→ skip): {_fmt(skip)}")
if remap:
pairs = " ".join(f"{a}→{','.join(b)}" for a, b in remap)
print(f" [REMAP] remapped: {pairs}")
total += n
if only in (None, "flatpak") and flat_c:
to = flat_c["to_install"]
ok = flat_c["already_installed"]
print("\nFlatpak packages:")
if ok:
print(f" [OK] {len(ok):3d} already installed")
if to:
print(f" [INSTALL] {len(to):3d} to install: {_fmt(to)}")
total += len(to)
if only in (None, "custom"):
to = cust_c["to_install"]
ok = cust_c["already_installed"]
print("\nCustom packages:")
for pkg, path in ok:
print(f" [OK] {pkg.name} ({path})")
for pkg in to:
_, path = is_custom_pkg_installed(pkg)
print(f" [INSTALL] {pkg.name}" + (f" → {path}" if path else ""))
total += len(to)
return total
# ── ssh key + github auth ─────────────────────────────────────────────────────
def _yn(prompt: str) -> bool:
"""Ask a y/N question. Returns True only for 'y'."""
try:
return input(prompt).strip().lower() == "y"
except (EOFError, KeyboardInterrupt):
print()
return False
def _gh_logged_in() -> bool:
return subprocess.run(
["gh", "auth", "status"], capture_output=True, check=False
).returncode == 0
def _offer_github_upload(pub_keys: list[Path]) -> None:
if not pub_keys:
print(" No public key found to upload.")
return
pub_key = max(pub_keys, key=lambda p: p.stat().st_mtime)
if not _yn(f"\n Upload {pub_key.name} to your GitHub profile? [y/N] "):
return
default_title = f"{getpass.getuser()}@{os.uname().nodename}"
try:
title = input(f" Key title [{default_title}]: ").strip() or default_title
except (EOFError, KeyboardInterrupt):
title = default_title
def check_and_setup_ssh() -> None:
if not has_cmd("gh"):
print("\n[GitHub CLI] gh not installed — skipping authentication.")
return
if _gh_logged_in():
print("\n[GitHub CLI] Already authenticated.")
return
if not _yn("\n[GitHub CLI] Would you like to authenticate the GitHub CLI? [y/N] "):
return
result = subprocess.run(["gh", "auth", "login"], check=False)
if result.returncode != 0:
err("gh auth login failed — skipping key upload.")
return
# ── file parser ───────────────────────────────────────────────────────────────
def parse_packages_file(path: Path) -> tuple[list[str], list[str], list[CustomPackage]]:
system_pkgs: list[str] = []
flatpak_pkgs: list[str] = []
custom_pkgs: list[CustomPackage] = []
current: dict = {}
section: Optional[str] = None
def flush_custom():
if "name" in current:
custom_pkgs.append(CustomPackage(**current))
current.clear()
for raw in path.read_text().splitlines():
line = raw.strip()
if line == "=== System Packages ===":
section = "system"
elif line == "=== Flatpak Packages ===":
flush_custom()
section = "flatpak"
elif line == "=== Custom Installed Packages ===":
flush_custom()
section = "custom"
elif not line:
if section == "custom":
flush_custom()
elif line.startswith("==="):
pass
elif section == "system":
system_pkgs.append(line)
elif section == "flatpak":
flatpak_pkgs.append(line)
elif section == "custom" and " - " in line:
key, _, val = line.partition(" - ")
key = key.strip().lower().replace(" ", "_")
val = val.strip()
if key == "name":
current["name"] = val
elif key == "url":
current["url"] = val
elif key == "sha256":
current["sha256"] = val
elif key == "sha256_url_tar":
current["sha256_url"] = val
elif key == "minisign_key":
current["minisign_key"] = val
elif key == "install_path":
current["install_path"] = val
flush_custom()
return system_pkgs, flatpak_pkgs, custom_pkgs
# ── entry point ───────────────────────────────────────────────────────────────
def main() -> None:
ap = argparse.ArgumentParser(
description="Bootstrap packages from formatted_packages.txt"
)
ap.add_argument("--only", choices=["system", "flatpak", "custom"],
help="Install only the named section")
ap.add_argument("--no-gui", action="store_true",
help="Skip GUI applications (suitable for headless environments). "
"Excludes GUI system packages and the entire Flatpak section.")
ap.add_argument("--file", default=str(PACKAGES_FILE), metavar="PATH",
help="Path to packages file (default: formatted_packages.txt next to this script)")
args = ap.parse_args()
pkg_file = Path(args.file)
if not pkg_file.exists():
sys.exit(f"Packages file not found: {pkg_file}")
system_pkgs, flatpak_pkgs, custom_pkgs = parse_packages_file(pkg_file)
print(f"Architecture: {ARCH}")
print(f"Package manager: {PKG_MGR}")
if args.no_gui:
print("Mode: headless (--no-gui) — skipping GUI apps and Flatpak")
print("Checking installed packages ...")
if args.no_gui:
skipped_gui = [p for p in system_pkgs if p in _GUI_SYSTEM_PKGS]
system_pkgs = [p for p in system_pkgs if p not in _GUI_SYSTEM_PKGS]
if skipped_gui:
print(f" [NO-GUI] Skipping GUI system packages: {_fmt(skipped_gui)}")
flatpak_pkgs = []
do_flatpak = args.only in (None, "flatpak") and not args.no_gui
sys_c = check_system_packages(system_pkgs) if args.only in (None, "system") else {}
flat_c = check_flatpak_packages(flatpak_pkgs) if do_flatpak else {}
cust_c = check_custom_packages(custom_pkgs) if args.only in (None, "custom") else {}
total = print_check_summary(sys_c, flat_c, cust_c, args.only)
if total == 0:
print("\nAll packages already installed.")
write_run_log()
return
try:
answer = input(f"\n{total} item(s) to install. Proceed? [y/N] ").strip().lower()
except (EOFError, KeyboardInterrupt):
print()
sys.exit("Aborted.")
if answer != "y":
sys.exit("Aborted.")
check_sudo()
if args.only in (None, "system"):
install_system_packages(sys_c["to_install_regular"], sys_c["to_install_special"])
if do_flatpak:
install_flatpak_packages(flat_c["to_install"])
pyenv_thread: Optional[threading.Thread] = None
if args.only in (None, "custom"):
install_custom_packages(cust_c["to_install"])
ensure_node_lts()
pyenv_thread = ensure_python_latest()
if args.only is None:
check_and_setup_ssh()
_clone_nvim_config()
if pyenv_thread is not None:
if pyenv_thread.is_alive():
print("\n[pyenv] Waiting for background Python install to finish ...")
pyenv_thread.join()
write_run_log()
print("\nDone.")
if __name__ == "__main__":
main()