linuxrouter/routlin/install.py
2026-06-03 01:09:12 -04:00

715 lines
24 KiB
Python

#!/usr/bin/env python3
"""
install.py -- Routlin one-time setup wizard.
Installs required system packages, optionally sets up the web dashboard
(Docker + docker-compose), and optionally configures external HTTPS access
via Caddy.
Usage:
sudo python3 install.py
"""
import argparse
import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
from pathlib import Path
AUTO_YES = False
SCRIPT_DIR = Path(__file__).parent.resolve()
COMPOSE_FILE = SCRIPT_DIR.parent / "docker" / "routlin-dash" / "docker-compose.yml"
CADDYFILE = Path("/etc/caddy/Caddyfile")
FLASK_PORT = 25327
PRODUCT_NAME = os.environ.get('PRODUCT_NAME', 'routlin')
SYSTEMD_DIR = Path("/etc/systemd/system")
# Dashboard dotfiles
DASHB_QUEUE_FILE = SCRIPT_DIR / ".dashboard-queue"
DASHB_DONE_FILE = SCRIPT_DIR / ".dashboard-done"
DASHB_LAST_RUN_FILE = SCRIPT_DIR / ".dashboard-last-run"
DASHB_LOCK_FILE = SCRIPT_DIR / ".dashboard-lock"
DASHB_PENDING_FILE = SCRIPT_DIR / ".dashboard-pending"
DASHB_SCRIPT_FILE = SCRIPT_DIR / "do_dashboard_queue.sh"
HEALTH_FILE = SCRIPT_DIR / ".health"
SNAPSHOTS_DIR = SCRIPT_DIR / ".snapshots"
# Dashboard systemd timer
DASHB_TIMER_NAME = f"{PRODUCT_NAME}-dashboard-queue"
DASHB_TIMER_FILE = SYSTEMD_DIR / f"{DASHB_TIMER_NAME}.timer"
DASHB_TIMER_SVC_FILE = SYSTEMD_DIR / f"{DASHB_TIMER_NAME}.service"
DASHB_TIMER_INTERVAL_SEC = 60
# ===================================================================
# Helpers
# ===================================================================
def die(msg):
print(f"\nERROR: {msg}", file=sys.stderr)
sys.exit(1)
def _compose_env():
"""Return an env dict with HOME set to the invoking user's home, not root's."""
import pwd
sudo_user = os.environ.get('SUDO_USER')
if sudo_user:
try:
home = pwd.getpwnam(sudo_user).pw_dir
return {**os.environ, 'HOME': home}
except KeyError:
pass
return os.environ.copy()
def check_root():
if os.geteuid() != 0:
die("This script must be run as root (sudo python3 install.py).")
def prompt_yn(question, default=None):
"""Prompt for yes/no. default: 'y', 'n', or None (require explicit answer)."""
if AUTO_YES:
choice = (default == "y") if default else True
print(f" {question} [auto: {'y' if choice else 'n'}]")
return choice
hint = {None: "[y/n]", "y": "[Y/n]", "n": "[y/N]"}[default]
while True:
ans = input(f" {question} {hint}: ").strip().lower()
if ans in ("y", "yes"):
return True
if ans in ("n", "no"):
return False
if ans == "" and default:
return default == "y"
print(" Please enter y or n.")
def prompt_str(question, default=None, secret=False):
"""Prompt for a string value."""
hint = f" [{default}]" if default else ""
if secret:
import getpass
ans = getpass.getpass(f" {question}{hint}: ")
else:
ans = input(f" {question}{hint}: ").strip()
if not ans and default is not None:
return default
return ans
def run(cmd, check=True, capture=False):
"""Run a shell command list."""
return subprocess.run(cmd, check=check,
capture_output=capture, text=True)
def header(title):
print()
print("=" * 62)
print(f" {title}")
print("=" * 62)
print()
# ===================================================================
# Package manager detection
# ===================================================================
def detect_pm():
"""Return 'apt', 'dnf', 'yum', 'pacman', 'zypper', or None."""
for pm in ("apt-get", "dnf", "yum", "pacman", "zypper"):
if shutil.which(pm):
return pm.replace("-get", "") # apt-get -> apt
return None
# Core packages: binary-to-check -> {pm: package-name}
_CORE_PKGS = {
"dnsmasq": {
"apt": "dnsmasq",
"dnf": "dnsmasq",
"yum": "dnsmasq",
"pacman": "dnsmasq",
"zypper": "dnsmasq",
},
"nft": {
"apt": "nftables",
"dnf": "nftables",
"yum": "nftables",
"pacman": "nftables",
"zypper": "nftables",
},
"chronyd": {
"apt": "chrony",
"dnf": "chrony",
"yum": "chrony",
"pacman": "chrony",
"zypper": "chrony",
},
"freeradius": {
"apt": "freeradius",
"dnf": "freeradius",
"yum": "freeradius",
"pacman": "freeradius",
"zypper": "freeradius",
},
"avahi-daemon": {
"apt": "avahi-daemon",
"dnf": "avahi",
"yum": "avahi",
"pacman": "avahi",
"zypper": "avahi",
},
}
_INSTALL_CMD = {
"apt": ["apt-get", "install", "-y"],
"dnf": ["dnf", "install", "-y"],
"yum": ["yum", "install", "-y"],
"pacman": ["pacman", "-S", "--noconfirm"],
"zypper": ["zypper", "install", "-y"],
}
def install_core_packages(pm):
header("Core Package Installation")
missing = []
for binary, pm_map in _CORE_PKGS.items():
if not shutil.which(binary):
pkg = pm_map.get(pm)
if pkg:
missing.append((binary, pkg))
else:
print(f" WARNING: no package mapping for '{binary}' on {pm}. Install manually.")
if not missing:
print(" All required packages are already installed.")
fix_freeradius_service_file()
return
print(" The following packages are not installed:")
for binary, pkg in missing:
print(f" {pkg} (provides: {binary})")
print()
if not prompt_yn("Install them now?", default="y"):
die("Cannot continue without required packages.")
packages = [pkg for _, pkg in missing]
cmd = _INSTALL_CMD[pm] + packages
print()
result = subprocess.run(cmd)
if result.returncode != 0:
die("Package installation failed. Install manually and retry.")
print("\n Core packages installed.")
fix_freeradius_service_file()
def fix_freeradius_service_file():
svc = Path("/lib/systemd/system/freeradius.service")
if not svc.exists():
return
text = svc.read_text()
if "MemoryLimit=" not in text:
return
svc.write_text(text.replace("MemoryLimit=", "MemoryMax="))
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
print(" Fixed deprecated MemoryLimit= in freeradius.service (replaced with MemoryMax=).")
# ===================================================================
# Docker
# ===================================================================
def _docker_installed():
return shutil.which("docker") is not None
def install_docker(pm):
if _docker_installed():
print(" Docker is already installed.")
return
print(" Docker is not installed.")
print()
if pm in ("apt", "dnf", "yum"):
if not prompt_yn("Install Docker via the official get.docker.com script?", default="y"):
die("Docker is required for the dashboard. Install it manually and re-run.")
print()
result = subprocess.run(
["bash", "-c", "curl -fsSL https://get.docker.com | sh"],
check=False
)
if result.returncode != 0:
die("Docker installation failed. Install manually and re-run.")
# Add current user to docker group
sudo_user = os.environ.get("SUDO_USER")
if sudo_user:
subprocess.run(["usermod", "-aG", "docker", sudo_user], check=False)
print(f"\n Added {sudo_user} to the docker group.")
print("\n Docker installed.")
else:
print(" Automatic Docker installation is not supported on this package manager.")
print(" Please install Docker manually, then re-run install.py.")
die("Docker required.")
# ===================================================================
# docker-compose.yml setup
# ===================================================================
def _set_env_var(content, key, value):
"""Replace ' - KEY=...' line in docker-compose env block."""
pattern = rf"^(\s*- {re.escape(key)}=).*$"
replacement = rf"\g<1>{value}"
new, count = re.subn(pattern, replacement, content, flags=re.MULTILINE)
if count == 0:
# Key not found; shouldn't happen with our template, warn and move on
print(f" WARNING: could not find {key} in docker-compose.yml")
return new
def _dash_already_configured():
if not COMPOSE_FILE.exists():
return False
return bool(re.search(r"^\s*- SECRET_KEY=\S", COMPOSE_FILE.read_text(), re.MULTILINE))
def setup_docker_compose(reuse_config=False):
header("Dashboard Configuration")
if not COMPOSE_FILE.exists():
die(f"docker-compose.yml not found at {COMPOSE_FILE}\n"
f" Ensure the routlin-dash directory is at {COMPOSE_FILE.parent}")
if reuse_config:
import time
cache_bust = str(int(time.time()))
env = _compose_env()
print("\n Stopping existing container...")
subprocess.run(["docker", "compose", "down"], cwd=COMPOSE_FILE.parent, env=env, check=False)
print("\n Building dashboard image...")
result = subprocess.run(
["docker", "compose", "build", "--build-arg", f"CACHE_BUST={cache_bust}"],
cwd=COMPOSE_FILE.parent, env=env, check=False
)
if result.returncode != 0:
die("docker compose build failed. Check the output above.")
print("\n Starting dashboard container...")
result = subprocess.run(
["docker", "compose", "up", "-d"],
cwd=COMPOSE_FILE.parent, env=env, check=False
)
if result.returncode != 0:
die("docker compose up failed. Check the output above.")
print(" Dashboard container started.")
return
content = COMPOSE_FILE.read_text()
print(" Generating SECRET_KEY...")
secret_key = secrets.token_urlsafe(96) # ~128 chars
print()
print(" SMTP is used to send email verification codes for new accounts.")
print(" (Gmail users: use an App Password, not your account password.)")
print()
manager_email = prompt_str("Initial manager account email")
while not manager_email or "@" not in manager_email:
print(" Please enter a valid email address.")
manager_email = prompt_str("Initial manager account email")
smtp_host = prompt_str("SMTP host", default="smtp.gmail.com")
smtp_port = prompt_str("SMTP port", default="587")
smtp_user = prompt_str("SMTP username (email)")
smtp_password = prompt_str("SMTP password", secret=True)
smtp_from = prompt_str("SMTP From address", default=smtp_user)
content = _set_env_var(content, "SECRET_KEY", secret_key)
content = _set_env_var(content, "INITIAL_MANAGER_EMAIL", manager_email)
content = _set_env_var(content, "SMTP_HOST", smtp_host)
content = _set_env_var(content, "SMTP_PORT", smtp_port)
content = _set_env_var(content, "SMTP_USER", smtp_user)
content = _set_env_var(content, "SMTP_PASSWORD", smtp_password)
content = _set_env_var(content, "SMTP_FROM", smtp_from)
COMPOSE_FILE.write_text(content)
print(f"\n Written: {COMPOSE_FILE}")
env = _compose_env()
print("\n Stopping existing container...")
subprocess.run(["docker", "compose", "down"], cwd=COMPOSE_FILE.parent, env=env, check=False)
print("\n Starting dashboard container...")
result = subprocess.run(
["docker", "compose", "up", "-d", "--build"],
cwd=COMPOSE_FILE.parent, env=env, check=False
)
if result.returncode != 0:
die("docker compose up failed. Check the output above.")
print(" Dashboard container started.")
def create_dotfiles():
dir_stat = SCRIPT_DIR.stat()
uid, gid = dir_stat.st_uid, dir_stat.st_gid
if not SNAPSHOTS_DIR.exists():
SNAPSHOTS_DIR.mkdir()
os.chown(SNAPSHOTS_DIR, uid, gid)
for f in (DASHB_QUEUE_FILE, DASHB_DONE_FILE, DASHB_LAST_RUN_FILE, DASHB_LOCK_FILE, DASHB_PENDING_FILE, HEALTH_FILE):
if not f.exists():
f.touch()
os.chown(f, uid, gid)
# ===================================================================
# Dashboard systemd timer
# ===================================================================
def install_dashboard_timer():
description = "Routlin dashboard pending-update processor"
timer_content = "\n".join([
"# Generated by install.py -- do not edit manually.",
"",
"[Unit]",
f"Description={description}",
"",
"[Timer]",
f"OnActiveSec={DASHB_TIMER_INTERVAL_SEC}s",
f"OnUnitActiveSec={DASHB_TIMER_INTERVAL_SEC}s",
"AccuracySec=10s",
"",
"[Install]",
"WantedBy=timers.target",
"",
])
service_content = "\n".join([
"# Generated by install.py -- do not edit manually.",
"",
"[Unit]",
f"Description={description}",
"",
"[Service]",
"Type=oneshot",
f"ExecStart=/bin/bash {DASHB_SCRIPT_FILE}",
"",
])
for path, content in ((DASHB_TIMER_FILE, timer_content), (DASHB_TIMER_SVC_FILE, service_content)):
path.write_text(content)
print(f" Written: {path}")
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
active = subprocess.run(
["systemctl", "is-active", f"{DASHB_TIMER_NAME}.timer"],
capture_output=True, text=True,
).stdout.strip() == "active"
subprocess.run(["systemctl", "enable", f"{DASHB_TIMER_NAME}.timer"], capture_output=True, text=True)
subprocess.run(["systemctl", "restart" if active else "start", f"{DASHB_TIMER_NAME}.timer"],
capture_output=True, text=True)
print(f" Timer {DASHB_TIMER_NAME}.timer enabled (runs every {DASHB_TIMER_INTERVAL_SEC}s).")
def remove_dashboard_timer():
subprocess.run(["systemctl", "disable", "--now", f"{DASHB_TIMER_NAME}.timer"],
capture_output=True, text=True)
for f in (DASHB_TIMER_FILE, DASHB_TIMER_SVC_FILE):
if f.exists():
f.unlink()
print(f" Removed: {f}")
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
# ===================================================================
# Caddy
# ===================================================================
def _caddy_installed():
return shutil.which("caddy") is not None
def install_caddy(pm):
if _caddy_installed():
print(" Caddy is already installed.")
return
print(" Caddy is not installed.")
print()
if pm == "apt":
if not prompt_yn("Install Caddy via the official Caddy apt repository?", default="y"):
die("Caddy is required for external access. Install manually and re-run.")
print()
cmds = [
["apt-get", "install", "-y", "debian-keyring", "debian-archive-keyring", "apt-transport-https", "curl"],
["bash", "-c",
"curl -1sLf 'https://dl.cloudsmith.io/public/caddy/stable/gpg.key' "
"| gpg --dearmor -o /usr/share/keyrings/caddy-stable-archive-keyring.gpg"],
["bash", "-c",
"curl -1sLf 'https://dl.cloudsmith.io/public/caddy/stable/debian.deb.txt' "
"| tee /etc/apt/sources.list.d/caddy-stable.list"],
["apt-get", "update"],
["apt-get", "install", "-y", "caddy"],
]
for cmd in cmds:
result = subprocess.run(cmd, check=False)
if result.returncode != 0:
die("Caddy installation failed. Install manually and re-run.")
print("\n Caddy installed.")
elif pm in ("dnf", "yum"):
if not prompt_yn("Install Caddy via COPR (dnf/yum)?", default="y"):
die("Caddy is required for external access. Install manually and re-run.")
print()
cmds = [
[pm, "copr", "enable", "-y", "@caddy/caddy"],
[pm, "install", "-y", "caddy"],
]
for cmd in cmds:
result = subprocess.run(cmd, check=False)
if result.returncode != 0:
die("Caddy installation failed. Install manually and re-run.")
print("\n Caddy installed.")
else:
print(" Automatic Caddy installation is not supported on this package manager.")
print(" Install Caddy manually from https://caddyserver.com/docs/install")
print(" then re-run install.py.")
die("Caddy required for external access.")
def _external_access_domain():
"""Return the configured domain if a Routlin Caddy block exists, else None."""
if not CADDYFILE.exists():
return None
m = re.search(rf'(\S+)\s*\{{[^}}]*127\.0\.0\.1:{FLASK_PORT}', CADDYFILE.read_text(), re.DOTALL)
return m.group(1) if m else None
def setup_caddy(domain, email):
block = (
f"\n# Routlin Dashboard\n"
f"{domain} {{\n"
f" encode gzip\n"
f" tls {email}\n"
f" reverse_proxy 127.0.0.1:{FLASK_PORT}\n"
f"}}\n"
)
if CADDYFILE.exists():
existing = CADDYFILE.read_text()
if f"127.0.0.1:{FLASK_PORT}" in existing:
print(f" Routlin block already present in {CADDYFILE}, skipping.")
else:
CADDYFILE.write_text(existing + block)
print(f" Appended {domain} block to {CADDYFILE}")
else:
CADDYFILE.parent.mkdir(parents=True, exist_ok=True)
CADDYFILE.write_text(block.lstrip())
print(f" Written: {CADDYFILE}")
# Start or reload
active = subprocess.run(
["systemctl", "is-active", "caddy"],
capture_output=True, text=True
).stdout.strip() == "active"
if active:
subprocess.run(["systemctl", "reload", "caddy"], check=False)
print(" Caddy reloaded.")
else:
subprocess.run(["systemctl", "enable", "--now", "caddy"], check=False)
print(" Caddy enabled and started.")
def _lan_ip():
"""Read the LAN IP from routlin's networkd config files.
Prefers the physical (untagged) interface - its Name has no dot.
Falls back to the first address found across all routlin network files.
"""
import glob
fallback = None
try:
for path in sorted(glob.glob("/etc/systemd/network/10-routlin-*.network")):
content = Path(path).read_text()
name_m = re.search(r'^Name=(.+)$', content, re.MULTILINE)
addr_m = re.search(r'^Address=(\d+\.\d+\.\d+\.\d+)/', content, re.MULTILINE)
if not name_m or not addr_m:
continue
ip = addr_m.group(1)
if '.' not in name_m.group(1).strip(): # physical interface, no dot
return ip
if fallback is None:
fallback = ip
except Exception:
pass
return fallback or "this server"
# ===================================================================
# Main
# ===================================================================
def main():
global AUTO_YES
parser = argparse.ArgumentParser(description="Routlin setup wizard.")
parser.add_argument('-y', '--yes', action='store_true',
help='Auto-accept all yes/no prompts using their defaults.')
args = parser.parse_args()
AUTO_YES = args.yes
check_root()
print()
print("=" * 62)
print(" Routlin Setup")
print("=" * 62)
print()
print(" This wizard installs required packages and optionally")
print(" sets up the web dashboard and external HTTPS access.")
# Package manager ===================================================
pm = detect_pm()
if pm is None:
print()
print(" WARNING: Could not detect a supported package manager.")
print(" Supported: apt, dnf, yum, pacman, zypper")
print(" Please install packages manually.")
pm_ok = False
else:
print(f"\n Detected package manager: {pm}")
pm_ok = True
# Core packages =====================================================
if pm_ok:
install_core_packages(pm)
# Dashboard =========================================================
header("Dashboard (optional)")
print(" The Routlin Dashboard is a web UI for managing the router.")
print(" It runs as a Docker container. Without it, config.json must")
print(" be edited manually.")
print()
next_step = (
f"\n Next step: use the web dashboard to configure your network, or\n"
f" configure {SCRIPT_DIR}/config.json manually and then run:\n"
f" sudo python3 {SCRIPT_DIR}/core.py --apply"
)
dash_installed = _dash_already_configured()
if dash_installed:
want_dashboard = prompt_yn("Web dashboard is already installed. Rebuild Docker image?", default="y")
else:
want_dashboard = prompt_yn("Install the web dashboard?", default="y")
if not want_dashboard:
print()
print(" Skipping dashboard setup.")
remove_dashboard_timer()
print()
print("Done.")
print(next_step)
return
reuse_config = False
if dash_installed:
reuse_config = prompt_yn(
"Re-use existing Docker configuration? (Keeps SECRET_KEY and SMTP credentials, preserving active sessions and email settings)",
default="y"
)
# Docker ============================================================
header("Docker")
if pm_ok:
install_docker(pm)
elif not _docker_installed():
die("Docker is required for the dashboard. Install it manually and re-run.")
else:
print(" Docker is already installed.")
# docker-compose.yml ================================================
setup_docker_compose(reuse_config=reuse_config)
create_dotfiles()
# Dashboard timer ===================================================
header("Dashboard Timer")
install_dashboard_timer()
# External access ===================================================
header("External Access (optional)")
ext_domain = _external_access_domain()
if ext_domain:
lan = _lan_ip()
print(f" External access to the web dashboard is already configured.")
print(f" https://{ext_domain}/")
print(f" The dashboard is also reachable on your local network at:")
print(f" http://{lan}:{FLASK_PORT}/")
print()
print("Done.")
print(next_step)
return
print(" External access lets you reach the dashboard from outside")
print(" your LAN via HTTPS using a domain name and Caddy.")
print()
want_external = prompt_yn("Configure external HTTPS access?", default="n")
if not want_external:
lan = _lan_ip()
print()
print(" Dashboard is accessible on your local network at:")
print(f" http://{lan}:{FLASK_PORT}/")
print()
print("Done.")
print(next_step)
return
# Caddy =============================================================
header("Caddy (HTTPS)")
if pm_ok:
install_caddy(pm)
elif not _caddy_installed():
print(" Caddy is not installed. Install it from https://caddyserver.com/docs/install")
die("Caddy required for external access.")
else:
print(" Caddy is already installed.")
print()
domain = prompt_str("Domain name for the dashboard (e.g. routlin.example.com)")
while not domain or "." not in domain:
print(" Please enter a valid domain name.")
domain = prompt_str("Domain name")
email = prompt_str("Email address for TLS certificate (Let's Encrypt)")
while not email or "@" not in email:
print(" Please enter a valid email address.")
email = prompt_str("Email address for TLS certificate")
print()
setup_caddy(domain, email)
print()
print(" Dashboard will be accessible at:")
print(f" https://{domain}/")
print()
print(" NOTE: HTTPS certificate provisioning may take up to a minute.")
print(" Ensure DNS for your domain points to this machine's WAN IP")
print(" and that port 80 and 443 are forwarded to this machine.")
print()
print("Done.")
print(next_step)
if __name__ == "__main__":
main()