auracaster/castbox-provisioning
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement device provisioning with app update, service management and reboot

Browse Source
This commit is contained in:
pstruebi
2025-10-14 14:13:50 +02:00
parent 2be380f614
commit 750c019806
1 changed files with 148 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
src/provision.py
View File

@@ -21,17 +21,6 @@ SSH_PORT = int(os.getenv("IOT_SSH_PORT", "22"))
SSH_KEY = os.getenv("SSH_KEY") or None # path or None
PROVISION_LOG = os.getenv("PROVISION_LOG") or str((Path(__file__).resolve().parent / "provision.log"))
def _get(obj, *names):
"""Safe attribute/dict getter trying several candidate names."""
for n in names:
if hasattr(obj, n):
return getattr(obj, n)
if isinstance(obj, dict) and n in obj:
return obj[n]
return None
def scp_and_enable(ssh_host, config_text):
tmp = Path(tempfile.gettempdir()) / f"{WG_IFACE}.conf"
@@ -47,9 +36,11 @@ def scp_and_enable(ssh_host, config_text):
sudo mkdir -p /etc/wireguard
sudo install -m 600 /tmp/{WG_IFACE}.conf /etc/wireguard/{WG_IFACE}.conf
# If interface exists already, bring it down first to reload config cleanly
sudo wg-quick down {WG_IFACE} || true
sudo systemctl stop wg-quick@{WG_IFACE} || sudo wg-quick down {WG_IFACE} || true
sudo ip link del {WG_IFACE} || true
sudo systemctl reset-failed wg-quick@{WG_IFACE} || true
sudo systemctl daemon-reload || true
# Bring interface up and ensure service is enabled+active
sudo wg-quick up {WG_IFACE} || true
sudo systemctl enable --now wg-quick@{WG_IFACE}
sudo systemctl is-enabled wg-quick@{WG_IFACE} || true
sudo systemctl is-active wg-quick@{WG_IFACE} || true
@@ -123,7 +114,6 @@ def step_set_hostname(iot_host: str, hostname: str | None):
# Known locations where the script might live on the device
candidates = [
"/home/caster/bumble-auracast/src/auracast/server/provision_domain_hostname.sh",
"/home/caster/bumble-auracast/src/server/provision_domain_hostname.sh",
]
domain = "local"
@@ -172,16 +162,38 @@ def step_set_hostname(iot_host: str, hostname: str | None):
"err": stderr[-500:],
}
def step_update_app(iot_host: str, services: list[str] | None = None):
def step_update_app(iot_host: str):
"""Placeholder: start/enable required system services on the device.
Intention: SSH into the device and run systemctl enable --now <service> for required services.
Currently does nothing.
"""
# TODO: run the app update script
services = services or []
print("⏭️ [placeholder] Skipping start services (no-op).")
return {"services": services}
remote = (
"set -e\n"
"cd ~/bumble-auracast\n"
"git pull\n"
"/home/caster/.local/bin/poetry config virtualenvs.in-project true\n"
"/home/caster/.local/bin/poetry install\n"
)
ssh_cmd = ["ssh", "-p", str(SSH_PORT)]
if SSH_KEY:
ssh_cmd += ["-i", SSH_KEY]
ssh_cmd += [f"{SSH_USER}@{iot_host}", remote]
proc = subprocess.run(ssh_cmd, check=False, capture_output=True, text=True)
stdout = (proc.stdout or "").strip()
stderr = (proc.stderr or "").strip()
if proc.returncode != 0:
print(f"❌ update app: failed rc={proc.returncode}: {stderr}")
else:
print("✅ update app: poetry install completed")
return {
"rc": proc.returncode,
"out": stdout[-500:],
"err": stderr[-500:],
}
def step_start_app(iot_host: str, app: str):
"""Placeholder: start the application flow (script or UI) on the device.
@@ -189,9 +201,90 @@ def step_start_app(iot_host: str, app: str):
Intention: depending on --app, run an app launcher over SSH.
Currently does nothing.
"""
# TODO: start with pulling the latest main from the repo
print(f"⏭️ [placeholder] Skipping start app (no-op). --app={app}")
return {"app_mode": app}
scripts = [
"update_and_run_pw_aes67.sh",
"update_and_run_server_and_frontend.sh",
]
remote = (
"set -e\n"
"loginctl enable-linger \"$USER\"\n"
"cd ~/bumble-auracast\n"
"for s in " + " ".join(shlex.quote(s) for s in scripts) + "; do\n"
" if [ -f \"$s\" ]; then\n"
" echo \"▶ running $s\"\n"
" bash \"$s\"\n"
" else\n"
" echo \"⚠️ missing $s\" 1>&2\n"
" fi\n"
"done\n"
"# Ensure services are enabled and started\n"
"systemctl --user daemon-reload\n"
"systemctl --user enable --now auracast-server.service\n"
"sudo systemctl daemon-reload\n"
"sudo systemctl enable --now auracast-frontend.service\n"
"systemctl --user is-enabled auracast-server.service || true\n"
"systemctl --user is-active auracast-server.service || true\n"
"sudo systemctl is-enabled auracast-frontend.service || true\n"
"sudo systemctl is-active auracast-frontend.service || true\n"
)
ssh_cmd = ["ssh", "-p", str(SSH_PORT)]
if SSH_KEY:
ssh_cmd += ["-i", SSH_KEY]
ssh_cmd += [f"{SSH_USER}@{iot_host}", remote]
proc = subprocess.run(ssh_cmd, check=False, capture_output=True, text=True)
stdout = (proc.stdout or "").strip()
stderr = (proc.stderr or "").strip()
if proc.returncode != 0:
print(f"❌ start app: failed rc={proc.returncode}")
if stderr:
print(f"stderr: {stderr[-1000:]}")
if stdout:
print(f"stdout: {stdout[-1000:]}")
else:
print("✅ start app: scripts executed")
if stdout:
print(f"Output:\n{stdout[-1000:]}")
return {
"app_mode": app,
"rc": proc.returncode,
"out": stdout[-1000:],
"err": stderr[-1000:],
}
def step_finish(iot_host: str):
"""Finalize setup on the device: enable linger for the user and reboot.
Executes on remote:
- loginctl enable-linger "$USER"
- sudo reboot
"""
remote = (
"set -e\n"
"sudo reboot\n"
)
ssh_cmd = ["ssh", "-p", str(SSH_PORT)]
if SSH_KEY:
ssh_cmd += ["-i", SSH_KEY]
ssh_cmd += [f"{SSH_USER}@{iot_host}", remote]
proc = subprocess.run(ssh_cmd, check=False, capture_output=True, text=True)
stdout = (proc.stdout or "").strip()
stderr = (proc.stderr or "").strip()
if proc.returncode != 0:
print(f"❌ finish: failed rc={proc.returncode}: {stderr}")
else:
print("✅ finish: reboot initiated")
return {
"rc": proc.returncode,
"out": stdout[-500:],
"err": stderr[-500:],
}
def main():
import argparse
@@ -202,40 +295,40 @@ def main():
)
)
ap.add_argument("iot_host", help="Local/LAN IP or hostname of the IoT device reachable via SSH")
ap.add_argument("--name", required=True, help="Device name to use for both hostname and WireGuard client")
ap.add_argument("-n", "--name", required=True, help="Device name to use for both hostname and WireGuard client")
ap.add_argument(
"--steps",
nargs="+",
choices=["wg", "hostname", "services", "app", "all"],
choices=["wg", "hostname", "update_app", "start_app", "finish", "all"],
default=["all"],
help="Which steps to run. Default: all",
)
# Hostname will be taken from --name
ap.add_argument("--app", choices=["ui", "script"], default="ui", help="Application mode to start")
ap.add_argument("--services", nargs="*", default=[], help="Services to start/enable (logged)")
args = ap.parse_args()
# Validate wg-easy env
get_env_auth()
# Derive device name: if numeric, prefix with 'summitwave-beacon'
name = args.name
if re.fullmatch(r"\d+", name):
name = f"summitwave-beacon{name}"
# Normalize steps
steps = args.steps
if "all" in steps:
steps = ["wg", "hostname", "services", "app"]
steps = ["hostname", "wg", "update_app", "start_app", "finish"]
# Backward compatibility: map 'app' to 'start_app' if present
steps = ["start_app" if s == "app" else s for s in steps]
# Gather device facts once (may change after hostname step, but we at least log the initial state)
facts = get_device_facts(args.iot_host)
# Execute selected steps in order with logging
if "wg" in steps:
wg_info = step_wireguard_provision(args.iot_host, args.name)
write_provision_log({
"action": "wg",
**facts,
**wg_info,
})
if "hostname" in steps:
host_info = step_set_hostname(args.iot_host, args.name)
host_info = step_set_hostname(args.iot_host, name)
# refresh hostname after step (if a real implementation later changes it)
facts_post = get_device_facts(args.iot_host)
write_provision_log({
@@ -243,20 +336,34 @@ def main():
**facts_post,
**host_info,
})
if "services" in steps:
svc_info = step_update_app(args.iot_host, args.services)
if "wg" in steps:
wg_info = step_wireguard_provision(args.iot_host, name)
write_provision_log({
"action": "services",
**get_device_facts(args.iot_host),
**svc_info,
"action": "wg",
**facts,
**wg_info,
})
if "app" in steps:
if "update_app" in steps:
upd_info = step_update_app(args.iot_host)
write_provision_log({
"action": "update_app",
**get_device_facts(args.iot_host),
**upd_info,
})
if "start_app" in steps:
app_info = step_start_app(args.iot_host, args.app)
write_provision_log({
"action": "app",
"action": "start_app",
**get_device_facts(args.iot_host),
**app_info,
})
if "finish" in steps:
fin_info = step_finish(args.iot_host)
write_provision_log({
"action": "finish",
**get_device_facts(args.iot_host),
**fin_info,
})
if __name__ == "__main__":
main()