from db_bind_user_to_iface import bind_user_to_iface
from db_init_ip_table_registration import assign_missing_rt_table_ids
from db_init_ip_iface_addr_assign import reconcile_kernel_and_db_ipv4_addresses
+from db_init_route_defaults import seed_default_routes
ROOT = Path(__file__).resolve().parent
DB = ic.DB_PATH
def db_init_StanleyPark() -> int:
"""
- Given the local SQLite DB at ic.DB_PATH,
- it loads schema, upserts ifaces (x6, US), upserts server (x6, US),
- binds users (Thomas-x6→x6, Thomas-US→US), generates missing keypairs,
- commits, and prints public keys. Returns 0 on success (raises on failure).
+ Given the local SQLite DB at ic.DB_PATH, this:
+ 1) loads schema
+ 2) upserts ifaces (x6, US)
+ 3) upserts servers (x6, US)
+ 4) binds users (Thomas-x6→x6, Thomas-US→US)
+ 5) seeds per-iface default routes into Route
+ 6) assigns missing rt_table_id values from /etc/iproute2/rt_tables
+ 7) reconciles/assigns interface IPv4 addresses (kernel→DB, then pool)
+ 8) commits and prints status
+ Returns 0 on success (raises on failure).
"""
# 1) Schema
msg_wrapped_call("db_schema_load.sh", _run_local, "db_schema_load.sh")
# 2) DB work in one connection/commit
with ic.open_db(DB) as conn:
+ # ifaces + servers + user bindings
msg_wrapped_call("db_init_iface_x6.py (init_iface_x6)", init_iface_x6, conn)
msg_wrapped_call("db_init_server_x6.py (init_server_x6)", init_server_x6, conn)
msg_wrapped_call("bind_user_to_iface: Thomas-x6 → x6", bind_user_to_iface, conn, "x6", "Thomas-x6")
msg_wrapped_call("db_init_server_US.py (init_server_US)", init_server_US, conn)
msg_wrapped_call("bind_user_to_iface: Thomas-US → US", bind_user_to_iface, conn, "US", "Thomas-US")
+ # 5) seed default routes for the selected ifaces (no duplicates; idempotent)
msg_wrapped_call(
- "db_init_ip_table_registration"
- ,lambda: assign_missing_rt_table_ids(conn ,low=20000 ,high=29999 ,dry_run=False)
+ "db_init_route_defaults (x6,US)",
+ lambda: seed_default_routes(conn, iface_names=["x6","US"], overwrite=False)
)
+ # 6) assign rt_table_id from system tables (DB-only; no file writes)
msg_wrapped_call(
- "db_init_ip_iface_addr_assign"
- ,lambda: reconcile_kernel_and_db_ipv4_addresses(conn ,pool_cidr="10.0.0.0/16" ,assign_prefix=32 ,reserve_first=0 ,dry_run=False)
+ "db_init_ip_table_registration",
+ lambda: assign_missing_rt_table_ids(conn, low=20000, high=29999, dry_run=False)
)
+ # 7) reconcile/assign interface IPv4 addresses (kernel → DB; pool for missing)
+ msg_wrapped_call(
+ "db_init_ip_iface_addr_assign",
+ lambda: reconcile_kernel_and_db_ipv4_addresses(
+ conn,
+ pool_cidr="10.0.0.0/16",
+ assign_prefix=32,
+ reserve_first=0,
+ dry_run=False
+ )
+ )
+
+ # 8) commit
conn.commit()
print("✔ commit: database updated")
--- /dev/null
+#!/usr/bin/env python3
+"""
+db_init_route_defaults.py
+
+Business API:
+ seed_default_routes(conn ,iface_names ,overwrite=False ,metric=None)
+ -> (inserted_count ,notes[list])
+
+What it does:
+- For each iface in iface_names, ensure a default route "0.0.0.0/0"
+ is present in the Route table (on_up=1, no via/metric/table override).
+- If overwrite=True, it first deletes existing Route rows for those ifaces,
+ then inserts the defaults.
+- Writes **DB only**. It does not touch the kernel or /etc/iproute2/rt_tables.
+
+Why:
+- Your apply script reads Route rows and emits `ip -4 route replace … table <rtname>`.
+ Seeding a per-iface default route makes policy-routed tables usable out of the box.
+"""
+
+from __future__ import annotations
+import argparse
+import sqlite3
+from typing import Dict ,Iterable ,List ,Optional ,Sequence ,Tuple
+
+# import helper to open DB when run as CLI; the business API accepts a conn
+try:
+ import incommon as ic # type: ignore
+except Exception:
+ ic = None # ok when used as a lib
+
+
+def _iface_map(conn: sqlite3.Connection ,iface_names: Sequence[str]) -> Dict[str ,int]:
+ """Return {iface_name -> iface_id} for provided names (must exist)."""
+ if not iface_names:
+ return {}
+ ph = ",".join("?" for _ in iface_names)
+ sql = f"""SELECT id ,iface FROM Iface WHERE iface IN ({ph}) ORDER BY id;"""
+ rows = conn.execute(sql ,tuple(iface_names)).fetchall()
+ found = {str(name): int(iid) for (iid ,name) in rows}
+ missing = [n for n in iface_names if n not in found]
+ if missing:
+ raise RuntimeError(f"iface(s) not found: {', '.join(missing)}")
+ return found
+
+
+def _existing_defaults(conn: sqlite3.Connection ,iface_ids: Iterable[int]) -> Dict[int ,bool]:
+ """Return {iface_id -> True/False} whether a default route row already exists (on_up=1)."""
+ ids = list(iface_ids)
+ if not ids:
+ return {}
+ ph = ",".join("?" for _ in ids)
+ sql = f"""
+ SELECT iface_id ,COUNT(1)
+ FROM Route
+ WHERE iface_id IN ({ph})
+ AND cidr='0.0.0.0/0'
+ AND on_up=1
+ GROUP BY iface_id;
+ """
+ out: Dict[int ,bool] = {i: False for i in ids}
+ for iid ,cnt in conn.execute(sql ,tuple(ids)).fetchall():
+ out[int(iid)] = int(cnt) > 0
+ return out
+
+
+def seed_default_routes(
+ conn: sqlite3.Connection
+ ,iface_names: Sequence[str]
+ ,overwrite: bool = False
+ ,metric: Optional[int] = None
+) -> Tuple[int ,List[str]]:
+ """
+ Upsert per-iface default routes into Route.
+
+ Inserts rows:
+ (iface_id ,cidr='0.0.0.0/0' ,via=NULL ,table_name=NULL ,metric=<metric or NULL> ,on_up=1 ,on_down=0)
+ """
+ if not iface_names:
+ raise RuntimeError("no interfaces provided")
+
+ id_map = _iface_map(conn ,iface_names)
+ iface_ids = list(id_map.values())
+ notes: List[str] = []
+ inserted = 0
+
+ with conn:
+ if overwrite:
+ ph = ",".join("?" for _ in iface_ids)
+ conn.execute(f"DELETE FROM Route WHERE iface_id IN ({ph});" ,tuple(iface_ids))
+ notes.append(f"cleared existing Route rows for: {', '.join(iface_names)}")
+
+ exists = _existing_defaults(conn ,iface_ids)
+
+ for name in iface_names:
+ iid = id_map[name]
+ if exists.get(iid):
+ notes.append(f"keep: default route already present for {name}")
+ continue
+ conn.execute(
+ """
+ INSERT INTO Route(iface_id ,cidr ,via ,table_name ,metric ,on_up ,on_down
+ ,created_at ,updated_at)
+ VALUES( ? ,'0.0.0.0/0' ,NULL ,NULL ,? ,1 ,0
+ ,strftime('%Y-%m-%dT%H:%M:%SZ','now') ,strftime('%Y-%m-%dT%H:%M:%SZ','now'))
+ """
+ ,(iid ,metric)
+ )
+ inserted += 1
+ notes.append(f"add: default route 0.0.0.0/0 for {name}")
+
+ return (inserted ,notes)
+
+
+# ---- thin CLI for ad-hoc use ----
+
+def main(argv: Optional[Sequence[str]] = None) -> int:
+ ap = argparse.ArgumentParser(description="Seed per-iface default Route rows.")
+ ap.add_argument("ifaces" ,nargs="+")
+ ap.add_argument("--overwrite" ,action="store_true")
+ ap.add_argument("--metric" ,type=int ,default=None)
+ args = ap.parse_args(argv)
+
+ if ic is None:
+ print("error: cannot locate incommon.open_db() for CLI use")
+ return 2
+
+ with ic.open_db() as conn:
+ n ,notes = seed_default_routes(conn ,args.ifaces ,overwrite=args.overwrite ,metric=args.metric)
+ if notes:
+ print("\n".join(notes))
+ print(f"inserted: {n}")
+ return 0
+
+
+if __name__ == "__main__":
+ import sys
+ sys.exit(main())
--- /dev/null
+#!/usr/bin/env python3
+"""
+db_wipe.py
+
+Remove regular (non-directory) files in ./db, keeping the directory.
+
+Safety
+- Refuses to run if the target directory does not exist or its basename is not exactly "db".
+- Prints a plan, then asks "Are you sure? [y/N]" unless --force is used.
+- --dry-run prints what would be removed without deleting.
+- Hidden files (names starting with '.') are preserved by default; use --include-hidden to delete them too.
+
+Usage
+ ./db_wipe.py # plan + prompt, non-hidden files only, ./db next to this script
+ ./db_wipe.py --force # no prompt
+ ./db_wipe.py --dry-run # show what would be deleted
+ ./db_wipe.py --include-hidden
+ ./db_wipe.py --db /path/to/db
+"""
+
+from __future__ import annotations
+from pathlib import Path
+from typing import Iterable, List, Tuple
+import argparse
+import sys
+import os
+
+# ---------- business ----------
+
+def plan_db_wipe(db_dir: Path, include_hidden: bool = False) -> List[Path]:
+ """
+ Return a sorted list of file Paths (depth=1) to delete from db_dir.
+ """
+ if not db_dir.exists():
+ raise FileNotFoundError(f"not found: {db_dir}")
+ if not db_dir.is_dir():
+ raise NotADirectoryError(f"not a directory: {db_dir}")
+ if db_dir.name != "db":
+ raise RuntimeError(f"expected directory named 'db', got: {db_dir.name}")
+
+ def _is_hidden(p: Path) -> bool:
+ return p.name.startswith(".")
+
+ files = [p for p in db_dir.iterdir() if p.is_file()]
+ if not include_hidden:
+ files = [p for p in files if not _is_hidden(p)]
+
+ # Sort by name for stable output
+ return sorted(files, key=lambda p: p.name)
+
+
+def wipe_db(
+ db_dir: Path,
+ include_hidden: bool = False,
+ dry_run: bool = False,
+ assume_yes: bool = False,
+ _prompt_fn=input,
+) -> Tuple[int, List[str]]:
+ """
+ Delete planned files from db_dir. Returns (deleted_count, logs).
+ Does not prompt if assume_yes=True or dry_run=True.
+ """
+ targets = plan_db_wipe(db_dir, include_hidden=include_hidden)
+
+ logs: List[str] = []
+ script_dir = Path(__file__).resolve().parent
+
+ if not targets:
+ logs.append(f"db_wipe: no matching files in: {db_dir.relative_to(script_dir)}")
+ return (0, logs)
+
+ logs.append("db_wipe: plan")
+ for p in targets:
+ # Show path relative to script directory like the original
+ rel = p.resolve().relative_to(script_dir)
+ logs.append(f" delete: {rel}")
+
+ if dry_run:
+ logs.append("db_wipe: dry-run; no changes made")
+ return (0, logs)
+
+ if not assume_yes:
+ print("\n".join(logs))
+ try:
+ ans = _prompt_fn("Are you sure? [y/N] ").strip().lower()
+ except EOFError:
+ ans = ""
+ if ans not in ("y", "yes"):
+ logs.append("db_wipe: aborted")
+ return (0, logs)
+
+ deleted = 0
+ for p in targets:
+ try:
+ p.unlink(missing_ok=True) # py3.8+: if not available, catch FileNotFoundError
+ deleted += 1
+ except FileNotFoundError:
+ # Equivalent to rm -f
+ pass
+
+ rel_db = db_dir.resolve().relative_to(script_dir)
+ logs.append(f"db_wipe: deleted {deleted} file(s) from {rel_db}")
+ return (deleted, logs)
+
+
+# ---------- CLI wrapper ----------
+
+def _default_db_dir() -> Path:
+ return Path(__file__).resolve().parent / "db"
+
+def main(argv: list[str] | None = None) -> int:
+ ap = argparse.ArgumentParser(description="Remove regular files in ./db, keeping the directory.")
+ ap.add_argument("--db", default=str(_default_db_dir()), help="path to the db directory (default: ./db next to this script)")
+ ap.add_argument("--force", action="store_true", help="do not prompt for confirmation")
+ ap.add_argument("--dry-run", action="store_true", help="print what would be removed without deleting")
+ ap.add_argument("--include-hidden", action="store_true", help="include dotfiles (e.g., .gitignore)")
+ args = ap.parse_args(argv)
+
+ db_dir = Path(args.db)
+
+ try:
+ deleted, logs = wipe_db(
+ db_dir=db_dir,
+ include_hidden=args.include_hidden,
+ dry_run=args.dry_run,
+ assume_yes=args.force or args.dry_run,
+ )
+ if logs:
+ print("\n".join(logs))
+ return 0
+ except (FileNotFoundError, NotADirectoryError, RuntimeError) as e:
+ print(f"❌ {e}", file=sys.stderr)
+ return 1
+ except Exception as e:
+ print(f"❌ unexpected error: {e}", file=sys.stderr)
+ return 2
+
+if __name__ == "__main__":
+ sys.exit(main())
+++ /dev/null
-#!/usr/bin/env bash
-# Usage: db_wipe.sh [--force] [--dry-run] [--include-hidden]
-# Job: Remove regular non-hidden files in ./db (e.g., store, store-wal, store-shm), keeping the directory.
-# Safety:
-# - Refuses to run if ./db does not exist or is not named exactly "db".
-# - Prints a plan, then asks: "Are you sure? [y/N]" unless --force is used.
-# - --dry-run prints what would be removed without deleting.
-# - Hidden files (names starting with '.') are preserved by default (e.g., .gitignore).
-# Notes:
-# - Comments avoid possessive pronouns.
-
-set -euo pipefail
-
-SELF_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-DB_DIR="$SELF_DIR/db"
-
-FORCE=0
-DRYRUN=0
-INCLUDE_HIDDEN=0
-
-while (($#)); do
- case "$1" in
- --force) FORCE=1 ;;
- --dry-run) DRYRUN=1 ;;
- --include-hidden) INCLUDE_HIDDEN=1 ;;
- -h|--help) sed -n '2,30p' "$0"; exit 0 ;;
- *) echo "unknown option: $1" >&2; exit 2 ;;
- esac
- shift || true
-done
-
-# Guards
-[[ -d "$DB_DIR" ]] || { echo "❌ not found: $DB_DIR"; exit 1; }
-[[ "$(basename -- "$DB_DIR")" == "db" ]] || { echo "❌ expected directory named 'db', got: $(basename -- "$DB_DIR")"; exit 1; }
-
-# Build find expression
-if (( INCLUDE_HIDDEN )); then
- # include all regular files
- mapfile -t TARGETS < <(find "$DB_DIR" -maxdepth 1 -type f -print | sort)
-else
- # exclude dotfiles (preserve .gitignore and other hidden files)
- mapfile -t TARGETS < <(find "$DB_DIR" -maxdepth 1 -type f ! -name '.*' -print | sort)
-fi
-
-if ((${#TARGETS[@]}==0)); then
- echo "db_wipe: no matching files in: ${DB_DIR#$SELF_DIR/}"
- exit 0
-fi
-
-echo "db_wipe: plan"
-for f in "${TARGETS[@]}"; do
- echo " delete: ${f#$SELF_DIR/}"
-done
-
-if (( DRYRUN )); then
- echo "db_wipe: dry-run; no changes made"
- exit 0
-fi
-
-if (( ! FORCE )); then
- printf "Are you sure? [y/N] "
- read -r ans || true
- case "${ans,,}" in y|yes) ;; *) echo "db_wipe: aborted"; exit 0 ;; esac
-fi
-
-# Delete
-for f in "${TARGETS[@]}"; do
- rm -f -- "$f"
-done
-
-echo "db_wipe: deleted ${#TARGETS[@]} file(s) from ${DB_DIR#$SELF_DIR/}"
--- /dev/null
+#!/usr/bin/env python3
+"""
+deploy_StanleyPark.py — stop → install staged files → start (for selected ifaces)
+
+- Requires root. Exits after reporting *all* detected CLI/import errors.
+- Calls business functions directly:
+ * stop_clean_iface.stop_clean_ifaces(ifaces)
+ * install_staged_tree.install_staged_tree(stage_root, dest_root, create_dirs, skip_identical)
+ * start_iface.start_ifaces(ifaces)
+- If no ifaces provided on CLI, it discovers them from the stage tree.
+
+Usage:
+ sudo ./deploy_StanleyPark.py # discover ifaces from stage, stop→install→start
+ sudo ./deploy_StanleyPark.py x6 US # explicit iface list
+ sudo ./deploy_StanleyPark.py --no-stop # skip stop step
+ sudo ./deploy_StanleyPark.py --no-start # skip start step
+ sudo ./deploy_StanleyPark.py --stage ./stage --root / --create-dirs
+"""
+
+from __future__ import annotations
+from pathlib import Path
+from typing import List, Sequence, Tuple
+import argparse
+import os
+import sys
+import traceback
+
+ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(ROOT)) # ensure sibling modules importable
+
+# --- lightweight staged-iface discovery (duplicated here to avoid importing internals) ---
+def _discover_ifaces_from_stage(stage_root: Path) -> List[str]:
+ names = set()
+ # from /etc/wireguard/<iface>.conf
+ wg_dir = stage_root / "etc" / "wireguard"
+ if wg_dir.is_dir():
+ for p in wg_dir.glob("*.conf"):
+ names.add(p.stem)
+ # from /etc/systemd/system/wg-quick@<iface>.service.d/
+ sysd = stage_root / "etc" / "systemd" / "system"
+ if sysd.is_dir():
+ for d in sysd.glob("wg-quick@*.service.d"):
+ nm = d.name # wg-quick@IFACE.service.d
+ at = nm.find("@")
+ dot = nm.find(".service.d")
+ if at != -1 and dot != -1 and dot > at:
+ names.add(nm[at+1:dot])
+ return sorted(names)
+
+def _is_root() -> bool:
+ try:
+ return os.geteuid() == 0
+ except AttributeError:
+ # Non-POSIX: best effort
+ return False
+
+def _validate_iface_name(n: str) -> bool:
+ # conservative: letters, digits, dash, underscore (WireGuard allows more, but keep it safe)
+ import re
+ return bool(re.fullmatch(r"[A-Za-z0-9_-]{1,32}", n))
+
+def _collect_errors(args) -> Tuple[List[str], List[str]]:
+ """
+ Return (errors, ifaces). Does *not* raise.
+ """
+ errors: List[str] = []
+
+ # Root required
+ if not _is_root():
+ errors.append("must be run as root (sudo)")
+
+ # Stage root
+ stage_root = Path(args.stage)
+ if not stage_root.exists():
+ errors.append(f"stage path does not exist: {stage_root}")
+
+ # Import modules
+ inst_mod = None
+ stop_mod = None
+ start_mod = None
+ try:
+ import install_staged_tree as inst_mod # type: ignore
+ except Exception as e:
+ errors.append(f"failed to import install_staged_tree: {e}")
+ try:
+ import stop_clean_iface as stop_mod # type: ignore
+ except Exception as e:
+ errors.append(f"failed to import stop_clean_iface: {e}")
+ try:
+ import start_iface as start_mod # type: ignore
+ except Exception as e:
+ errors.append(f"failed to import start_iface: {e}")
+
+ # Business functions existence (only if imports worked)
+ if inst_mod is not None and not hasattr(inst_mod, "install_staged_tree"):
+ errors.append("install_staged_tree module missing function: install_staged_tree")
+ if stop_mod is not None and not hasattr(stop_mod, "stop_clean_ifaces"):
+ errors.append("stop_clean_iface module missing function: stop_clean_ifaces")
+ if start_mod is not None and not hasattr(start_mod, "start_ifaces"):
+ errors.append("start_iface module missing function: start_ifaces")
+
+ # Ifaces
+ ifaces: List[str]
+ if args.ifaces:
+ ifaces = list(dict.fromkeys(args.ifaces)) # dedup preserve order
+ else:
+ ifaces = _discover_ifaces_from_stage(stage_root)
+ if not ifaces:
+ errors.append("no interfaces provided and none discovered from stage")
+ else:
+ bad = [n for n in ifaces if not _validate_iface_name(n)]
+ if bad:
+ errors.append(f"invalid iface name(s): {', '.join(bad)}")
+
+ return (errors, ifaces)
+
+def deploy_StanleyPark(
+ ifaces: Sequence[str],
+ stage_root: Path,
+ dest_root: Path,
+ create_dirs: bool,
+ skip_identical: bool,
+ do_stop: bool,
+ do_start: bool,
+) -> int:
+ """
+ Orchestration: stop (optional) → install → start (optional).
+ """
+ # Late imports so unit tests can monkeypatch easily
+ import install_staged_tree as inst
+ import stop_clean_iface as stopm
+ import start_iface as startm
+
+ print(f"Deploy plan:\n ifaces: {', '.join(ifaces)}\n stage: {stage_root}\n root: {dest_root}\n")
+
+ # Stop
+ if do_stop:
+ print(f"Stopping: {' '.join(ifaces)}")
+ try:
+ stop_logs = stopm.stop_clean_ifaces(ifaces)
+ if isinstance(stop_logs, (list, tuple)):
+ for line in stop_logs:
+ print(line)
+ except Exception:
+ print("warn: stop_clean_ifaces raised an exception (continuing):")
+ traceback.print_exc()
+
+ # Install
+ print("\nInstalling staged artifacts…")
+ try:
+ logs, detected = inst.install_staged_tree(
+ stage_root=stage_root,
+ dest_root=dest_root,
+ create_dirs=create_dirs,
+ skip_identical=skip_identical,
+ )
+ for line in logs:
+ print(line)
+ except Exception:
+ print("❌ install failed with exception:", file=sys.stderr)
+ traceback.print_exc()
+ return 2
+
+ # Start
+ if do_start:
+ # Prefer explicit ifaces; fall back to what installer detected
+ start_list = list(ifaces) if ifaces else list(detected)
+ if not start_list:
+ print("\nNo interfaces to start (none detected).")
+ else:
+ print(f"\nStarting: {' '.join(start_list)}")
+ try:
+ start_logs = startm.start_ifaces(start_list)
+ if isinstance(start_logs, (list, tuple)):
+ for line in start_logs:
+ print(line)
+ except Exception:
+ print("warn: start_ifaces raised an exception:", file=sys.stderr)
+ traceback.print_exc()
+ return 2
+
+ print("\n✓ Deploy complete.")
+ return 0
+
+def main(argv: List[str] | None = None) -> int:
+ ap = argparse.ArgumentParser(description="Deploy staged WG artifacts for StanleyPark (stop→install→start).")
+ ap.add_argument("ifaces", nargs="*", help="interfaces to manage (default: discover from stage)")
+ ap.add_argument("--stage", default=str(ROOT / "stage"), help="stage root (default: ./stage)")
+ ap.add_argument("--root", default="/", help="destination root (default: /)")
+ ap.add_argument("--create-dirs", action="store_true", help="create missing parent directories")
+ ap.add_argument("--no-skip-identical", action="store_true", help="always replace even if content identical")
+ ap.add_argument("--no-stop", action="store_true", help="do not stop interfaces before install")
+ ap.add_argument("--no-start", action="store_true", help="do not start interfaces after install")
+ args = ap.parse_args(argv)
+
+ # Collect all errors up front
+ errors, ifaces = _collect_errors(args)
+ if errors:
+ print("❌ deploy preflight found issue(s):", file=sys.stderr)
+ for e in errors:
+ print(f" - {e}", file=sys.stderr)
+ return 2
+
+ # Proceed
+ return deploy_StanleyPark(
+ ifaces=ifaces,
+ stage_root=Path(args.stage),
+ dest_root=Path(args.root),
+ create_dirs=args.create_dirs,
+ skip_identical=(not args.no_skip_identical),
+ do_stop=(not args.no_stop),
+ do_start=(not args.no_start),
+ )
+
+if __name__ == "__main__":
+ sys.exit(main())
--- /dev/null
+
+1. create/update the client configuration files.
+
+ These are the configuration files for the machine called StanleyPark, which is on
+ our local network. (Yes, we capitalize popper nouns, and thus have some "bad names".)
+
+ db_init_StanleyPark.py
+ stage_StanleyPark
+ deploy_StanleyPark
+
+ They are in Python.
+
+2. Wipe the database and the stage.
+
+ Wiping the db will erase keys and any other client configurations. This does not effect already installed configuration files. Also, the database can always be rebuilt by running the client configuration files again.
+
+ ./db_wipe.py
+ ./stage_wipe.py
+
+3. Setup the database
+
+ ./db_init_StanleyPark
+
+4. setup the keys
+
+ ./key_generate StanleyPark.py
+ ./key_server_set.py <sever's public key>
+
+ to see the keys in the database
+
+ ./ls_key.py
+
+ if the database was wiped, it will be necessary to key_generate again. Currently
+ there is one client machine key pair.
+
+5. stage the configuration files to be installed
+
+ ./stage_StanleyPark
+
+ check them make sure they are what you want
+
+6. install the staged files
+
+ ./deploy_StanlwayPark
+
+
+The goal here is work towards each subu as a container, with its networking tunneled
+to the specified interface. Perhaps the configuration scripts should be subu based instead of client machine based. Perhaps in the next version.
+
+
+
--- /dev/null
+#!/usr/bin/env python3
+# inspect_client_public_key.py — show the client's WireGuard public key for one iface
+# Sources checked (in this order): DB, staged conf, installed conf, kernel
+# The “client public key” is generated locally from the client’s PrivateKey and must be
+# copied to the **server** as the peer’s PublicKey in the server’s WireGuard config.
+
+from __future__ import annotations
+from pathlib import Path
+from typing import List, Optional, Tuple
+import argparse
+import os
+import subprocess
+import sqlite3
+import sys
+
+# Project helper providing DB_PATH and open_db()
+import incommon as ic
+
+ROOT = Path(__file__).resolve().parent
+DEFAULT_STAGE = ROOT / "stage"
+LIVE_WG_DIR = Path("/etc/wireguard")
+
+def _is_root() -> bool:
+ return os.geteuid() == 0
+
+def _format_table(headers: List[str], rows: List[Tuple]) -> str:
+ if not rows:
+ return "(none)"
+ cols = list(zip(*([headers] + [[("" if c is None else str(c)) for c in r] for r in rows])))
+ widths = [max(len(x) for x in col) for col in cols]
+ def line(r): return " ".join(f"{str(c):<{w}}" for c, w in zip(r, widths))
+ out = [line(headers), line(tuple("-"*w for w in widths))]
+ for r in rows:
+ out.append(line(r))
+ return "\n".join(out)
+
+def _read_conf_private_key(conf_path: Path) -> Optional[str]:
+ """Return the PrivateKey value from a wg conf (first [Interface] block), or None."""
+ try:
+ txt = conf_path.read_text()
+ except FileNotFoundError:
+ return None
+ section = None
+ for raw in txt.splitlines():
+ line = raw.strip()
+ if not line or line.startswith("#") or line.startswith(";"):
+ continue
+ if line.startswith("[") and line.endswith("]"):
+ section = line[1:-1].strip()
+ continue
+ if section == "Interface":
+ if line.lower().startswith("privatekey"):
+ parts = line.split("=", 1)
+ if len(parts) == 2:
+ val = parts[1].strip()
+ return val if val else None
+ return None
+
+def _pub_from_private_key(priv: str) -> Optional[str]:
+ """Compute public key from a WireGuard base64 private key using `wg pubkey`."""
+ if not priv:
+ return None
+ try:
+ cp = subprocess.run(
+ ["wg", "pubkey"],
+ input=(priv + "\n").encode("utf-8"),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ check=True,
+ )
+ pub = cp.stdout.decode("utf-8", "replace").strip()
+ return pub or None
+ except (subprocess.CalledProcessError, FileNotFoundError):
+ return None
+
+def _kernel_iface_public_key(iface: str) -> Optional[str]:
+ try:
+ cp = subprocess.run(
+ ["wg", "show", iface, "public-key"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ check=True,
+ )
+ k = cp.stdout.decode("utf-8", "replace").strip()
+ return k or None
+ except (subprocess.CalledProcessError, FileNotFoundError):
+ return None
+
+def _db_client_public_key(conn: sqlite3.Connection, iface: str) -> Optional[str]:
+ row = conn.execute("SELECT public_key FROM Iface WHERE iface=? LIMIT 1;", (iface,)).fetchone()
+ if not row:
+ return None
+ k = row[0]
+ return k if k else None
+
+def _rel_from_stage(path: Path, stage_root: Path) -> str:
+ """Return a short, stage-relative display path when under stage_root."""
+ try:
+ rel = path.relative_to(stage_root)
+ return str(rel)
+ except ValueError:
+ return str(path)
+
+def _gather(iface: str, stage_root: Path) -> Tuple[List[Tuple[str, str, str]], List[str]]:
+ """
+ Return (rows, notes)
+ rows: list of (source, location, public_key or "(missing)")
+ """
+ notes: List[str] = []
+
+ # DB
+ db_pub: Optional[str] = None
+ if ic.DB_PATH.exists():
+ try:
+ with ic.open_db() as conn:
+ db_pub = _db_client_public_key(conn, iface)
+ except sqlite3.Error as e:
+ notes.append(f"DB error: {e}")
+ else:
+ notes.append(f"DB not found at {ic.DB_PATH}")
+
+ # staged conf -> derive pub from PrivateKey
+ staged_conf = stage_root / "etc" / "wireguard" / f"{iface}.conf"
+ staged_priv = _read_conf_private_key(staged_conf)
+ staged_pub = _pub_from_private_key(staged_priv) if staged_priv else None
+ if staged_priv is None and staged_conf.exists():
+ notes.append(f"staged conf present but PrivateKey missing: { _rel_from_stage(staged_conf, stage_root) }")
+
+ # live conf -> derive pub from PrivateKey
+ live_conf = LIVE_WG_DIR / f"{iface}.conf"
+ live_priv = _read_conf_private_key(live_conf)
+ live_pub = _pub_from_private_key(live_priv) if live_priv else None
+ if live_conf.exists() and live_priv is None:
+ notes.append(f"installed conf present but PrivateKey missing: {live_conf}")
+
+ # kernel
+ kern_pub = _kernel_iface_public_key(iface)
+
+ rows: List[Tuple[str, str, str]] = []
+ rows.append(("DB", f"Iface.public_key[{iface}]", db_pub or "(missing)"))
+ rows.append(("Stage", _rel_from_stage(staged_conf, stage_root),
+ staged_pub or ("(missing)" if not staged_conf.exists() else "(could not derive)")))
+ rows.append(("Installed", str(live_conf),
+ live_pub or ("(missing)" if not live_conf.exists() else "(could not derive)")))
+ rows.append(("Kernel", f"wg show {iface} public-key", kern_pub or "(missing)"))
+
+ # Quick consistency summary
+ present = [v for _s, _loc, v in rows if not v.startswith("(")]
+ if len(present) >= 2:
+ all_same = all(v == present[0] for v in present[1:])
+ if all_same:
+ notes.append("All present sources agree.")
+ else:
+ notes.append("Mismatch detected between sources.")
+ elif len(present) == 1:
+ notes.append("Only one source has a key (cannot check consistency).")
+ else:
+ notes.append("No source has a client public key.")
+
+ return (rows, notes)
+
+def inspect_client_public_key(iface: str, stage_root: Optional[Path] = None) -> str:
+ """
+ Business function: returns a formatted report string.
+ """
+ sr = stage_root or DEFAULT_STAGE
+ rows, notes = _gather(iface, sr)
+
+ header = (
+ f"Client public key inspection for iface '{iface}'\n"
+ "This public key is generated locally from the client’s PrivateKey and must be\n"
+ "installed on the *server* as the peer’s PublicKey in the server’s WireGuard config.\n"
+ )
+ table = _format_table(["source", "where", "public_key"], rows)
+ if notes:
+ note_block = "\nNotes:\n- " + "\n- ".join(notes)
+ else:
+ note_block = ""
+ return f"{header}\n{table}\n{note_block}\n"
+
+def main(argv: Optional[List[str]] = None) -> int:
+ ap = argparse.ArgumentParser(
+ description="Inspect the client’s WireGuard public key for a single interface."
+ )
+ # Make iface optional so we can aggregate errors ourselves
+ ap.add_argument("iface", nargs="?", help="interface name (e.g., x6)")
+ ap.add_argument("--stage-root", default=str(DEFAULT_STAGE), help="stage directory (default: ./stage)")
+ args = ap.parse_args(argv)
+
+ # Aggregate invocation errors
+ errors: List[str] = []
+ if not _is_root():
+ errors.append("must run as root (needs access to /etc/wireguard and wg)")
+ if not args.iface:
+ errors.append("missing required positional argument: iface")
+ if args.stage_root:
+ sr = Path(args.stage_root)
+ if not sr.exists():
+ errors.append(f"--stage-root does not exist: {sr}")
+ elif not sr.is_dir():
+ errors.append(f"--stage-root is not a directory: {sr}")
+
+ if errors:
+ ap.print_usage(sys.stderr)
+ print(f"{ap.prog}: error: " + "; ".join(errors), file=sys.stderr)
+ return 2
+
+ try:
+ report = inspect_client_public_key(args.iface, Path(args.stage_root))
+ print(report, end="")
+ return 0
+ except Exception as e:
+ print(f"❌ {e}", file=sys.stderr)
+ return 2
+
+if __name__ == "__main__":
+ sys.exit(main())
+++ /dev/null
-#!/usr/bin/env python3
-"""
-install_staged_tree.py
-
-Given:
- - A staged tree (default: ./stage) containing:
- /usr/local/bin/apply_ip_state.sh
- /etc/wireguard/*.conf
- /etc/systemd/wg-quick@IFACE.service.d/*.conf
- /etc/iproute2/rt_tables
- - A destination root (default: /) whose *parent directories already exist*
-
-Does:
- - For each whitelisted staged file:
- * if a target already exists, copy it *back into the stage* as a timestamped backup
- * atomically replace target with staged version
- * set root:root ownership and deterministic permissions (see MODE_MAP)
- - Optionally `systemctl daemon-reload` and restart provided wg-quick@IFACE units
-
-Returns:
- - Exit 0 on success; non-zero on error
- - Prints a concise log of actions
-
-Errors:
- - Fails if a target parent directory is missing (unless --create-dirs is given)
- - Fails on any copy/permission error and reports which path caused it
-"""
-
-from __future__ import annotations
-from pathlib import Path
-from typing import Dict ,Iterable ,List ,Optional ,Sequence ,Tuple
-import argparse
-import datetime as dt
-import hashlib
-import os
-import shutil
-import subprocess
-import sys
-
-ROOT = Path(__file__).resolve().parent
-DEFAULT_STAGE = ROOT / "stage"
-
-# Whitelist → permissions
-# (relative glob inside stage) → (relative dest base, file mode)
-MODE_MAP: Dict[str,Tuple[str,int]] = {
- "usr/local/bin/*": ("usr/local/bin",0o500) # scripts: rx for root only
- , "etc/wireguard/*.conf": ("etc/wireguard",0o600) # WG confs
- , "etc/systemd/wg-quick@" : ("etc/systemd",0o644) # handled per-dropin below
- , "etc/iproute2/rt_tables": ("etc/iproute2",0o644) # route tables file
-}
-
-def _sha256(path: Path) -> str:
- h = hashlib.sha256()
- with path.open("rb") as f:
- for chunk in iter(lambda: f.read(1<<20), b""):
- h.update(chunk)
- return h.hexdigest()
-
-def _iter_dropins(stage_root: Path) -> List[Tuple[Path,int]]:
- """Return [(relpath,mode)] for systemd wg-quick drop-ins."""
- out: List[Tuple[Path,int]] = []
- base = stage_root / "etc" / "systemd"
- if not base.exists():
- return out
- for p in base.rglob("wg-quick@*.service.d/*.conf"):
- rel = p.relative_to(stage_root)
- out.append((rel,0o644))
- return out
-
-def _gather_stage_files(stage_root: Path) -> List[Tuple[Path,int]]:
- """Resolve whitelist into [(relpath,mode)]."""
- items: List[Tuple[Path,int]] = []
- # explicit patterns
- for pat,(_dest_base,mode) in MODE_MAP.items():
- if pat.endswith("@"): # systemd base marker handled separately
- continue
- for p in (stage_root / pat).parent.glob(Path(pat).name):
- rel = p.relative_to(stage_root)
- items.append((rel,mode))
- # systemd drop-ins
- items += _iter_dropins(stage_root)
- # de-dup in order
- seen = set()
- uniq: List[Tuple[Path,int]] = []
- for rel,mode in items:
- if rel not in seen:
- uniq.append((rel,mode))
- seen.add(rel)
- return uniq
-
-def _ensure_parents(dest_root: Path ,rel: Path ,create: bool) -> None:
- parent = (dest_root / rel).parent
- if parent.exists():
- return
- if not create:
- raise RuntimeError(f"missing parent directory: {parent}")
- parent.mkdir(parents=True,exist_ok=True)
-
-def _backup_existing_to_stage(stage_root: Path ,dest_root: Path ,rel: Path) -> Optional[Path]:
- """If target exists, copy it back into stage/_backups/<ts>/<rel> and return backup path."""
- target = dest_root / rel
- if not target.exists():
- return None
- ts = dt.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
- backup = stage_root / "_backups" / ts / rel
- backup.parent.mkdir(parents=True,exist_ok=True)
- shutil.copy2(target,backup)
- return backup
-
-def _atomic_install(src: Path ,dst: Path ,mode: int) -> None:
- tmp = dst.with_suffix(dst.suffix + ".tmp")
- # copy *bytes*, then set perms/owner, then atomic replace
- shutil.copyfile(src,tmp)
- os.chmod(tmp,mode)
- try:
- os.chown(tmp,0,0) # root:root
- except PermissionError:
- # setuid root expected; if not root, we still proceed for dry-run contexts
- pass
- os.replace(tmp,dst)
-
-def _maybe_daemon_reload(perform: bool) -> None:
- if not perform:
- return
- subprocess.run(
- ["systemctl","daemon-reload"]
- ,check=False
- ,stdout=subprocess.DEVNULL
- ,stderr=subprocess.DEVNULL
- )
-
-def _maybe_restart_ifaces(ifaces: Sequence[str]) -> None:
- for iface in ifaces:
- unit = f"wg-quick@{iface}.service"
- subprocess.run(
- ["systemctl","restart",unit]
- ,check=False
- ,stdout=subprocess.DEVNULL
- ,stderr=subprocess.DEVNULL
- )
-
-def install_staged_tree(
- stage_root: Path
- ,dest_root: Path
- ,create_dirs: bool = False
- ,skip_identical: bool = True
- ,daemon_reload: bool = False
- ,restart_ifaces: Sequence[str] = ()
-) -> List[str]:
- """
- Core business function.
-
- Given:
- stage_root, dest_root, flags
- Does:
- safe, deterministic copy with backups and explicit perms
- Returns:
- list of log lines
- """
- # Do not rely on process umask; set restrictive default, then override per-file.
- old_umask = os.umask(0o077)
- logs: List[str] = []
- try:
- staged = _gather_stage_files(stage_root)
- if not staged:
- raise RuntimeError("nothing to install (stage is empty or whitelist didn’t match)")
-
- for rel,mode in staged:
- src = stage_root / rel
- dst = dest_root / rel
-
- _ensure_parents(dest_root,rel,create_dirs)
-
- backup = _backup_existing_to_stage(stage_root,dest_root,rel)
- if backup:
- logs.append(f"backup: {dst} -> {backup}")
-
- if skip_identical and dst.exists():
- try:
- if _sha256(src) == _sha256(dst):
- logs.append(f"identical: skip {rel}")
- continue
- except Exception:
- pass
-
- _atomic_install(src,dst,mode)
- logs.append(f"install: {rel} (mode {oct(mode)})")
-
- if daemon_reload:
- _maybe_daemon_reload(True)
- logs.append("systemctl: daemon-reload")
-
- if restart_ifaces:
- _maybe_restart_ifaces(restart_ifaces)
- logs.append(f"systemctl: restart wg-quick@{','.join(restart_ifaces)}")
-
- return logs
- finally:
- os.umask(old_umask)
-
-def _require_root() -> None:
- if os.geteuid() != 0:
- raise RuntimeError("must run as root (installer sets ownership/permissions)")
-
-def main(argv: Optional[Sequence[str]] = None) -> int:
- ap = argparse.ArgumentParser(description="Install staged artifacts into a target root (root-only).")
- ap.add_argument("--stage" ,default=str(DEFAULT_STAGE))
- ap.add_argument("--root" ,default="/")
- ap.add_argument("--create-dirs" ,action="store_true" ,help="create missing parent directories")
- ap.add_argument("--no-skip-identical" ,action="store_true" ,help="always replace even if content identical")
- ap.add_argument("--daemon-reload" ,action="store_true" ,help="run systemctl daemon-reload after install")
- ap.add_argument("--restart-ifaces" ,nargs="*" ,default=[] ,help="optionally restart these wg-quick@IFACE units")
- args = ap.parse_args(argv)
-
- try:
- _require_root()
- logs = install_staged_tree(
- stage_root=Path(args.stage)
- ,dest_root=Path(args.root)
- ,create_dirs=args.create_dirs
- ,skip_identical=(not args.no_skip_identical)
- ,daemon_reload=args.daemon_reload
- ,restart_ifaces=args.restart_ifaces
- )
- for line in logs:
- print(line)
- return 0
- except Exception as e:
- print(f"❌ install failed: {e}",file=sys.stderr)
- return 2
-
-if __name__ == "__main__":
- sys.exit(main())
--- /dev/null
+#!/usr/bin/env python3
+"""
+install_staged_tree.py
+
+A dumb installer: copy staged files into the target root with backups and
+deterministic permissions. No systemd stop/start, no daemon-reload.
+
+Given:
+ - A staged tree (default: ./stage) containing any of:
+ /usr/local/bin/apply_ip_state.sh
+ /etc/wireguard/*.conf
+ /etc/systemd/system/wg-quick@IFACE.service.d/*.conf
+ /etc/iproute2/rt_tables
+ - A destination root (default: /). Parent dirs may be created with --create-dirs.
+
+Does:
+ - For each whitelisted staged file:
+ * if a target already exists, copy it back into the stage as a timestamped backup
+ * atomically replace target with staged version
+ * set root:root ownership (best-effort) and explicit permissions
+ - Prints a summary and suggests next steps (e.g., ./start_iface.py <ifaces>)
+
+Returns:
+ - Exit 0 on success; non-zero on error
+"""
+
+from __future__ import annotations
+from pathlib import Path
+from typing import Dict, Iterable, List, Optional, Sequence, Tuple
+import argparse
+import datetime as dt
+import hashlib
+import os
+import shutil
+import sys
+
+ROOT = Path(__file__).resolve().parent
+DEFAULT_STAGE = ROOT / "stage"
+
+# Whitelisted install targets → mode
+# (These are *relative* to the stage root)
+MODE_RULES: List[Tuple[str, int]] = [
+ ("usr/local/bin", 0o500), # files under here (scripts)
+ ("etc/wireguard", 0o600), # *.conf
+ ("etc/systemd/system", 0o644), # wg-quick@*.service.d/*.conf
+ ("etc/iproute2", 0o644), # rt_tables
+]
+
+def _sha256(path: Path) -> str:
+ h = hashlib.sha256()
+ with path.open("rb") as f:
+ for chunk in iter(lambda: f.read(1<<20), b""):
+ h.update(chunk)
+ return h.hexdigest()
+
+def _ensure_parents(dest_root: Path, rel: Path, create: bool) -> None:
+ parent = (dest_root / rel).parent
+ if parent.exists():
+ return
+ if not create:
+ raise RuntimeError(f"missing parent directory: {parent}")
+ parent.mkdir(parents=True, exist_ok=True)
+
+def _backup_existing_to_stage(stage_root: Path, dest_root: Path, rel: Path) -> Optional[Path]:
+ """If target exists, copy it back into stage/_backups/<ts>/<rel> and return backup path."""
+ target = dest_root / rel
+ if not target.exists():
+ return None
+ ts = dt.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
+ backup = stage_root / "_backups" / ts / rel
+ backup.parent.mkdir(parents=True, exist_ok=True)
+ shutil.copy2(target, backup)
+ return backup
+
+def _atomic_install(src: Path, dst: Path, mode: int) -> None:
+ tmp = dst.with_suffix(dst.suffix + ".tmp")
+ shutil.copyfile(src, tmp)
+ os.chmod(tmp, mode)
+ try:
+ os.chown(tmp, 0, 0) # best-effort; may fail if not root
+ except PermissionError:
+ pass
+ os.replace(tmp, dst)
+
+def _mode_for_rel(rel: Path) -> Optional[int]:
+ """Choose a mode based on the relative path bucket."""
+ s = str(rel)
+ if s.startswith("usr/local/bin/"):
+ return 0o500
+ if s.startswith("etc/wireguard/") and rel.suffix == ".conf":
+ return 0o600
+ if s == "etc/iproute2/rt_tables":
+ return 0o644
+ if s.startswith("etc/systemd/system/") and s.endswith(".conf"):
+ return 0o644
+ return None
+
+def _iter_stage_targets(stage_root: Path) -> List[Path]:
+ """Return a list of *relative* paths under stage that match our whitelist."""
+ rels: List[Path] = []
+
+ # /usr/local/bin/*
+ bin_dir = stage_root / "usr" / "local" / "bin"
+ if bin_dir.is_dir():
+ for p in sorted(bin_dir.glob("*")):
+ if p.is_file():
+ rels.append(p.relative_to(stage_root))
+
+ # /etc/wireguard/*.conf
+ wg_dir = stage_root / "etc" / "wireguard"
+ if wg_dir.is_dir():
+ for p in sorted(wg_dir.glob("*.conf")):
+ rels.append(p.relative_to(stage_root))
+
+ # /etc/systemd/system/wg-quick@*.service.d/*.conf
+ sysd_dir = stage_root / "etc" / "systemd" / "system"
+ if sysd_dir.is_dir():
+ for p in sorted(sysd_dir.rglob("wg-quick@*.service.d/*.conf")):
+ rels.append(p.relative_to(stage_root))
+
+ # /etc/iproute2/rt_tables
+ rt = stage_root / "etc" / "iproute2" / "rt_tables"
+ if rt.is_file():
+ rels.append(rt.relative_to(stage_root))
+
+ return rels
+
+def _discover_ifaces_from_stage(stage_root: Path) -> List[str]:
+ """Peek into staged artifacts to guess iface names (for friendly next-steps)."""
+ names = set()
+
+ # from /etc/wireguard/<iface>.conf
+ wg_dir = stage_root / "etc" / "wireguard"
+ if wg_dir.is_dir():
+ for p in wg_dir.glob("*.conf"):
+ names.add(p.stem)
+
+ # from /etc/systemd/system/wg-quick@<iface>.service.d/
+ sysd = stage_root / "etc" / "systemd" / "system"
+ if sysd.is_dir():
+ for d in sysd.glob("wg-quick@*.service.d"):
+ name = d.name
+ # name looks like: wg-quick@X.service.d
+ at = name.find("@")
+ dot = name.find(".service.d")
+ if at != -1 and dot != -1 and dot > at:
+ names.add(name[at+1:dot])
+
+ return sorted(names)
+
+def install_staged_tree(
+ stage_root: Path,
+ dest_root: Path,
+ create_dirs: bool = False,
+ skip_identical: bool = True,
+) -> Tuple[List[str], List[str]]:
+ """
+ Copy files from stage_root to dest_root.
+ Returns (logs, detected_ifaces).
+ """
+ old_umask = os.umask(0o077)
+ logs: List[str] = []
+ try:
+ staged = _iter_stage_targets(stage_root)
+ if not staged:
+ raise RuntimeError("nothing to install (stage is empty or whitelist didn’t match)")
+
+ for rel in staged:
+ src = stage_root / rel
+ dst = dest_root / rel
+
+ mode = _mode_for_rel(rel)
+ if mode is None:
+ logs.append(f"skip (not whitelisted): {rel}")
+ continue
+
+ _ensure_parents(dest_root, rel, create_dirs)
+
+ backup = _backup_existing_to_stage(stage_root, dest_root, rel)
+ if backup:
+ logs.append(f"backup: {dst} -> {backup}")
+
+ if skip_identical and dst.exists():
+ try:
+ if _sha256(src) == _sha256(dst):
+ logs.append(f"identical: skip {rel}")
+ continue
+ except Exception:
+ pass
+
+ _atomic_install(src, dst, mode)
+ logs.append(f"install: {rel} (mode {oct(mode)})")
+
+ ifaces = _discover_ifaces_from_stage(stage_root)
+ return (logs, ifaces)
+ finally:
+ os.umask(old_umask)
+
+def _require_root(allow_nonroot: bool) -> None:
+ if not allow_nonroot and os.geteuid() != 0:
+ raise RuntimeError("must run as root (use --force-nonroot to override)")
+
+def main(argv: Optional[Sequence[str]] = None) -> int:
+ ap = argparse.ArgumentParser(description="Install staged artifacts into a target root. No service control.")
+ ap.add_argument("--stage", default=str(DEFAULT_STAGE))
+ ap.add_argument("--root", default="/")
+ ap.add_argument("--create-dirs", action="store_true", help="create missing parent directories")
+ ap.add_argument("--no-skip-identical", action="store_true", help="always replace even if content identical")
+ ap.add_argument("--force-nonroot", action="store_true", help="allow non-root install (ownership may be wrong)")
+ args = ap.parse_args(argv)
+
+ try:
+ _require_root(allow_nonroot=args.force_nonroot)
+ logs, ifaces = install_staged_tree(
+ stage_root=Path(args.stage),
+ dest_root=Path(args.root),
+ create_dirs=args.create_dirs,
+ skip_identical=(not args.no_skip_identical),
+ )
+ for line in logs:
+ print(line)
+
+ # Summary + suggested next steps
+ print("\n=== Summary ===")
+ print(f"Installed {sum(1 for l in logs if l.startswith('install:'))} file(s).")
+ if ifaces:
+ lst = " ".join(ifaces)
+ print(f"Detected interfaces from stage: {lst}")
+ print(f"\nNext steps:")
+ print(f" # (optional) verify configs")
+ print(f" sudo wg-quick strip /etc/wireguard/{ifaces[0]}.conf >/dev/null 2>&1 || true")
+ print(f"\n # start interfaces")
+ print(f" sudo ./start_iface.py {lst}")
+ else:
+ print("No interfaces detected in staged artifacts.")
+ print("\nNext steps:")
+ print(" # start your interface(s)")
+ print(" sudo ./start_iface.py <iface> [more ifaces]")
+ return 0
+ except Exception as e:
+ print(f"❌ install failed: {e}", file=sys.stderr)
+ return 2
+
+if __name__ == "__main__":
+ sys.exit(main())
def list_client_keys(conn: sqlite3.Connection, iface: str | None, banner=False) -> str:
if banner:
- print("\n=== Public keys generated locally by client, probably by using `db_update_client_key`===")
+ print("\n=== Public keys generated locally by client, probably by using `key_client_generate.py`===")
rows = conn.execute(
"SELECT iface, public_key AS client_public_key "
"FROM Iface "
--- /dev/null
+
+*
+!.gitignore
+
+++ /dev/null
-[Service]
-ExecStartPost=+/usr/local/bin/apply_ip_state.sh US
+++ /dev/null
-[Service]
-ExecStartPost=+/usr/local/bin/apply_ip_state.sh x6
[Interface]
-PrivateKey = ACd0vEyoZejb+WkXL1LcheHAYm2oRBbw52dJB5+tmUQ=
+PrivateKey = 0OUqldVHE0GSUM2XUw4o9kgc/OR6smcwED6Wk1HJgGQ=
Table = off
# ListenPort = 51820
[Interface]
-PrivateKey = ACd0vEyoZejb+WkXL1LcheHAYm2oRBbw52dJB5+tmUQ=
+PrivateKey = 0OUqldVHE0GSUM2XUw4o9kgc/OR6smcwED6Wk1HJgGQ=
Table = off
# ListenPort = 51820
exists_iface(){ ip -o link show dev "$1" >/dev/null 2>&1; }
-ensure_addr(){
+# Reset address: delete the exact CIDR if present, then add it back.
+reset_addr(){
local iface=$1; local cidr=$2
- if ip -4 -o addr show dev "$iface" | awk '{print $4}' | grep -Fxq "$cidr"; then
- logger "addr ok: $iface $cidr"
+ ip -4 addr del "$cidr" dev "$iface" >/dev/null 2>&1 || true
+ if ip -4 addr add "$cidr" dev "$iface"; then
+ logger "addr set: $iface $cidr"
else
- ip -4 addr add "$cidr" dev "$iface"
- logger "addr add: $iface $cidr"
+ logger "addr add failed (non-fatal): $iface $cidr"
fi
}
+# Ensure route using replace; log but do not fail the unit if kernel says 'exists'.
ensure_route(){
local table=$1; local cidr=$2; local dev=$3; local via=${4:-}; local metric=${5:-}
if [ -n "$via" ] && [ -n "$metric" ]; then
- ip -4 route replace "$cidr" via "$via" dev "$dev" table "$table" metric "$metric"
+ if ip -4 route replace "$cidr" via "$via" dev "$dev" table "$table" metric "$metric" 2>/dev/null; then
+ logger "route ensure: table=$table cidr=$cidr dev=$dev via=$via metric=$metric"
+ else
+ logger "route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev via=$via metric=$metric"
+ fi
elif [ -n "$via" ]; then
- ip -4 route replace "$cidr" via "$via" dev "$dev" table "$table"
+ if ip -4 route replace "$cidr" via "$via" dev "$dev" table "$table" 2>/dev/null; then
+ logger "route ensure: table=$table cidr=$cidr dev=$dev via=$via"
+ else
+ logger "route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev via=$via"
+ fi
elif [ -n "$metric" ]; then
- ip -4 route replace "$cidr" dev "$dev" table "$table" metric "$metric"
+ if ip -4 route replace "$cidr" dev "$dev" table "$table" metric "$metric" 2>/dev/null; then
+ logger "route ensure: table=$table cidr=$cidr dev=$dev metric=$metric"
+ else
+ logger "route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev metric=$metric"
+ fi
else
- ip -4 route replace "$cidr" dev "$dev" table "$table"
+ if ip -4 route replace "$cidr" dev "$dev" table "$table" 2>/dev/null; then
+ logger "route ensure: table=$table cidr=$cidr dev=$dev"
+ else
+ logger "route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev"
+ fi
fi
- logger "route ensure: table=$table cidr=$cidr dev=$dev${via:+ via=$via}${metric:+ metric=$metric}"
}
-add_ip_rule_if_absent(){
- local needle=$1; shift
- if ! ip -4 rule show | grep -F -q -- "$needle"; then
- ip -4 rule add "$@"
- logger "rule add: $*"
+# Reset a policy rule by numeric preference: delete-by-pref, then add.
+reset_IP_rule(){
+ # Usage: reset_IP_rule <pref> <rule-args...>
+ local pref=$1; shift
+ ip -4 rule del pref "$pref" >/dev/null 2>&1 || true
+ if ip -4 rule add "$@" pref "$pref"; then
+ logger "rule set: pref=$pref $*"
else
- logger "rule ok: $needle"
+ logger "rule add failed (non-fatal): pref=$pref $*"
fi
}
if want_iface x6; then
- if exists_iface x6; then ensure_addr x6 10.8.0.2/32; else logger "skip: iface missing: x6"; fi
+ if exists_iface x6; then reset_addr x6 10.8.0.2/32; else logger "skip: iface missing: x6"; fi
fi
if want_iface US; then
- if exists_iface US; then ensure_addr US 10.0.0.1/32; else logger "skip: iface missing: US"; fi
+ if exists_iface US; then reset_addr US 10.0.0.1/32; else logger "skip: iface missing: US"; fi
fi
if want_iface x6; then
- add_ip_rule_if_absent "from 10.8.0.2/32 lookup x6" from "10.8.0.2/32" lookup "x6" pref 17000
+ if exists_iface x6; then ensure_route "x6" "0.0.0.0/0" "x6" "" ""; else logger "skip: iface missing: x6"; fi
+fi
+if want_iface US; then
+ if exists_iface US; then ensure_route "US" "0.0.0.0/0" "US" "" ""; else logger "skip: iface missing: US"; fi
+fi
+if want_iface x6; then
+ reset_IP_rule 17010 from "10.8.0.2/32" lookup "x6"
fi
if want_iface x6; then
- add_ip_rule_if_absent "uidrange 2018-2018 lookup x6" uidrange "2018-2018" lookup "x6" pref 17010
+ reset_IP_rule 17011 uidrange "2018-2018" lookup "x6"
fi
if want_iface US; then
- add_ip_rule_if_absent "from 10.0.0.1/32 lookup US" from "10.0.0.1/32" lookup "US" pref 17000
+ reset_IP_rule 17020 from "10.0.0.1/32" lookup "US"
fi
if want_iface US; then
- add_ip_rule_if_absent "uidrange 2017-2017 lookup US" uidrange "2017-2017" lookup "US" pref 17010
+ reset_IP_rule 17021 uidrange "2017-2017" lookup "US"
fi
-add_ip_rule_if_absent "from 10.0.0.0/24 prohibit" from "10.0.0.0/24" prohibit pref 18050
+reset_IP_rule 18050 from "10.0.0.0/24" prohibit
Given:
- A SQLite DB (schema you’ve defined), with:
- * Iface(id ,iface ,local_address_cidr ,rt_table_name)
- * v_iface_effective(id ,rt_table_name_eff ,local_address_cidr)
- * Route(iface_id ,cidr ,via ,table_name ,metric ,on_up ,on_down)
- * "User"(iface_id ,username ,uid) — table formerly User_Binding
- * Meta(key='subu_cidr' ,value)
+ * Iface(id, iface, local_address_cidr, rt_table_name, rt_table_id)
+ * v_iface_effective(id, rt_table_name_eff, local_address_cidr)
+ * Route(iface_id, cidr, via, table_name, metric, on_up, on_down)
+ * "User"(iface_id, username, uid) — table formerly User_Binding
+ * Meta(key='subu_cidr', value)
- A list of interface names to include (e.g., ["x6","US"]).
Does:
- Reads DB once and *synthesizes a single* idempotent runtime script
that, for the selected interfaces, on each `wg-quick@IFACE` start:
- 1) ensures IPv4 addresses exist on the iface (if present in DB)
+ 1) resets IPv4 addresses on the iface (delete-if-present, then add)
2) ensures all configured routes exist (using `ip -4 route replace`)
- 3) ensures policy rules exist for src-cidr ,uidrange ,and a `prohibit`
+ 3) resets policy rules by preference number (delete-by-pref, then add)
+ with **per-iface prefs** to avoid collisions.
- Stages that script under: stage/usr/local/bin/<script_name>
- Stages per-iface systemd drop-ins:
- stage/etc/systemd/wg-quick@IFACE.service.d/<prio>-postup-ip-state.conf
+ stage/etc/systemd/system/wg-quick@IFACE.service.d/<prio>-postup-IP-state.conf
which call the script (default prio = 20).
+ - Stages a merged copy of rt_tables (does not write the live /etc/iproute2/rt_tables).
Returns:
- (script_path ,notes[list of strings])
+ (script_path, notes[list of strings])
Errors:
- Raises RuntimeError if no interfaces provided or there’s nothing to emit.
- - Does not write /etc/iproute2/rt_tables (that’s handled by your registration stager).
- Does not modify kernel state — this is staging only.
Notes:
- - The generated script is idempotent:
- * addresses: “add if missing”
- * routes: `ip -4 route replace`
- * rules: add only if a grep needle is not found
- - It accepts optional IFACE args at runtime to limit application to a subset.
+ - Addresses: reset pattern (del → add) for deterministic convergence.
+ - Routes: `ip -4 route replace` (best-practice) with tolerant logging.
+ - Rules: reset by `pref` (del-by-pref → add). Prefs are unique per iface:
+ base = 17000 + Iface.id * 10
+ from_pref = base + 0
+ uid_pref = base + 1
+ - The runtime script accepts optional IFACE args to limit application.
"""
from __future__ import annotations
from pathlib import Path
-from typing import Dict ,Iterable ,List ,Optional ,Sequence ,Tuple
+from typing import Dict, Iterable, List, Optional, Sequence, Tuple
import argparse
import sqlite3
import sys
ROOT = Path(__file__).resolve().parent
STAGE_ROOT = ROOT / "stage"
+RT_TABLES_PATH = Path("/etc/iproute2/rt_tables")
+
+
+# ---------- helpers for notes ----------
+
+def _stage_note(path: Path, stage_root: Path) -> str:
+ """Return a short path like 'stage:/usr/local/bin/apply_IP_state.sh'."""
+ try:
+ rel = path.relative_to(stage_root)
+ return f"stage:/{rel.as_posix()}"
+ except ValueError:
+ return str(path)
+
+
+# ---------- rt_tables helpers ----------
+
+def _parse_rt_tables(path: Path) -> Tuple[List[str], Dict[str, int], set[int]]:
+ """
+ Returns (lines, name_to_num, used_nums).
+ Keeps original lines for a non-destructive merge.
+ """
+ text = path.read_text() if path.exists() else ""
+ lines = text.splitlines()
+ name_to_num: Dict[str, int] = {}
+ used_nums: set[int] = set()
+ for ln in lines:
+ s = ln.strip()
+ if not s or s.startswith("#"):
+ continue
+ parts = s.split()
+ if len(parts) >= 2 and parts[0].isdigit():
+ n = int(parts[0]); nm = parts[1]
+ if nm not in name_to_num and n not in used_nums:
+ name_to_num[nm] = n
+ used_nums.add(n)
+ return (lines, name_to_num, used_nums)
+
+
+def _first_free_id(used_nums: Iterable[int], low: int, high: int) -> int:
+ used = set(used_nums)
+ for n in range(low, high + 1):
+ if n not in used:
+ return n
+ raise RuntimeError(f"no free routing-table IDs in [{low},{high}]")
+
+
+def _stage_rt_tables(
+ stage_root: Path,
+ meta: Dict[str, Tuple[int, Optional[int], str, Optional[str]]],
+ low: int = 20000,
+ high: int = 29999
+) -> Tuple[Path, List[str]]:
+ """
+ Ensure entries for all effective table names present in `meta`.
+ Prefer DB rt_table_id when available and not conflicting.
+ Write merged file to stage/etc/iproute2/rt_tables.
+ Returns (staged_path, notes)
+ """
+ lines, name_to_num, used_nums = _parse_rt_tables(RT_TABLES_PATH)
+
+ # Build eff_name -> preferred_num mapping (first non-None rt_id wins)
+ eff_to_preferred: Dict[str, Optional[int]] = {}
+ for _n, (_iid, rtid, eff, _cidr) in meta.items():
+ if eff not in eff_to_preferred:
+ eff_to_preferred[eff] = rtid if rtid is not None else None
+
+ additions: List[Tuple[int, str]] = []
+ for eff_name, preferred_num in eff_to_preferred.items():
+ if eff_name in name_to_num:
+ continue # already present
+ if preferred_num is not None and preferred_num not in used_nums:
+ num = preferred_num
+ else:
+ num = _first_free_id(used_nums, low, high)
+ name_to_num[eff_name] = num
+ used_nums.add(num)
+ additions.append((num, eff_name))
+
+ out = stage_root / "etc" / "iproute2" / "rt_tables"
+ out.parent.mkdir(parents=True, exist_ok=True)
+
+ if not additions:
+ # still write a copy of current file so install step is uniform
+ out.write_text("\n".join(lines) + ("\n" if lines else ""))
+ return (out, ["rt_tables: no additions (kept existing map)"])
+
+ new_lines = list(lines)
+ for num, name in sorted(additions):
+ new_lines.append(f"{num} {name}")
+
+ out.write_text("\n".join(new_lines) + "\n")
+ notes = [f"rt_tables: add {num} {name}" for num, name in sorted(additions)]
+ return (out, notes)
+
# ---------- DB access ----------
-def _fetch_meta_subu_cidr(conn: sqlite3.Connection ,default="10.0.0.0/24") -> str:
+def _fetch_meta_subu_cidr(conn: sqlite3.Connection, default="10.0.0.0/24") -> str:
row = conn.execute("SELECT value FROM Meta WHERE key='subu_cidr' LIMIT 1;").fetchone()
return str(row[0]) if row and row[0] else default
-def _fetch_iface_meta(conn: sqlite3.Connection ,iface_names: Sequence[str]) -> Dict[str ,Tuple[int ,str ,Optional[str]]]:
+def _fetch_iface_meta(conn: sqlite3.Connection, iface_names: Sequence[str]) -> Dict[str, Tuple[int, Optional[int], str, Optional[str]]]:
"""
- Return {iface_name -> (iface_id ,rt_table_name_eff ,local_address_cidr_or_None)}.
+ Return {iface_name -> (iface_id, rt_table_id, rt_table_name_eff, local_address_cidr_or_None)}.
"""
if not iface_names:
return {}
ph = ",".join("?" for _ in iface_names)
sql = f"""
- SELECT i.id
- , i.iface
- , v.rt_table_name_eff
- , NULLIF(TRIM(v.local_address_cidr),'') AS cidr
- FROM Iface i
- JOIN v_iface_effective v ON v.id = i.id
- WHERE i.iface IN ({ph})
- ORDER BY i.id;
+ SELECT i.id,
+ i.iface,
+ i.rt_table_id,
+ v.rt_table_name_eff,
+ NULLIF(TRIM(v.local_address_cidr),'') AS cidr
+ FROM Iface i
+ JOIN v_iface_effective v ON v.id = i.id
+ WHERE i.iface IN ({ph})
+ ORDER BY i.id;
"""
- rows = conn.execute(sql ,tuple(iface_names)).fetchall()
- out: Dict[str ,Tuple[int ,str ,Optional[str]]] = {}
+ rows = conn.execute(sql, tuple(iface_names)).fetchall()
+ out: Dict[str, Tuple[int, Optional[int], str, Optional[str]]] = {}
for r in rows:
- iface_id = int(r[0]); name = str(r[1]); eff = str(r[2]); cidr = (str(r[3]) if r[3] is not None else None)
- out[name] = (iface_id ,eff ,cidr)
+ iface_id = int(r[0]); name = str(r[1])
+ rt_id = (int(r[2]) if r[2] is not None else None)
+ eff = str(r[3])
+ cidr = (str(r[4]) if r[4] is not None else None)
+ out[name] = (iface_id, rt_id, eff, cidr)
return out
def _fetch_routes_by_iface_id(
- conn: sqlite3.Connection
- ,iface_ids: Sequence[int]
- ,only_on_up: bool = True
-) -> Dict[int ,List[Tuple[str ,Optional[str] ,Optional[str] ,Optional[int]]]]:
+ conn: sqlite3.Connection,
+ iface_ids: Sequence[int],
+ only_on_up: bool = True
+) -> Dict[int, List[Tuple[str, Optional[str], Optional[str], Optional[int]]]]:
"""
- Return {iface_id -> [(cidr ,via ,table_name_or_None ,metric_or_None),...]}.
+ Return {iface_id -> [(cidr, via, table_name_or_None, metric_or_None), ...]}.
"""
if not iface_ids:
return {}
ph = ",".join("?" for _ in iface_ids)
sql = f"""
- SELECT iface_id
- , cidr
- , NULLIF(TRIM(via),'') AS via
- , NULLIF(TRIM(table_name),'') AS table_name
- , metric
- , on_up
- FROM Route
- WHERE iface_id IN ({ph})
- ORDER BY id;
+ SELECT iface_id,
+ cidr,
+ NULLIF(TRIM(via),'') AS via,
+ NULLIF(TRIM(table_name),'') AS table_name,
+ metric,
+ on_up
+ FROM Route
+ WHERE iface_id IN ({ph})
+ ORDER BY id;
"""
- rows = conn.execute(sql ,tuple(iface_ids)).fetchall()
- out: Dict[int ,List[Tuple[str ,Optional[str] ,Optional[str] ,Optional[int]]]] = {}
- for iface_id ,cidr ,via ,tname ,metric ,on_up in rows:
+ rows = conn.execute(sql, tuple(iface_ids)).fetchall()
+ out: Dict[int, List[Tuple[str, Optional[str], Optional[str], Optional[int]]]] = {}
+ for iface_id, cidr, via, tname, metric, on_up in rows:
if only_on_up and int(on_up) != 1:
continue
- out.setdefault(int(iface_id) ,[]).append(
- (str(cidr) ,(str(via) if via is not None else None) ,(str(tname) if tname is not None else None)
- ,(int(metric) if metric is not None else None))
+ out.setdefault(int(iface_id), []).append(
+ (str(cidr),
+ (str(via) if via is not None else None),
+ (str(tname) if tname is not None else None),
+ (int(metric) if metric is not None else None))
)
return out
-def _fetch_uids_by_iface_id(conn: sqlite3.Connection ,iface_ids: Sequence[int]) -> Dict[int ,List[int]]:
+def _fetch_uids_by_iface_id(conn: sqlite3.Connection, iface_ids: Sequence[int]) -> Dict[int, List[int]]:
"""
- Return {iface_id -> [uid,...]} using table "User".
+ Return {iface_id -> [uid, ...]} using table "User".
"""
if not iface_ids:
return {}
ph = ",".join("?" for _ in iface_ids)
sql = f"""
- SELECT iface_id
- , uid
- FROM "User"
- WHERE iface_id IN ({ph})
- AND uid IS NOT NULL
- AND CAST(uid AS TEXT) != ''
- ORDER BY iface_id ,uid;
+ SELECT iface_id,
+ uid
+ FROM "User"
+ WHERE iface_id IN ({ph})
+ AND uid IS NOT NULL
+ AND CAST(uid AS TEXT) != ''
+ ORDER BY iface_id, uid;
"""
- rows = conn.execute(sql ,tuple(iface_ids)).fetchall()
- out: Dict[int ,List[int]] = {}
- for iface_id ,uid in rows:
- out.setdefault(int(iface_id) ,[]).append(int(uid))
+ rows = conn.execute(sql, tuple(iface_ids)).fetchall()
+ out: Dict[int, List[int]] = {}
+ for iface_id, uid in rows:
+ out.setdefault(int(iface_id), []).append(int(uid))
return out
# ---------- rendering ----------
def _render_composite_script(
- plan_ifaces: List[str]
- ,meta: Dict[str ,Tuple[int ,str ,Optional[str]]]
- ,routes_by_id: Dict[int ,List[Tuple[str ,Optional[str] ,Optional[str] ,Optional[int]]]]
- ,uids_by_id: Dict[int ,List[int]]
- ,subu_cidr: str
+ plan_ifaces: List[str],
+ meta: Dict[str, Tuple[int, Optional[int], str, Optional[str]]],
+ routes_by_id: Dict[int, List[Tuple[str, Optional[str], Optional[str], Optional[int]]]],
+ uids_by_id: Dict[int, List[int]],
+ subu_cidr: str
) -> str:
"""
Build a single bash script that ensures addresses → routes → rules.
"""
lines: List[str] = [
- "#!/usr/bin/env bash"
- ,"# apply IP state for selected interfaces (addresses, routes, rules) — idempotent"
- ,"set -euo pipefail"
- ,""
- ,"ALL_ARGS=(\"$@\")"
- ,""
- ,"want_iface(){"
- ," local t=$1"
- ," if [ ${#ALL_ARGS[@]} -eq 0 ]; then return 0; fi"
- ," for a in \"${ALL_ARGS[@]}\"; do [ \"$a\" = \"$t\" ] && return 0; done"
- ," return 1"
- ,"}"
- ,""
- ,"exists_iface(){ ip -o link show dev \"$1\" >/dev/null 2>&1; }"
- ,""
- ,"ensure_addr(){"
- ," local iface=$1; local cidr=$2"
- ," if ip -4 -o addr show dev \"$iface\" | awk '{print $4}' | grep -Fxq \"$cidr\"; then"
- ," logger \"addr ok: $iface $cidr\""
- ," else"
- ," ip -4 addr add \"$cidr\" dev \"$iface\""
- ," logger \"addr add: $iface $cidr\""
- ," fi"
- ,"}"
- ,""
- ,"ensure_route(){"
- ," local table=$1; local cidr=$2; local dev=$3; local via=${4:-}; local metric=${5:-}"
- ," if [ -n \"$via\" ] && [ -n \"$metric\" ]; then"
- ," ip -4 route replace \"$cidr\" via \"$via\" dev \"$dev\" table \"$table\" metric \"$metric\""
- ," elif [ -n \"$via\" ]; then"
- ," ip -4 route replace \"$cidr\" via \"$via\" dev \"$dev\" table \"$table\""
- ," elif [ -n \"$metric\" ]; then"
- ," ip -4 route replace \"$cidr\" dev \"$dev\" table \"$table\" metric \"$metric\""
- ," else"
- ," ip -4 route replace \"$cidr\" dev \"$dev\" table \"$table\""
- ," fi"
- ," logger \"route ensure: table=$table cidr=$cidr dev=$dev${via:+ via=$via}${metric:+ metric=$metric}\""
- ,"}"
- ,""
- ,"add_ip_rule_if_absent(){"
- ," local needle=$1; shift"
- ," if ! ip -4 rule show | grep -F -q -- \"$needle\"; then"
- ," ip -4 rule add \"$@\""
- ," logger \"rule add: $*\""
- ," else"
- ," logger \"rule ok: $needle\""
- ," fi"
- ,"}"
- ,""
+ "#!/usr/bin/env bash",
+ "# apply IP state for selected interfaces (addresses, routes, rules) — idempotent",
+ "set -euo pipefail",
+ "",
+ "ALL_ARGS=(\"$@\")",
+ "",
+ "want_iface(){",
+ " local t=$1",
+ " if [ ${#ALL_ARGS[@]} -eq 0 ]; then return 0; fi",
+ " for a in \"${ALL_ARGS[@]}\"; do [ \"$a\" = \"$t\" ] && return 0; done",
+ " return 1",
+ "}",
+ "",
+ "exists_iface(){ ip -o link show dev \"$1\" >/dev/null 2>&1; }",
+ "",
+ "# Reset address: delete the exact CIDR if present, then add it back.",
+ "reset_addr(){",
+ " local iface=$1; local cidr=$2",
+ " ip -4 addr del \"$cidr\" dev \"$iface\" >/dev/null 2>&1 || true",
+ " if ip -4 addr add \"$cidr\" dev \"$iface\"; then",
+ " logger \"addr set: $iface $cidr\"",
+ " else",
+ " logger \"addr add failed (non-fatal): $iface $cidr\"",
+ " fi",
+ "}",
+ "",
+ "# Ensure route using replace; log but do not fail the unit if kernel says 'exists'.",
+ "ensure_route(){",
+ " local table=$1; local cidr=$2; local dev=$3; local via=${4:-}; local metric=${5:-}",
+ " if [ -n \"$via\" ] && [ -n \"$metric\" ]; then",
+ " if ip -4 route replace \"$cidr\" via \"$via\" dev \"$dev\" table \"$table\" metric \"$metric\" 2>/dev/null; then",
+ " logger \"route ensure: table=$table cidr=$cidr dev=$dev via=$via metric=$metric\"",
+ " else",
+ " logger \"route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev via=$via metric=$metric\"",
+ " fi",
+ " elif [ -n \"$via\" ]; then",
+ " if ip -4 route replace \"$cidr\" via \"$via\" dev \"$dev\" table \"$table\" 2>/dev/null; then",
+ " logger \"route ensure: table=$table cidr=$cidr dev=$dev via=$via\"",
+ " else",
+ " logger \"route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev via=$via\"",
+ " fi",
+ " elif [ -n \"$metric\" ]; then",
+ " if ip -4 route replace \"$cidr\" dev \"$dev\" table \"$table\" metric \"$metric\" 2>/dev/null; then",
+ " logger \"route ensure: table=$table cidr=$cidr dev=$dev metric=$metric\"",
+ " else",
+ " logger \"route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev metric=$metric\"",
+ " fi",
+ " else",
+ " if ip -4 route replace \"$cidr\" dev \"$dev\" table \"$table\" 2>/dev/null; then",
+ " logger \"route ensure: table=$table cidr=$cidr dev=$dev\"",
+ " else",
+ " logger \"route ensure (tolerated failure): table=$table cidr=$cidr dev=$dev\"",
+ " fi",
+ " fi",
+ "}",
+ "",
+ "# Reset a policy rule by numeric preference: delete-by-pref, then add.",
+ "reset_IP_rule(){",
+ " # Usage: reset_IP_rule <pref> <rule-args...>",
+ " local pref=$1; shift",
+ " ip -4 rule del pref \"$pref\" >/dev/null 2>&1 || true",
+ " if ip -4 rule add \"$@\" pref \"$pref\"; then",
+ " logger \"rule set: pref=$pref $*\"",
+ " else",
+ " logger \"rule add failed (non-fatal): pref=$pref $*\"",
+ " fi",
+ "}",
+ "",
]
any_action = False
- # 1) Addresses
+ # 1) Addresses (reset)
for name in plan_ifaces:
- _iid ,rtname ,cidr = meta[name]
+ _iid, _rtid, rtname, cidr = meta[name]
if cidr:
lines += [
- f'if want_iface {name}; then'
- ,f' if exists_iface {name}; then ensure_addr {name} {cidr}; else logger "skip: iface missing: {name}"; fi'
- ,'fi'
+ f'if want_iface {name}; then',
+ f' if exists_iface {name}; then reset_addr {name} {cidr}; else logger "skip: iface missing: {name}"; fi',
+ 'fi'
]
any_action = True
# 2) Routes
for name in plan_ifaces:
- iid ,rtname ,_cidr = meta[name]
- rows = routes_by_id.get(iid ,[])
- for cidr ,via ,t_override ,metric in rows:
+ iid, _rtid, rtname, _cidr = meta[name]
+ rows = routes_by_id.get(iid, [])
+ for cidr, via, t_override, metric in rows:
table_eff = t_override or rtname
viastr = (via if via is not None else "")
mstr = (str(metric) if metric is not None else "")
lines += [
- f'if want_iface {name}; then'
- ,f' if exists_iface {name}; then ensure_route "{table_eff}" "{cidr}" "{name}" "{viastr}" "{mstr}"; else logger "skip: iface missing: {name}"; fi'
- ,'fi'
+ f'if want_iface {name}; then',
+ f' if exists_iface {name}; then ensure_route "{table_eff}" "{cidr}" "{name}" "{viastr}" "{mstr}"; else logger "skip: iface missing: {name}"; fi',
+ 'fi'
]
any_action = True
- # 3) Rules (src, uids, and one prohibit for the subu block)
+ # 3) Rules (reset by pref: src-cidr, uids, and one global prohibit)
for name in plan_ifaces:
- iid ,rtname ,cidr = meta[name]
+ iid, _rtid, rtname, cidr = meta[name]
+
+ # Per-iface preference block (no collisions)
+ base_pref = 17000 + iid * 10
+ from_pref = base_pref + 0
+ uid_pref = base_pref + 1
+
if cidr:
lines += [
- f'if want_iface {name}; then'
- ,f' add_ip_rule_if_absent "from {cidr} lookup {rtname}" from "{cidr}" lookup "{rtname}" pref 17000'
- ,'fi'
+ f'if want_iface {name}; then',
+ f' reset_IP_rule {from_pref} from "{cidr}" lookup "{rtname}"',
+ 'fi'
]
any_action = True
- uids = uids_by_id.get(iid ,[])
+
+ uids = uids_by_id.get(iid, [])
for u in uids:
lines += [
- f'if want_iface {name}; then'
- ,f' add_ip_rule_if_absent "uidrange {u}-{u} lookup {rtname}" uidrange "{u}-{u}" lookup "{rtname}" pref 17010'
- ,'fi'
+ f'if want_iface {name}; then',
+ f' reset_IP_rule {uid_pref} uidrange "{u}-{u}" lookup "{rtname}"',
+ 'fi'
]
any_action = True
- # One global prohibit for subu block (emit once)
if subu_cidr:
lines += [
- f'add_ip_rule_if_absent "from {subu_cidr} prohibit" from "{subu_cidr}" prohibit pref 18050'
+ f'reset_IP_rule 18050 from "{subu_cidr}" prohibit'
]
any_action = True
if not any_action:
raise RuntimeError("no IP state to emit for requested interfaces")
- lines += [""] # trailing newline
+ lines += [""]
return "\n".join(lines)
def _write_dropin_for_iface(stage_root: Path ,iface: str ,script_name: str ,priority: int) -> Path:
- d = stage_root / "etc" / "systemd" / f"wg-quick@{iface}.service.d"
+ # correct systemd path: /etc/systemd/system/wg-quick@IFACE.service.d/
+ d = stage_root / "etc" / "systemd" / "system" / f"wg-quick@{iface}.service.d"
d.mkdir(parents=True ,exist_ok=True)
- p = d / f"{priority}-postup-ip-state.conf"
+ p = d / f"{priority}-postup-IP-state.conf"
content = (
"[Service]\n"
f"ExecStartPost=+/usr/local/bin/{script_name} {iface}\n"
# ---------- business ----------
-def stage_ip_apply_script(
- conn: sqlite3.Connection
- ,iface_names: Sequence[str]
- ,stage_root: Optional[Path] = None
- ,script_name: str = "apply_ip_state.sh"
- ,dropin_priority: int = 20
- ,only_on_up: bool = True
- ,with_dropins: bool = True
- ,dry_run: bool = False
-) -> Tuple[Path ,List[str]]:
+def stage_IP_apply_script(
+ conn: sqlite3.Connection,
+ iface_names: Sequence[str],
+ stage_root: Optional[Path] = None,
+ script_name: str = "apply_IP_state.sh",
+ dropin_priority: int = 20,
+ only_on_up: bool = True,
+ with_dropins: bool = True,
+ dry_run: bool = False
+) -> Tuple[Path, List[str]]:
"""
- Plan and stage the unified runtime script and per-iface drop-ins.
+ Plan and stage the unified runtime script, a merged rt_tables, and per-iface drop-ins.
"""
if not iface_names:
raise RuntimeError("no interfaces provided")
- meta = _fetch_iface_meta(conn ,iface_names)
+ meta = _fetch_iface_meta(conn, iface_names)
if not meta:
raise RuntimeError("none of the requested interfaces exist in DB")
ifaces_in_order = [n for n in iface_names if n in meta]
iface_ids = [meta[n][0] for n in ifaces_in_order]
- routes_by_id = _fetch_routes_by_iface_id(conn ,iface_ids ,only_on_up=only_on_up)
- uids_by_id = _fetch_uids_by_iface_id(conn ,iface_ids)
- subu_cidr = _fetch_meta_subu_cidr(conn ,default="10.0.0.0/24")
+ routes_by_id = _fetch_routes_by_iface_id(conn, iface_ids, only_on_up=only_on_up)
+ uids_by_id = _fetch_uids_by_iface_id(conn, iface_ids)
+ subu_cidr = _fetch_meta_subu_cidr(conn, default="10.0.0.0/24")
sr = stage_root or STAGE_ROOT
out = sr / "usr" / "local" / "bin" / script_name
- out.parent.mkdir(parents=True ,exist_ok=True)
+ out.parent.mkdir(parents=True, exist_ok=True)
- content = _render_composite_script(ifaces_in_order ,meta ,routes_by_id ,uids_by_id ,subu_cidr)
+ content = _render_composite_script(ifaces_in_order, meta, routes_by_id, uids_by_id, subu_cidr)
notes: List[str] = []
if dry_run:
- notes.append(f"dry-run: would write {out}")
+ notes.append(f"dry-run: would write {_stage_note(out, sr)}")
if with_dropins:
for n in ifaces_in_order:
- notes.append(f"dry-run: would write drop-in for {n} at priority {dropin_priority}")
- return (out ,notes)
+ notes.append(f"dry-run: would write {_stage_note(sr / 'etc' / 'systemd' / 'system' / f'wg-quick@{n}.service.d' / f'{dropin_priority}-postup-IP-state.conf', sr)}")
+ rt_out = sr / "etc" / "iproute2" / "rt_tables"
+ notes.append(f"dry-run: would write {_stage_note(rt_out, sr)}")
+ return (out, notes)
+
+ # ensure rt_tables entries for the effective names used by these ifaces
+ rt_path, rt_notes = _stage_rt_tables(sr, meta)
+ notes.extend(rt_notes)
+ notes.append(f"staged: {_stage_note(rt_path, sr)}")
out.write_text(content)
out.chmod(0o500)
- notes.append(f"staged: {out}")
+ notes.append(f"staged: {_stage_note(out, sr)}")
if with_dropins:
for n in ifaces_in_order:
- dp = _write_dropin_for_iface(sr ,n ,script_name ,dropin_priority)
- notes.append(f"staged: {dp}")
+ dp = _write_dropin_for_iface(sr, n, script_name, dropin_priority)
+ notes.append(f"staged: {_stage_note(dp, sr)}")
+
+ return (out, notes)
- return (out ,notes)
+# Backwards-compatible alias for callers that still import the old name.
+stage_ip_apply_script = stage_IP_apply_script
# ---------- CLI ----------
def main(argv=None) -> int:
- ap = argparse.ArgumentParser(description="Stage one script that applies addresses, routes, and rules for selected ifaces.")
- ap.add_argument("ifaces" ,nargs="+" ,help="interface names to include")
- ap.add_argument("--script-name" ,default="apply_ip_state.sh")
- ap.add_argument("--dropin-priority" ,type=int ,default=20)
- ap.add_argument("--all" ,action="store_true" ,help="include routes where on_up=0 as well")
- ap.add_argument("--no-dropins" ,action="store_true" ,help="do not stage systemd drop-ins")
- ap.add_argument("--dry-run" ,action="store_true")
+ ap = argparse.ArgumentParser(description="Stage one script that applies IP addresses, routes, and rules for selected ifaces.")
+ ap.add_argument("ifaces", nargs="+", help="interface names to include")
+ ap.add_argument("--script-name", default="apply_IP_state.sh")
+ ap.add_argument("--dropin-priority", type=int, default=20)
+ ap.add_argument("--all", action="store_true", help="include routes where on_up=0 as well")
+ ap.add_argument("--no-dropins", action="store_true", help="do not stage systemd drop-ins")
+ ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args(argv)
with ic.open_db() as conn:
try:
- out ,notes = stage_ip_apply_script(
- conn
- ,args.ifaces
- ,script_name=args.script_name
- ,dropin_priority=args.dropin_priority
- ,only_on_up=(not args.all)
- ,with_dropins=(not args.no_dropins)
- ,dry_run=args.dry_run
+ out, notes = stage_IP_apply_script(
+ conn,
+ args.ifaces,
+ script_name=args.script_name,
+ dropin_priority=args.dropin_priority,
+ only_on_up=(not args.all),
+ with_dropins=(not args.no_dropins),
+ dry_run=args.dry_run
)
except Exception as e:
- print(f"error: {e}" ,file=sys.stderr)
+ print(f"error: {e}", file=sys.stderr)
return 2
if notes:
stage_client.py
Given:
- - A SQLite DB reachable via incommon.open_db()
- - A client machine name (used to locate ./key/<client_machine_name> for WG PrivateKey)
+ - A SQLite DB via incommon.open_db()
+ - A client machine name (for WG PrivateKey lookup under ./key/<client>)
- One or more interface names (e.g., x6, US)
Does:
- 1) Stage WireGuard confs for each iface (Table=off; ListenPort commented if NULL)
- 2) Stage /etc/iproute2/rt_tables entries for those ifaces
- 3) Stage a unified IP apply script (addresses, routes, rules)
- 4) Stage per-iface systemd drop-ins to invoke the apply script on wg-quick up
+ 1) Stage WireGuard confs for each iface
+ 2) Stage a unified IP apply script (addresses, routes, rules) + per-iface drop-ins
Returns:
- - True on success, False on failure
- - Prints human-readable progress for each step
-
-Errors:
- - Raises or prints clear ❌ messages on failure
+ True on success, False on failure (prints progress)
"""
from __future__ import annotations
return _msg_wrapped_call(f"stage_wg_conf ({client_name}; {','.join(ifaces)})" ,_do)
-def _stage_rt_tables_step(ifaces: Sequence[str]) -> bool:
- def _do():
- try:
- from stage_IP_register_route_table import stage_ip_register_route_table # type: ignore
- with ic.open_db() as conn:
- path ,notes = stage_ip_register_route_table(
- conn
- ,ifaces
- ,stage_root=STAGE_ROOT
- ,dry_run=False
- )
- return (path ,notes)
- except Exception:
- return _call_cli([str(ROOT / "stage_IP_register_route_table.py") ,*ifaces])
- return _msg_wrapped_call(f"stage_IP_register_route_table ({','.join(ifaces)})" ,_do)
-
-
def _stage_apply_ip_state_step(ifaces: Sequence[str]) -> bool:
def _do():
try:
,iface_names: Sequence[str]
,stage_root: Optional[Path] = None
) -> bool:
- """
- Orchestrate staging for a client+ifaces. Prints progress and returns success.
- """
if not iface_names:
raise ValueError("no interfaces provided")
if stage_root:
ok = True
ok = _stage_wg_conf_step(client_name ,iface_names) and ok
- ok = _stage_rt_tables_step(iface_names) and ok
ok = _stage_apply_ip_state_step(iface_names) and ok
return ok
--- /dev/null
+#!/usr/bin/env python3
+"""
+start_iface.py
+
+Given:
+ - One or more WireGuard interface names (e.g., x6, US).
+ - Optional presence of systemd and wg-quick(8).
+ - Expected config at /etc/wireguard/<iface>.conf.
+ - Optional staged IP state script at /usr/local/bin/apply_ip_state.sh.
+
+Does:
+ - For each iface (best-effort, non-fatal steps):
+ 0) (optional) systemctl daemon-reload
+ 1) Start via systemd: systemctl start wg-quick@IFACE.service (unless --no-systemd)
+ else via wg-quick: wg-quick up IFACE (unless --no-wg-quick)
+ If the iface already exists and --force is given, it will attempt a
+ best-effort teardown then retry the start once.
+ 2) If started (or already present), optionally run IP state script:
+ /usr/local/bin/apply_ip_state.sh IFACE (unless --skip-ip-state)
+ - Logs each action taken or skipped.
+
+Returns:
+ - Exit 0 on success (even if some steps were no-ops); 2 on argument/privilege errors.
+ - Prints a concise, per-iface action log.
+
+Errors:
+ - If no ifaces are provided, or if not running as root (unless --force-nonroot).
+
+Notes:
+ - This does NOT edit config files or DB; it just brings the iface up cleanly.
+ - Safe to re-run: “already up/exist” conditions are handled. Use --force to
+ tear down and recreate if needed.
+"""
+
+from __future__ import annotations
+from pathlib import Path
+from typing import Iterable, List, Sequence
+import argparse
+import os
+import shutil
+import subprocess
+import sys
+
+
+# ---------- helpers ----------
+
+def _run(cmd: Sequence[str]) -> tuple[int, str, str]:
+ """Run a command, capture stdout/stderr, return (rc, out, err)."""
+ try:
+ cp = subprocess.run(cmd, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ return (cp.returncode, cp.stdout.strip(), cp.stderr.strip())
+ except FileNotFoundError:
+ return (127, "", f"{cmd[0]}: not found")
+
+def _exists_iface(name: str) -> bool:
+ rc, _, _ = _run(["ip", "-o", "link", "show", "dev", name])
+ return rc == 0
+
+def _systemd_present() -> bool:
+ return shutil.which("systemctl") is not None
+
+def _wg_quick_present() -> bool:
+ return shutil.which("wg-quick") is not None
+
+def _conf_present(name: str) -> bool:
+ return Path(f"/etc/wireguard/{name}.conf").is_file()
+
+def _best_effort_teardown(name: str, logs: List[str]) -> None:
+ """Try to bring an iface down using systemd/wg-quick, then delete link; non-fatal."""
+ unit = f"wg-quick@{name}.service"
+ if _systemd_present():
+ rc, out, err = _run(["systemctl", "stop", unit])
+ if rc == 0:
+ logs.append(f"systemctl: stopped {unit}")
+ else:
+ logs.append(f"systemctl: stop {unit} (ignored): {err or out or f'rc={rc}'}")
+ if _wg_quick_present():
+ rc, out, err = _run(["wg-quick", "down", name])
+ if rc == 0:
+ logs.append("wg-quick: down ok")
+ else:
+ logs.append(f"wg-quick: down (ignored): {err or out or f'rc={rc}'}")
+ if _exists_iface(name):
+ rc, out, err = _run(["ip", "link", "del", "dev", name])
+ if rc == 0:
+ logs.append("ip link: deleted leftover device")
+ else:
+ logs.append(f"ip link: delete (ignored): {err or out or f'rc={rc}'}")
+
+
+# ---------- business ----------
+
+def start_ifaces(
+ ifaces: Sequence[str],
+ use_systemd: bool = True,
+ use_wg_quick: bool = True,
+ run_ip_state: bool = True,
+ ip_state_path: str = "/usr/local/bin/apply_ip_state.sh",
+ daemon_reload: bool = False,
+ force: bool = False,
+) -> List[str]:
+ """
+ Start the given WG ifaces and optionally apply IP state.
+ Returns a list of log lines.
+ """
+ logs: List[str] = []
+
+ if not ifaces:
+ raise RuntimeError("no interfaces provided")
+
+ have_systemd = _systemd_present()
+ have_wgquick = _wg_quick_present()
+ have_ipstate = Path(ip_state_path).is_file()
+
+ if use_systemd and daemon_reload and have_systemd:
+ rc, _out, err = _run(["systemctl", "daemon-reload"])
+ if rc == 0:
+ logs.append("systemctl: daemon-reload")
+ else:
+ logs.append(f"systemctl: daemon-reload (ignored): {err or f'rc={rc}'}")
+
+ for name in ifaces:
+ logs.append(f"== {name} ==")
+
+ # Ensure config exists
+ if not _conf_present(name):
+ logs.append(f"config missing: /etc/wireguard/{name}.conf (skip start)")
+ logs.append(f"status: absent")
+ logs.append("")
+ continue
+
+ started = False
+ already_present = _exists_iface(name)
+
+ # Optionally force recreate if device already around
+ if already_present and force:
+ logs.append("iface exists, --force given: tearing down before start")
+ _best_effort_teardown(name, logs)
+ already_present = _exists_iface(name)
+
+ # Start via systemd or wg-quick
+ if use_systemd and have_systemd:
+ unit = f"wg-quick@{name}.service"
+ rc, out, err = _run(["systemctl", "start", unit])
+ if rc == 0:
+ logs.append(f"systemctl: started {unit}")
+ started = True
+ else:
+ # If iface already exists, treat as running
+ if _exists_iface(name):
+ logs.append(f"systemctl: start {unit} reported error, but iface exists (continuing): {err or out or f'rc={rc}'}")
+ started = True
+ else:
+ logs.append(f"systemctl: start {unit} failed: {err or out or f'rc={rc}'}")
+ elif use_wg_quick and have_wgquick:
+ if already_present:
+ logs.append("wg-quick: iface already present")
+ started = True
+ else:
+ rc, out, err = _run(["wg-quick", "up", name])
+ if rc == 0:
+ logs.append("wg-quick: up ok")
+ started = True
+ else:
+ # If iface popped up anyway, continue
+ if _exists_iface(name):
+ logs.append(f"wg-quick: up reported error, but iface exists (continuing): {err or out or f'rc={rc}'}")
+ started = True
+ else:
+ logs.append(f"wg-quick: up failed: {err or out or f'rc={rc}'}")
+
+ else:
+ logs.append("no start method available (systemd/wg-quick disabled or not found)")
+
+ # If requested, apply IP state post-start (useful when not using systemd drop-ins)
+ if run_ip_state and have_ipstate:
+ if _exists_iface(name):
+ rc, out, err = _run([ip_state_path, name])
+ if rc == 0:
+ logs.append(f"ip-state: applied ({ip_state_path} {name})")
+ else:
+ logs.append(f"ip-state: apply failed: {err or out or f'rc={rc}'}")
+ else:
+ logs.append("ip-state: skipped (iface not present)")
+
+ # Final status
+ logs.append(f"status: {'up' if _exists_iface(name) else 'down'}")
+ logs.append("") # spacer
+
+ return logs
+
+
+# ---------- CLI (wrapper only) ----------
+
+def _require_root(allow_nonroot: bool) -> None:
+ if not allow_nonroot and os.geteuid() != 0:
+ raise RuntimeError("must run as root (use --force-nonroot to override)")
+
+def main(argv: Sequence[str] | None = None) -> int:
+ ap = argparse.ArgumentParser(description="Start one or more WireGuard interfaces safely.")
+ ap.add_argument("ifaces", nargs="+", help="interface names to start (e.g., x6 US)")
+ ap.add_argument("--no-systemd", action="store_true", help="do not call systemctl start wg-quick@IFACE")
+ ap.add_argument("--no-wg-quick", action="store_true", help="do not call wg-quick up IFACE")
+ ap.add_argument("--skip-ip-state", action="store_true", help="do not run apply_ip_state.sh after start")
+ ap.add_argument("--ip-state-path", default="/usr/local/bin/apply_ip_state.sh", help="path to the IP state script")
+ ap.add_argument("--daemon-reload", action="store_true", help="run systemctl daemon-reload before starts")
+ ap.add_argument("--force", action="store_true", help="if iface exists, tear down first and retry start")
+ ap.add_argument("--force-nonroot", action="store_true", help="allow running without root (best-effort)")
+ args = ap.parse_args(argv)
+
+ try:
+ _require_root(allow_nonroot=args.force_nonroot)
+ logs = start_ifaces(
+ args.ifaces,
+ use_systemd=(not args.no_systemd),
+ use_wg_quick=(not args.no_wg_quick),
+ run_ip_state=(not args.skip_ip_state),
+ ip_state_path=args.ip_state_path,
+ daemon_reload=args.daemon_reload,
+ force=args.force,
+ )
+ for line in logs:
+ print(line)
+ return 0
+ except Exception as e:
+ print(f"error: {e}", file=sys.stderr)
+ return 2
+
+if __name__ == "__main__":
+ sys.exit(main(sys.argv[1:]))
--- /dev/null
+#!/usr/bin/env python3
+"""
+stop_clean_iface.py
+
+Stop one or more WireGuard interfaces and clean IP state (rules/routes/addresses).
+"""
+
+from __future__ import annotations
+from pathlib import Path
+from typing import Iterable, List, Optional, Sequence, Tuple, Set
+import argparse
+import os
+import re
+import shutil
+import subprocess
+import sys
+
+__VERSION__ = "1.1-agg-errors"
+
+RT_TABLES_FILE = Path("/etc/iproute2/rt_tables")
+
+# ---------- helpers (shell) ----------
+
+def _run(cmd: Sequence[str], dry: bool=False) -> tuple[int, str, str]:
+ if dry:
+ return (0, "", "")
+ try:
+ cp = subprocess.run(cmd, check=False, text=True,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ return (cp.returncode, cp.stdout.strip(), cp.stderr.strip())
+ except FileNotFoundError:
+ return (127, "", f"{cmd[0]}: not found")
+
+def _exists_iface(name: str) -> bool:
+ rc, _, _ = _run(["ip", "-o", "link", "show", "dev", name])
+ return rc == 0
+
+def _systemd_present() -> bool:
+ return shutil.which("systemctl") is not None
+
+def _wg_quick_present() -> bool:
+ return shutil.which("wg-quick") is not None
+
+# ---------- helpers (routing tables & rules) ----------
+
+def _rt_table_num_for_name(name: str) -> Optional[int]:
+ if not RT_TABLES_FILE.exists():
+ return None
+ try:
+ text = RT_TABLES_FILE.read_text()
+ except Exception:
+ return None
+ for line in text.splitlines():
+ s = line.strip()
+ if not s or s.startswith("#"):
+ continue
+ parts = s.split()
+ if len(parts) >= 2 and parts[0].isdigit():
+ num = int(parts[0]); nm = parts[1]
+ if nm == name:
+ return num
+ return None
+
+_RULE_RE = re.compile(r"""^\s*(\d+):\s*(.+?)\s*$""")
+
+def _current_rule_lines() -> List[Tuple[int,str]]:
+ rc, out, _ = _run(["ip", "-4", "rule", "show"])
+ if rc != 0 or not out:
+ return []
+ rows: List[Tuple[int,str]] = []
+ for ln in out.splitlines():
+ m = _RULE_RE.match(ln)
+ if not m:
+ continue
+ pref = int(m.group(1))
+ rest = m.group(2)
+ rows.append((pref, rest))
+ return rows
+
+def _prefs_matching_lookups(lookups: Sequence[str]) -> Set[int]:
+ toks = [t for t in lookups if t]
+ prefs: Set[int] = set()
+ if not toks:
+ return prefs
+ for pref, rest in _current_rule_lines():
+ for t in toks:
+ if re.search(rf"\blookup\s+{re.escape(t)}\b", rest):
+ prefs.add(pref)
+ break
+ return prefs
+
+def _rule_del_by_pref(pref: int, logs: List[str], dry: bool) -> None:
+ rc, _out, err = _run(["ip", "-4", "rule", "del", "pref", str(pref)], dry=dry)
+ if rc == 0:
+ logs.append(f"ip rule: deleted pref {pref}")
+ else:
+ logs.append(f"ip rule: delete pref {pref} (ignored): {err or f'rc={rc}'}")
+
+def _flush_routes_for_table(table: str, logs: List[str], dry: bool) -> None:
+ rc, _out, err = _run(["ip", "-4", "route", "flush", "table", table], dry=dry)
+ if rc == 0:
+ logs.append(f"ip route: flushed table {table}")
+ else:
+ logs.append(f"ip route: flush table {table} (ignored): {err or f'rc={rc}'}")
+
+def _addr_del_all_v4_on_iface(iface: str, logs: List[str], dry: bool) -> None:
+ rc, out, err = _run(["ip", "-4", "-o", "addr", "show", "dev", iface], dry=dry)
+ if rc != 0:
+ logs.append(f"ip addr: list on {iface} (ignored): {err or f'rc={rc}'}")
+ return
+ cidrs: List[str] = []
+ for ln in out.splitlines():
+ parts = ln.split()
+ if len(parts) >= 4:
+ cidrs.append(parts[3])
+ if not cidrs:
+ logs.append("ip addr: none to remove")
+ return
+ for cidr in cidrs:
+ rc2, _o2, e2 = _run(["ip", "-4", "addr", "del", cidr, "dev", iface], dry=dry)
+ if rc2 == 0:
+ logs.append(f"ip addr: deleted {cidr}")
+ else:
+ logs.append(f"ip addr: delete {cidr} (ignored): {e2 or f'rc={rc2}'}")
+
+# ---------- business ----------
+
+def _clean_iface_ip_state(name: str, logs: List[str], *, dry: bool=False, aggressive: bool=False) -> None:
+ tokens: List[str] = [name]
+ num = _rt_table_num_for_name(name)
+ if num is not None:
+ tokens.append(str(num))
+
+ # Delete rules matching either numeric or named lookup tokens; loop to catch chains.
+ deleted_any = True
+ safety = 0
+ while deleted_any and safety < 10:
+ safety += 1
+ prefs = sorted(_prefs_matching_lookups(tokens))
+ if not prefs:
+ deleted_any = False
+ break
+ for p in prefs:
+ _rule_del_by_pref(p, logs, dry)
+ if aggressive:
+ for p in range(17000, 17060):
+ _rule_del_by_pref(p, logs, dry)
+
+ # Flush routes in the table by name and numeric (if known)
+ _flush_routes_for_table(name, logs, dry)
+ if num is not None:
+ _flush_routes_for_table(str(num), logs, dry)
+
+ # Remove all IPv4 addresses on the iface
+ _addr_del_all_v4_on_iface(name, logs, dry)
+
+def stop_clean_ifaces(
+ ifaces: Sequence[str],
+ use_systemd: bool = True,
+ use_wg_quick: bool = True,
+ do_clean: bool = True,
+ aggressive: bool = False,
+ dry_run: bool = False,
+) -> List[str]:
+ logs: List[str] = []
+ if not ifaces:
+ raise RuntimeError("no interfaces provided")
+
+ have_systemd = _systemd_present()
+ have_wgquick = _wg_quick_present()
+
+ for name in ifaces:
+ logs.append(f"== {name} ==")
+
+ if use_systemd and have_systemd:
+ unit = f"wg-quick@{name}.service"
+ rc, out, err = _run(["systemctl", "stop", unit], dry=dry_run)
+ if rc == 0:
+ logs.append(f"systemctl: stopped {unit}")
+ else:
+ msg = err or out or f"rc={rc}"
+ logs.append(f"systemctl: stop {unit} (ignored): {msg}")
+ elif use_systemd and not have_systemd:
+ logs.append("systemctl: not found; skipped")
+
+ if use_wg_quick and have_wgquick:
+ rc, out, err = _run(["wg-quick", "down", name], dry=dry_run)
+ if rc == 0:
+ logs.append("wg-quick: down ok")
+ else:
+ msg = err or out or f"rc={rc}"
+ logs.append(f"wg-quick: down (ignored): {msg}")
+ elif use_wg_quick and not have_wgquick:
+ logs.append("wg-quick: not found; skipped")
+
+ if do_clean:
+ _clean_iface_ip_state(name, logs, dry=dry_run, aggressive=aggressive)
+ else:
+ logs.append("clean: skipped (--no-clean)")
+
+ if _exists_iface(name):
+ rc, out, err = _run(["ip", "link", "del", "dev", name], dry=dry_run)
+ if rc == 0:
+ logs.append("ip link: deleted device")
+ else:
+ msg = err or out or f"rc={rc}"
+ logs.append(f"ip link: delete (ignored): {msg}")
+ else:
+ logs.append("ip link: device not present; nothing to delete")
+
+ final_present = _exists_iface(name)
+ logs.append(f"status: {'gone' if not final_present else 'still present'}")
+ logs.append("")
+
+ return logs
+
+# ---------- CLI (wrapper with aggregated errors) ----------
+
+def main(argv: Sequence[str] | None = None) -> int:
+ ap = argparse.ArgumentParser(
+ description="Stop one or more WireGuard interfaces and clean IP state.",
+ add_help=True)
+ ap.add_argument("ifaces", nargs="*", help="interface names to stop (e.g., x6 US)")
+ ap.add_argument("--no-systemd", action="store_true", help="do not call systemctl stop wg-quick@IFACE")
+ ap.add_argument("--no-wg-quick", action="store_true", help="do not call wg-quick down IFACE")
+ ap.add_argument("--no-clean", action="store_true", help="skip IP cleanup (rules/routes/addresses)")
+ ap.add_argument("--aggressive", action="store_true", help="also purge common rule pref window (17000-17059)")
+ ap.add_argument("--dry-run", action="store_true", help="print what would be done without changing state")
+ ap.add_argument("--force-nonroot", action="store_true", help="allow running without root (best-effort)")
+
+ args = ap.parse_args(argv)
+
+ # Aggregate invocation errors
+ errors: List[str] = []
+ if os.geteuid() != 0 and not args.force_nonroot:
+ errors.append("must run as root (use --force-nonroot to override)")
+ if not args.ifaces:
+ errors.append("no interfaces provided")
+
+ if errors:
+ sys.stderr.write(ap.format_usage())
+ prog = Path(sys.argv[0]).name or "stop_clean_iface.py"
+ sys.stderr.write(f"{prog}: error: " + "; ".join(errors) + "\n")
+ return 2
+
+ try:
+ logs = stop_clean_ifaces(
+ args.ifaces,
+ use_systemd=(not args.no_systemd),
+ use_wg_quick=(not args.no_wg_quick),
+ do_clean=(not args.no_clean),
+ aggressive=args.aggressive,
+ dry_run=args.dry_run,
+ )
+ for line in logs:
+ print(line)
+ return 0
+ except Exception as e:
+ print(f"error: {e}", file=sys.stderr)
+ return 2
+
+if __name__ == "__main__":
+ sys.exit(main(sys.argv[1:]))
-#+TITLE: subu / WireGuard — TODO
+n#+TITLE: subu / WireGuard — TODO
#+AUTHOR: Thomas & Nerith (session)
#+LANGUAGE: en
#+OPTIONS: toc:2 num:t
- have the stage commands echo relative pathnames instead of absolute as they do now.
+- the one private key pair per client (instead of per interface), turns out to be a bad idea, as we can't manage tunnels individually, say, by revoking keys. We need to move to a key pair per interface instead.
+
+- db_wipe needs to delete the key directory contents also
+
+-------------------------------
+
- Known gaps / open decisions
- Systemd drop-in to call staged scripts on ~wg-quick@IFACE~ up (IPv4 addrs + policy rules).
- Staged policy-rules script (source-based + uidrange rules) to replace the old global ~IP_rule_add.sh~ usage.
projects / subu / commitdiff
