#!/usr/bin/env python3 """Validate a hostname change end to end: what is on disk, what is running, what the UIs were told, what DNS says. sudo vghosts-check.py on the server: everything vghosts-check.py --repo on a dev box: only the checks that need the repo (defaults, docs, pages) sudo vghosts-check.py --json machine-readable Source of truth is the Caddyfile (`# vanguard:` markers), the same place configure.py reads the hostnames back from. Every other place a hostname lives is checked AGAINST it: a host that is not one of Caddy's is stale. "ovh1." is the box's ssh name and must appear nowhere a browser is sent. ORDER MATTERS, and the sections below run in the order a change must be made, because each depends on the one before it (HOSTNAME-CHANGE.md, section E): 1 DNS nothing can be issued a certificate or reached until the name resolves to the box 2 Caddyfile the roles and their names: what configure.py --full / --hosts wrote; certificates follow 3 .env the same names as compose variables; a mismatch here means a half-done --full 4 mediamtx.yml webrtcAdditionalHosts = main host (WebRTC ICE); needs a MediaMTX restart to take effect 5 containers the env each running container was CREATED with (a .env edit without a recreate shows here) 6 HTTPS every host answers with a valid certificate; gated hosts send browsers to the auth host 7 Jellyfin SSO plugin OID endpoint, branding disclaimer, Vanguard plugin URLs (on disk) + the SSO start 8 Gitea app.ini, the auth source (live redirect), the runner, the server clone's remote 9 Authentik outpost host, proxy provider external host + cookie domain, Jellyfin/Gitea redirect URIs. LAST among the services, on purpose: an app changed before Authentik is merely unreachable on its new name until Authentik catches up; Authentik changed before the app locks you out of it 10 Discord the OAuth2 redirect in the developer portal — cannot be read; the exact value is printed 11 repo deployLive.sh / switchToClone.sh defaults, about page labels = Caddy roles, docs 12 streamers OBS dock URL and clock source: cannot be read; listed so nobody forgets A failure in an early section usually explains failures in later ones: fix top-down, re-run. The sweep runs the Jellyfin and Gitea checks before Authentik, and the report lists them that way, because that is the order to make the change in. """ import argparse, json, os, re, socket, ssl, subprocess, sys, urllib.request, urllib.error from pathlib import Path ROLES = ("main", "auth", "watch", "git", "bot", "live", "files", "sync", "clips", "diag") ENV_KEY = {"main": "MAIN_HOST", "watch": "WATCH_HOST", "git": "GIT_HOST", "bot": "BOT_HOST", "live": "LIVE_HOST", "files": "FILES_HOST", "sync": "SYNC_HOST", "clips": "CLIPS_HOST", "diag": "DIAG_HOST"} GATED = ("main", "files", "sync", "clips", "live", "bot", "diag") # forward-auth: anonymous -> auth host REPO = Path(os.environ.get("VANGUARD_REPO", str(Path.home() / "vanguard-stream") if Path.home().joinpath("vanguard-stream").is_dir() else ".")) APPDATA = Path("/srv/appdata") results = [] # (section, status, text) def rec(section, status, text): results.append((section, status, text)) def ok(section, text): rec(section, "PASS", text) def bad(section, text): rec(section, "FAIL", text) def warn(section, text): rec(section, "WARN", text) def manual(section, text): rec(section, "MANUAL", text) def sh(cmd, timeout=30, **kw): try: return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, **kw) except (OSError, subprocess.TimeoutExpired) as error: class R: returncode, stdout, stderr = 1, "", str(error) return R() def read_hosts(caddyfile): hosts, role = {}, None for line in caddyfile.read_text().splitlines(): if line[:1].isspace(): continue t = line.strip() m = re.match(r"#\s*vanguard:(\w+)", t) if m: role = m.group(1); continue if t and not t.startswith(("#", "}", "{")) and t.endswith("{") and " " in t: host = t.split()[0].rstrip(",") # "a, b {": an alias; the first name is the role's if "example.com" in host or role == "redirect": role = None; continue hosts[role or ("main" if "main" not in hosts else host)] = host role = None return hosts def read_env(path): env = {} try: for line in path.read_text().splitlines(): if "=" in line and not line.startswith("#"): k, v = line.split("=", 1); env[k.strip()] = v.strip() except OSError: pass return env URL_RE = re.compile(r"https?://([A-Za-z0-9.-]+\.[A-Za-z]{2,})(?::\d+)?") def hosts_in(text): return sorted(set(h.lower() for h in URL_RE.findall(text))) def classify(section, where, found, allowed, expect=None, base=""): """found: hostnames seen in `where`. allowed: every current host. expect: the one role's host, if specific.""" if not found: warn(section, f"{where}: no hostname found"); return for h in found: if expect and h == expect: ok(section, f"{where}: {h}") elif expect and h != expect and (h.endswith("." + base) or h.startswith("ovh1.")): bad(section, f"{where}: {h} — expected {expect}") elif h in allowed: ok(section, f"{where}: {h}") elif base and (h.endswith("." + base) or h == base or h.startswith("ovh1.")): bad(section, f"{where}: {h} — not a current host (stale)") else: pass # third-party hostnames (discord.com, github.com, …) are not ours to judge class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *a, **k): return None _opener = urllib.request.build_opener(_NoRedirect) def curl_head(url, timeout=12): """(status, location, error) for a browser-like GET, without following redirects. GET, not HEAD: the forward-auth gate answers HEAD with 200 and the SSO start endpoints are GET-only.""" req = urllib.request.Request(url, method="GET", headers={"User-Agent": "Mozilla/5.0 vghosts-check", "Accept": "text/html,*/*"}) try: with _opener.open(req, timeout=timeout) as r: r.read(2048) return r.status, r.headers.get("Location", ""), "" except urllib.error.HTTPError as e: return e.code, e.headers.get("Location", ""), "" except Exception as e: return 0, "", str(e) def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--repo", action="store_true", help="repo-only checks (dev box)") ap.add_argument("--json", action="store_true") ap.add_argument("--caddyfile", default=None, help="path to the generated Caddyfile (default: /Caddyfile)") args = ap.parse_args() repo = REPO caddyfile = Path(args.caddyfile) if args.caddyfile else repo / "Caddyfile" # ---- 2 Caddyfile: the truth ------------------------------------------------------------------- S = "2 Caddyfile" if not caddyfile.is_file(): if args.repo: hosts = {} warn(S, f"no Caddyfile at {caddyfile} (dev box): repo checks use the names in deployLive.sh / switchToClone.sh") else: bad(S, f"no Caddyfile at {caddyfile}: nothing to validate against — run configure.py --full first"); return report(args) else: hosts = read_hosts(caddyfile) main_host = hosts.get("main", "") base = main_host.split(".", 1)[1] if "." in main_host else "" allowed = set(hosts.values()) for role in (ROLES if hosts else ()): if role in hosts: ok(S, f"{role:6} {hosts[role]}") else: warn(S, f"{role:6} no site block (configure.py --hosts to add one)") if base and f"ovh1.{base}" in allowed: bad(S, f"ovh1.{base} still has a site block") # ---- 1 DNS ------------------------------------------------------------------------------------- S = "1 DNS" if hosts and not args.repo: ips = {} for role, h in hosts.items(): try: ips[h] = sorted(set(a[4][0] for a in socket.getaddrinfo(h, 443, proto=socket.IPPROTO_TCP))) except OSError as e: bad(S, f"{h}: does not resolve ({e})") role_ips = {h: v for r, h in hosts.items() for hh, v in ips.items() if hh == h and r in ROLES} distinct = set(tuple(v) for v in role_ips.values()) for h, v in ips.items(): if h in role_ips: (ok if len(distinct) == 1 else bad)(S, f"{h} -> {', '.join(v)}") else: warn(S, f"{h} -> {', '.join(v)} (apex/other block: a redirect may live anywhere, but Caddy must hold its certificate to answer)") if len(distinct) > 1: bad(S, "the role hosts resolve to different addresses — a CNAME points elsewhere") if base: try: old = sorted(set(a[4][0] for a in socket.getaddrinfo(f"ovh1.{base}", 22, proto=socket.IPPROTO_TCP))) ok(S, f"ovh1.{base} -> {', '.join(old)} (ssh only; fine to keep)") except OSError: warn(S, f"ovh1.{base} no longer resolves — ssh by IP or another name") # ---- 3 .env -------------------------------------------------------------------------------------- S = "3 .env" env = read_env(repo / ".env") if not args.repo: if not env: bad(S, f"{repo / '.env'} unreadable (run with sudo, from the checkout)") for role, key in ENV_KEY.items(): if role not in hosts: continue v = env.get(key, "") if v == hosts[role]: ok(S, f"{key}={v}") elif not v: bad(S, f"{key} unset — expected {hosts[role]}") else: bad(S, f"{key}={v} — Caddyfile says {hosts[role]}") apex, apex_to = env.get("APEX_HOST", ""), env.get("APEX_REDIRECT", "") if apex: found = hosts_in(apex_to); classify(S, f"APEX_REDIRECT ({apex} ->)", found, allowed, base=base) else: ok(S, "APEX_HOST empty (bare domain not served)") # ---- 4 mediamtx.yml ------------------------------------------------------------------------------ S = "4 mediamtx.yml" if not args.repo: mtx = repo / "mediamtx.yml" if mtx.is_file(): m = re.search(r"^webrtcAdditionalHosts:\s*\[([^\]]*)\]", mtx.read_text(), re.M) got = [x.strip().strip("'\"") for x in (m.group(1).split(",") if m else []) if x.strip()] if main_host in got: ok(S, f"webrtcAdditionalHosts contains {main_host}") else: bad(S, f"webrtcAdditionalHosts = {got or 'missing'} — expected {main_host} (then: docker compose restart mediamtx, between streams)") else: bad(S, "mediamtx.yml missing from the checkout") # ---- 5 running containers ------------------------------------------------------------------------ S = "5 containers" if not args.repo: want = {"jellyfin": ("JELLYFIN_PublishedServerUrl", "watch"), "gitea": ("GITEA__server__ROOT_URL", "git"), "act_runner": ("GITEA_INSTANCE_URL", "git"), "discord-bot": ("WEB_PUBLIC_URL", "bot"), "mediamtx": ("LIVE_HOST", "live"), "kibana": ("DIAG_HOST", "diag")} for container, (var, role) in want.items(): if role not in hosts: continue r = sh(["docker", "inspect", "-f", "{{range .Config.Env}}{{println .}}{{end}}", container]) if r.returncode != 0: warn(S, f"{container}: not running or no such container"); continue val = next((l.split("=", 1)[1] for l in r.stdout.splitlines() if l.startswith(var + "=")), None) if val is None: warn(S, f"{container}: {var} not in its environment"); continue if hosts[role] in val: ok(S, f"{container}: {var}={val}") else: bad(S, f"{container}: {var}={val} — expected {hosts[role]}; the container was created before .env changed: docker compose up -d {container}") gd = sh(["docker", "inspect", "-f", "{{range .Config.Env}}{{println .}}{{end}}", "gitea"]) for var in ("GITEA__server__DOMAIN", "GITEA__server__SSH_DOMAIN"): val = next((l.split("=", 1)[1] for l in gd.stdout.splitlines() if l.startswith(var + "=")), None) if val and "git" in hosts: (ok if val == hosts["git"] else bad)(S, f"gitea: {var}={val}" + ("" if val == hosts["git"] else f" — expected {hosts['git']}")) # ---- 6 HTTPS ------------------------------------------------------------------------------------- S = "6 HTTPS" if hosts and not args.repo: auth = hosts.get("auth", "") for role, h in hosts.items(): status, loc, err = curl_head(f"https://{h}/") if err: if role in ROLES: bad(S, f"{h}: {err[:90]} (no certificate yet? Caddy block missing? DNS?)") else: warn(S, f"{h}: {err[:90]} — the bare domain's A record does not reach this box, so its Caddy block cannot get a certificate; fix the record or clear APEX_HOST") continue loc_host = (hosts_in(loc) or [""])[0] if role in GATED: if status in (301, 302, 303, 307, 308) and loc_host == auth: ok(S, f"{h}: {status} -> {auth} (gate)") elif status in (301, 302, 303, 307, 308): bad(S, f"{h}: {status} -> {loc} — gate sends to the wrong host") else: warn(S, f"{h}: {status} without a redirect to {auth} — is the forward-auth gate on?") else: (ok if status < 400 else bad)(S, f"{h}: {status}{(' -> ' + loc) if loc else ''}") # a move in progress: aliases must answer like the primary names; old names must redirect to the new ones alt, red = env.get("ALT_BASE", "").strip().lower(), env.get("REDIRECT_BASE", "").strip().lower() for role, h in hosts.items(): if role not in ROLES or not h.endswith("." + base): continue if alt and alt != base: a = h[: -len(base)] + alt status, loc, err = curl_head(f"https://{a}/") (bad if err else ok)(S, f"alias {a}: {err[:80] if err else status}") if red and red != base: o = h[: -len(base)] + red status, loc, err = curl_head(f"https://{o}/") if err: warn(S, f"old name {o}: {err[:80]} (its certificate may still be issuing)") elif status == 308 and (hosts_in(loc) or [""])[0] == h: ok(S, f"old name {o}: 308 -> {h}") else: bad(S, f"old name {o}: {status} -> {loc or '(no Location)'} — expected 308 to {h}") if base: status, loc, err = curl_head(f"https://ovh1.{base}/", timeout=8) if err: ok(S, f"ovh1.{base}: no https (expected: ssh only)") else: bad(S, f"ovh1.{base} still answers https ({status}) — a Caddy block or another server is serving the old name") # ---- 7 Authentik ---------------------------------------------------------------------------------- S = "9 Authentik" if not args.repo and hosts: AK = r''' import json from urllib.parse import urlparse from authentik.outposts.models import Outpost from authentik.providers.proxy.models import ProxyProvider from authentik.providers.oauth2.models import OAuth2Provider from authentik.sources.oauth.models import OAuthSource from authentik.brands.models import Brand out = {"outposts": [], "proxy": [], "oauth2": [], "sources": [], "brands": []} for o in Outpost.objects.all(): out["outposts"].append({"name": o.name, "authentik_host": (o.config or {}).get("authentik_host", "") if isinstance(o.config, dict) else getattr(o.config, "authentik_host", "")}) for p in ProxyProvider.objects.all(): out["proxy"].append({"name": p.name, "external_host": p.external_host, "cookie_domain": p.cookie_domain, "mode": str(p.mode)}) for p in OAuth2Provider.objects.all(): uris = [] try: uris = [u.url for u in p.redirect_uris] except Exception: try: uris = [u.get("url", "") for u in (p._redirect_uris or [])] except Exception: uris = [str(getattr(p, "redirect_uris", ""))] out["oauth2"].append({"name": p.name, "redirect_uris": uris}) for s in OAuthSource.objects.all(): out["sources"].append({"name": s.name, "slug": s.slug, "provider_type": s.provider_type}) for b in Brand.objects.all(): out["brands"].append({"domain": b.domain, "default": b.default, "logo": str(b.branding_logo), "default_app": str(b.default_application) if b.default_application_id else ""}) print("VGHOSTS " + json.dumps(out)) ''' r = sh(["docker", "exec", "-i", "authentik-server", "ak", "shell", "-c", "import sys; exec(sys.stdin.read())"], input=AK, timeout=120) line = next((l for l in (r.stdout + r.stderr).splitlines() if l.startswith("VGHOSTS ")), None) if not line: bad(S, "could not read Authentik (" + (r.stderr.strip().splitlines() or ["no output"])[-1][:120] + ")") else: ak = json.loads(line[8:]) auth = hosts.get("auth", "") for o in ak["outposts"]: h = hosts_in(o["authentik_host"]) if h == [auth]: ok(S, f"outpost {o['name']!r}: authentik_host {o['authentik_host']}") elif not o["authentik_host"]: warn(S, f"outpost {o['name']!r}: authentik_host unset (the embedded outpost then uses its own address; set it to https://{auth})") else: bad(S, f"outpost {o['name']!r}: authentik_host {o['authentik_host']} — expected https://{auth}") for p in ak["proxy"]: h = hosts_in(p["external_host"]) classify(S, f"proxy provider {p['name']!r} external_host", h, allowed, base=base) cd = p["cookie_domain"] if "domain" in p["mode"].lower(): (ok if cd == base else bad)(S, f"proxy provider {p['name']!r} cookie_domain={cd or '(unset)'}" + ("" if cd == base else f" — expected {base}; sessions will not span the hosts")) for p in ak["oauth2"]: expect = hosts.get("watch") if "jellyfin" in p["name"].lower() else hosts.get("git") if "gitea" in p["name"].lower() else None for u in p["redirect_uris"]: classify(S, f"OAuth2 provider {p['name']!r} redirect", hosts_in(u), allowed, expect=expect, base=base) if not p["redirect_uris"]: warn(S, f"OAuth2 provider {p['name']!r}: no redirect URIs") for b in ak["brands"]: if b["domain"] in ("", "authentik-default"): ok(S, f"brand domain {b['domain']!r} (default)") elif b["domain"] == base or b["domain"].endswith(base): ok(S, f"brand domain {b['domain']}") else: warn(S, f"brand domain {b['domain']} — not under {base}") # ---- 8 Discord -------------------------------------------------------------------------- for s in ak["sources"]: if s["provider_type"] == "discord": manual("10 Discord", f"developer portal → OAuth2 → Redirects must contain exactly: https://{auth}/source/oauth/callback/{s['slug']}/") if not any(s["provider_type"] == "discord" for s in ak["sources"]): bad("10 Discord", "no Discord OAuth source in Authentik") # ---- 9 Jellyfin ----------------------------------------------------------------------------------- S = "7 Jellyfin" if not args.repo and hosts: jf = APPDATA / "jellyfin" / "config" if jf.is_dir(): for xml in sorted(jf.rglob("*.xml")): try: text = xml.read_text(errors="replace") except OSError: continue found = [h for h in hosts_in(text) if base and (h.endswith("." + base) or h == base or h.startswith("ovh1."))] if not found: continue name = xml.name.lower() expect = hosts.get("auth") if "sso" in name else None classify(S, f"{xml.relative_to(jf)}", found, allowed, expect=expect, base=base) else: warn(S, f"{jf} not found") if "watch" in hosts: status, loc, err = curl_head(f"https://{hosts['watch']}/sso/OID/start/authentik") lh = (hosts_in(loc) or [""])[0] if err: bad(S, f"SSO start: {err[:80]}") elif status in (301, 302, 303, 307) and lh == hosts.get("auth"): ok(S, f"SSO start -> {hosts['auth']} (plugin OID endpoint is right)") else: bad(S, f"SSO start answered {status} -> {loc or '(no Location)'} — SSO-Auth plugin OID endpoint should be https://{hosts.get('auth')}/application/o/jellyfin/") # ---- 10 Gitea --------------------------------------------------------------------------------------- S = "8 Gitea" if not args.repo and hosts and "git" in hosts: ini = APPDATA / "gitea" / "gitea" / "conf" / "app.ini" if ini.is_file(): text = ini.read_text(errors="replace") for key in ("ROOT_URL", "DOMAIN", "SSH_DOMAIN"): m = re.search(rf"^{key}\s*=\s*(\S+)", text, re.M) v = m.group(1) if m else "" (ok if hosts["git"] in v else bad)(S, f"app.ini {key}={v or '(missing)'}" + ("" if hosts["git"] in v else f" — expected {hosts['git']} (compose env sets it; recreate gitea)")) else: warn(S, f"{ini} not found") status, loc, err = curl_head(f"https://{hosts['git']}/user/oauth2/authentik") lh = (hosts_in(loc) or [""])[0] if err: bad(S, f"auth source: {err[:80]}") elif lh == hosts.get("auth"): ok(S, f"auth source 'authentik' -> {hosts['auth']} (discovery URL is right)") else: bad(S, f"auth source: {status} -> {loc or '(no Location)'} — Site administration → Authentication sources → authentik → discovery URL https://{hosts.get('auth')}/application/o/gitea/.well-known/openid-configuration") runner = Path("/srv/appdata/act_runner/.runner") if runner.is_file(): classify(S, "runner registration (.runner)", hosts_in(runner.read_text(errors="replace")), allowed, expect=hosts["git"], base=base) else: warn(S, "runner: /srv/appdata/act_runner/.runner not found — ./registerGiteaRunner.sh") r = sh(["git", "-C", str(repo), "remote", "get-url", "origin"]) if r.returncode == 0: url = r.stdout.strip(); m = re.search(r"@([A-Za-z0-9.-]+)", url) h = (m.group(1).lower() if m else "") (ok if h == hosts["git"] else bad)(S, f"server clone remote: {url}" + ("" if h == hosts["git"] else f" — expected {hosts['git']}")) else: warn(S, f"{repo}: not a git clone (deploy.sh pull needs one)") for cfg in (Path.home() / ".ssh" / "config.vanguard", Path("/home/debian/.ssh/config.vanguard")): if cfg.is_file(): classify(S, f"{cfg}", [h for h in re.findall(r"(?im)^\s*host(?:name)?\s+(\S+)", cfg.read_text()) if "." in h], allowed, expect=hosts["git"], base=base); break # ---- 11 repo ---------------------------------------------------------------------------------------- S = "11 repo" for script, var in (("deployLive.sh", "HOST"), ("switchToClone.sh", "REPO_SSH")): p = repo / script if not p.is_file(): warn(S, f"{script} missing"); continue m = re.search(rf"^{var}=.*", p.read_text(), re.M) found = [h for h in re.findall(r"@([A-Za-z0-9.-]+\.[a-z]{2,})", m.group(0))] if m else [] if hosts: classify(S, f"{script} {var}", [h.lower() for h in found], allowed, base=base) else: ok(S, f"{script} {var}: {', '.join(found) or '(none)'}") about = repo / "about" / "index.html" if about.is_file(): labels = sorted(set(re.findall(r"sub\('([a-z]+)'\)", about.read_text()))) if hosts: for label in labels: (ok if label in hosts else bad)(S, f"about page label '{label}'" + ("" if label in hosts else " — no such Caddy role; the gateway link will 404/TLS-fail")) else: ok(S, f"about page labels: {', '.join(labels)}") for doc in ("README.md", "FRESH-INSTALL.md", "STATUS.md", "LOGGING.md"): p = repo / doc if not p.is_file(): continue stale = [(n, l.strip()) for n, l in enumerate(p.read_text().splitlines(), 1) if re.search(r"https?://ovh1\.|ovh1\.[a-z.]+/", l) and not re.search(r"ssh|scp|ssh-only|ssh only", l)] if stale: bad(S, f"{doc}: {len(stale)} line(s) send a browser to ovh1 — first: {stale[0][0]}: {stale[0][1][:80]}") else: ok(S, f"{doc}: no browser links to ovh1") plan = repo / "CommunityDiscordMedia.md" if plan.is_file() and re.search(r"https?://ovh1\.", plan.read_text()): warn(S, "CommunityDiscordMedia.md still describes ovh1 as a web host (historical plan; mark the lines as superseded or leave)") # ---- 12 streamers ------------------------------------------------------------------------------- if hosts.get("live"): manual("12 streamers", f"OBS → Docks → Custom Browser Docks: https://{hosts['live']}/dock/ ; dock Settings re-saved; Timecode band → \"Add band + clock source\" again; manual RTMP/SRT configs re-copied from the dock") manual("12 streamers", f"Jellyfin apps on phones/TVs: server https://{hosts.get('watch','')} ; Discord pins: https://{main_host}/ (members), https://{hosts.get('clips','')}/ (gamers), https://{hosts.get('files','')}/ (editors)") return report(args) def report(args): order = lambda s: int(s.split()[0]) sections = sorted(set(s for s, _, _ in results), key=order) if args.json: print(json.dumps([{"section": s, "status": st, "text": t} for s, st, t in results], indent=1)); return 0 mark = {"PASS": " ok ", "FAIL": " FAIL ", "WARN": " warn ", "MANUAL": " todo "} for s in sections: print(f"\n== {s}") for sec, st, t in results: if sec == s: print(f" [{mark[st]}] {t}") fails = sum(1 for _, st, _ in results if st == "FAIL"); warns = sum(1 for _, st, _ in results if st == "WARN") todo = sum(1 for _, st, _ in results if st == "MANUAL") print(f"\n{fails} failed, {warns} warnings, {todo} to check by hand. Fix top-down: an early failure explains the later ones.") return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())