#!/usr/bin/env python3 """backhaul agent — the containerized DB-backup client. Runs as a sidecar on a database's private Docker network: it reaches the DB internally and calls home outbound over HTTPS. No inbound ports. Stdlib only; the DB client tools (pg_dump/mysqldump/mongodump) come from the image. Every minute (in-container cron) it CHECKS IN: heartbeats, reports the databases it can see, delivers any pending dump result, and asks what to do. The app owns the schedule and replies with a dump command when one is due; the agent then dumps and streams it home. The app also advertises the current agent-script version, and the agent self-updates (atomic swap after a compile check) — same mechanism as snapshoot. Config (env, from the `docker run`): BACKHAUL_URL e.g. https://backhaul.skui.io BACKHAUL_TOKEN agent token from the backhaul admin BACKHAUL_TARGETS comma-separated DB targets, each: engine:host:port:db:user:pass e.g. postgres:db:5432:nextcloud:nextcloud:nextcloud """ import json import os import subprocess import sys import urllib.request VERSION = "1.0" RESULT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".dump-result.json") def targets(): """Parse BACKHAUL_TARGETS into dicts. Each: engine:host:port:db:user:pass.""" out = [] for spec in os.environ.get("BACKHAUL_TARGETS", "").split(","): spec = spec.strip() if not spec: continue parts = spec.split(":") if len(parts) < 6: print(f"skipping malformed target: {spec}", file=sys.stderr) continue engine, host, port, db, user, pw = parts[:6] out.append({"engine": engine, "host": host, "port": port, "db": db, "user": user, "pw": pw, "name": f"{engine}/{db}"}) return out def db_version(t): """Fetch the engine version string so the app can pick a matching restore image.""" try: if t["engine"] == "postgres": env = dict(os.environ, PGPASSWORD=t["pw"]) out = subprocess.run(("psql", "-h", t["host"], "-p", t["port"], "-U", t["user"], "-d", t["db"], "-tAc", "select version()"), capture_output=True, text=True, timeout=30, env=env).stdout return out.strip().split(" on ")[0][:60] if t["engine"] in ("mysql", "mariadb"): out = subprocess.run(("mysql", "-h", t["host"], "-P", t["port"], "-u", t["user"], f"-p{t['pw']}", "-N", "-e", "select version()"), capture_output=True, text=True, timeout=30).stdout return "MySQL/MariaDB " + out.strip()[:40] if t["engine"] == "mongo": return "MongoDB" except Exception: pass return t["engine"] def dump_cmd(t): """The dump command + environment for an engine. Streams compressed to stdout.""" if t["engine"] == "postgres": env = dict(os.environ, PGPASSWORD=t["pw"]) # custom format is already compressed and is what pg_restore wants return (("pg_dump", "-h", t["host"], "-p", t["port"], "-U", t["user"], "-d", t["db"], "-Fc"), env) if t["engine"] in ("mysql", "mariadb"): return (("sh", "-c", f"mysqldump -h {t['host']} -P {t['port']} -u {t['user']} " f"-p{t['pw']} --single-transaction --routines {t['db']} | gzip"), dict(os.environ)) if t["engine"] == "mongo": return (("mongodump", "--host", t["host"], "--port", t["port"], "--db", t["db"], "--archive", "--gzip"), dict(os.environ)) raise ValueError(f"unknown engine {t['engine']}") def run_dump(t, url, token): """Stream a dump of one target straight to the app's ingest endpoint — never buffer the whole dump in memory or a temp file.""" cmd, env = dump_cmd(t) meta = json.dumps({"engine": t["engine"], "version": db_version(t)}) req = urllib.request.Request( url.rstrip("/") + "/api/dump", headers={"X-Agent-Token": token, "X-DB-Name": t["name"], "X-DB-Meta": meta, "Content-Type": "application/octet-stream", "Transfer-Encoding": "chunked", "User-Agent": f"backhaul-agent/{VERSION}"}) proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, env=env) # feed the dump process's stdout straight into the HTTP request body req.data = iter(lambda: proc.stdout.read(1 << 20), b"") with urllib.request.urlopen(req, timeout=3600) as resp: proc.wait() return {"name": t["name"], "ok": resp.status == 200 and proc.returncode == 0} def maybe_self_update(answer, url, token): """Swap our own script when the app advertises a newer version — atomic, compile-checked. Disable with BACKHAUL_AUTOUPDATE=0.""" latest = answer.get("agent_latest") if (not latest or latest == VERSION or os.environ.get("BACKHAUL_AUTOUPDATE", "1") == "0"): return me = os.path.abspath(__file__) try: req = urllib.request.Request(url.rstrip("/") + "/api/agent-script", headers={"X-Agent-Token": token}) with urllib.request.urlopen(req, timeout=30) as r: source = r.read() text = source.decode() if "backhaul-agent" not in text or f'VERSION = "{latest}"' not in text: return compile(text, me, "exec") tmp = me + ".new" with open(tmp, "wb") as f: f.write(source) os.chmod(tmp, os.stat(me).st_mode) os.replace(tmp, me) print(f"self-updated {VERSION} -> {latest}") except Exception as e: print("self-update failed:", e, file=sys.stderr) def checkin(url, token, tgts, pending): """POST the heartbeat: which DBs we see + any finished dump result. The reply tells us what to do next and the latest agent version.""" body = json.dumps({"agent_version": VERSION, "databases": [{"name": t["name"], "engine": t["engine"]} for t in tgts], "results": pending}).encode() req = urllib.request.Request(url.rstrip("/") + "/api/checkin", data=body, headers={"X-Agent-Token": token, "Content-Type": "application/json", "User-Agent": f"backhaul-agent/{VERSION}"}) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read() or b"{}") def main(): url = os.environ.get("BACKHAUL_URL", "").rstrip("/") token = os.environ.get("BACKHAUL_TOKEN", "") if not (url and token): sys.exit("BACKHAUL_URL and BACKHAUL_TOKEN must be set") tgts = targets() pending = [] if os.path.exists(RESULT_FILE): try: pending = json.load(open(RESULT_FILE)) except ValueError: pending = [] answer = checkin(url, token, tgts, pending) if pending: os.remove(RESULT_FILE) # run any dump the app asked for (matched by name), record result for next check-in results = [] for want in answer.get("dump") or []: t = next((t for t in tgts if t["name"] == want), None) if t: print(f"dumping {t['name']}...", flush=True) results.append(run_dump(t, url, token)) if results: json.dump(results, open(RESULT_FILE, "w")) maybe_self_update(answer, url, token) if __name__ == "__main__": main()