1406 lines
78 KiB
Python
Executable File
1406 lines
78 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Dump-tonight endpoint. Moves one streamer's night into trash (soft delete).
|
|
|
|
POST /dump identity from X-Authentik-Username / X-Authentik-Groups (set by Caddy after Authentik)
|
|
body (optional JSON): {"night":"YYYY-MM-DD","user":"name"}
|
|
GET /dump -> lists what a dump would move (dry run); ?user=all&night=YYYY-MM-DD select whose and when
|
|
Two doors, different reach: /dump on live. (the dock's Dump tonight) is ALWAYS the caller's own night,
|
|
admin or not — "user" is ignored there. /streaming/api/dump on diag. (the streaming page's Dump tab) is
|
|
where admins pick a streamer or "all"; non-admins still get only their own there.
|
|
GET /me/stream-token -> short-lived publish token for the caller (member of streamers/admins); the dock
|
|
and the live portal set it as the OBS/WHIP credential. No stored passwords anywhere.
|
|
GET /me/clips -> Syncthing self-service status for the caller: our device id, their folder/device state
|
|
POST /me/clips -> {"device_id": "..."}: create clips/<user> (+ .stignore), add their device to our Syncthing,
|
|
create the receive-only folder clips-<user> shared with them (min free 300 GB). Idempotent.
|
|
POST /deploy -> Gitea push webhook (bridge port only, HMAC-verified with DEPLOY_WEBHOOK_SECRET from
|
|
/etc/vanguard/deploy.env). Runs /usr/local/bin/vanguard-deploy pull ONLY when DEPLOY_AUTO=1;
|
|
otherwise logs the push and answers 202 "auto-deploy disabled".
|
|
POST /heartbeat -> LEGACY, nothing calls this now. It was the Watchdog route when Alertmanager, and then a
|
|
Kibana webhook connector, posted here. A basic Elastic licence has no webhook connector, so
|
|
Kibana's rules write into the vanguard-alerts index instead and the deadman reads the
|
|
heartbeat from there. Kept because it costs nothing and is a ready-made target if an
|
|
external monitor is ever added.
|
|
POST /deploy-bot -> Gitea Actions (job container -> host gateway) hands over a Discord-bot release bundle,
|
|
HMAC-SHA256 signed with BOT_DEPLOY_SECRET; installs it with vanguard-deploy-bot.
|
|
May carry X-Vanguard-Env (base64 JSON): the bot's own settings, kept in Gitea, written
|
|
into the stack .env for the keys BOT_ENV_KEYS allows. Signature covers header+body.
|
|
GET /me/stream-health -> the caller's live streams with B-frame depth: the dock warns when WebRTC will refuse
|
|
GET /streaming/api/live -> what is publishing, from where, loss on the way in, recorder restarts, transport history, blocks
|
|
POST /streaming/api/live -> {"action":"stop","path","kind","session","minutes"}: kick that publish session via MediaMTX's
|
|
API and block the account's publishing for N minutes (0 = kick only); {"action":"unblock","user"}. Admins only.
|
|
/mtx-auth denies a blocked account, so neither OBS's auto-reconnect nor a Go Live restart gets around it.
|
|
GET /streaming/api/stuck -> every unpublished recording and why it is stuck (any gated user)
|
|
GET /streaming/api/disk -> disk use by streamer / clipper / trash by month, system dirs, Docker, and logs (ES indices
|
|
by month, journald, Docker json logs). Cached 10 min; ?refresh=1 rewalks.
|
|
POST /streaming/api/stuck -> {"action":"retry"|"drop","src":...}: clear the blocker, or move it to trash with the
|
|
usual 3-day undo; {"action":"drop_under","mb":N} bins every stream fragment under N MB; {"action":"wipe"}
|
|
bins everything unprocessed (not being written). Admin groups only; every action runs bin/vgstuck.sh
|
|
so the web page and the CLI share one implementation.
|
|
GET /metrics -> Prometheus text for Metricbeat: the media scripts' counters (bin/vgmetrics.sh) plus live
|
|
gauges for the normalise queue and MediaMTX. Loopback listener only; Caddy routes nothing here.
|
|
POST /mtx-auth -> MediaMTX authHTTP hook: allows read/playback; allows publish iff user/pass carries a valid
|
|
token whose user owns the path prefix. Tokens are HMAC-signed with /etc/vanguard/stream-token.secret.
|
|
|
|
"Night" = everything the user recorded from 17:00 Eastern on <night> until 02:00 Eastern the next day.
|
|
Default night: today if it is 17:00 or later, else yesterday (so 00:00-02:00 still counts as last night).
|
|
Users may dump only their own prefix; members of the admins / media-admins groups may pass "user" to dump anyone, or "all",
|
|
but only through /streaming/api/dump (the diag triage page). On live.'s /dump everyone dumps their own night only.
|
|
"""
|
|
import base64, hashlib, hmac, json, os, re, shutil, subprocess, sys, threading
|
|
import time as _time
|
|
import urllib.parse
|
|
from datetime import datetime, timedelta, time
|
|
from zoneinfo import ZoneInfo
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(os.environ.get("MEDIA_ROOT", "/srv/media"))
|
|
SRC_DIRS = [ROOT / "streams", ROOT / "clips"]
|
|
TRASH = ROOT / "trash"
|
|
MARK = ROOT / ".markers"
|
|
TOKEN_SECRET_FILE = Path(os.environ.get("STREAM_TOKEN_SECRET_FILE", "/etc/vanguard/stream-token.secret"))
|
|
TOKEN_TTL_H = float(os.environ.get("STREAM_TOKEN_TTL_HOURS", "12"))
|
|
PUBLISH_GROUPS = set(g.strip() for g in os.environ.get("PUBLISH_GROUPS", "streamers,admins,media-admins").split(",") if g.strip())
|
|
AUTH_USER_HEADER = os.environ.get("AUTH_USER_HEADER", "X-Authentik-Username")
|
|
AUTH_GROUPS_HEADER = os.environ.get("AUTH_GROUPS_HEADER", "X-Authentik-Groups")
|
|
ADMIN_GROUPS = set(g.strip() for g in os.environ.get("DUMP_ADMIN_GROUPS", "admins,media-admins").split(",") if g.strip())
|
|
LISTEN = ("127.0.0.1", 8080) # /dump, /streaming/api/*, /me/* — only via Caddy (identity headers)
|
|
AUTH_LISTEN = (os.environ.get("MTX_AUTH_BIND", "0.0.0.0"), int(os.environ.get("MTX_AUTH_PORT", "8081"))) # /mtx-auth only — reachable from Docker's bridge (firewall), used by MediaMTX
|
|
TZ = ZoneInfo(os.environ.get("DUMP_TZ", "America/New_York"))
|
|
NIGHT_START = time(17, 0) # 5 pm
|
|
NIGHT_END = time(2, 0) # 2 am next day
|
|
|
|
AUTH_JWT_HEADER = os.environ.get("AUTH_JWT_HEADER", "X-Authentik-Jwt")
|
|
|
|
def proxy_claim(headers, name):
|
|
"""One claim out of the token the proxy injects, or None.
|
|
|
|
Authentik's proxy puts its signed token in X-Authentik-Jwt; a scope mapping on the provider adds the claims
|
|
we care about (authentik/vanguard-profile-scope-mapping.py supplies discord_id). The payload is only read,
|
|
never trusted beyond what the sibling headers already assert: all of them come from the same proxy, which is
|
|
the only thing that can reach this port. Returned for display, never for an authorisation decision here.
|
|
"""
|
|
token = (headers.get(AUTH_JWT_HEADER) or "").strip()
|
|
if token.count(".") != 2:
|
|
return None
|
|
body = token.split(".")[1]
|
|
try:
|
|
claims = json.loads(base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
value = claims.get(name)
|
|
return str(value) if value not in (None, "") else None
|
|
|
|
def identity(headers):
|
|
"""(username, is_admin) from the headers Caddy sets after Authentik; (None, False) if absent.
|
|
dumpd binds to loopback only, so these headers can only come from Caddy."""
|
|
who = (headers.get(AUTH_USER_HEADER) or "").strip().lower()
|
|
if not who:
|
|
return None, False
|
|
groups = {g.strip() for g in re.split(r"[|,]", headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()}
|
|
return who, bool(groups & ADMIN_GROUPS)
|
|
|
|
def default_night(now=None):
|
|
now = now or datetime.now(TZ)
|
|
d = now.date() if now.time() >= NIGHT_START else (now - timedelta(days=1)).date()
|
|
return d.isoformat()
|
|
|
|
def night_window(night):
|
|
d = datetime.strptime(night, "%Y-%m-%d").date()
|
|
return (datetime.combine(d, NIGHT_START, tzinfo=TZ),
|
|
datetime.combine(d + timedelta(days=1), NIGHT_END, tzinfo=TZ))
|
|
|
|
def candidates(user, night):
|
|
start, end = night_window(night)
|
|
out = []
|
|
for base in SRC_DIRS:
|
|
udirs = [p for p in base.iterdir() if p.is_dir()] if user == "all" else [base / user]
|
|
for ud in udirs:
|
|
if not ud.is_dir():
|
|
continue
|
|
for f in ud.rglob("*"):
|
|
if not f.is_file():
|
|
continue
|
|
mt = datetime.fromtimestamp(f.stat().st_mtime, TZ)
|
|
if start <= mt <= end:
|
|
out.append(f)
|
|
return out
|
|
|
|
def do_move(files, night):
|
|
moved, size = 0, 0
|
|
for f in files:
|
|
rel = f.relative_to(ROOT)
|
|
dst = TRASH / night / rel
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
size += f.stat().st_size
|
|
shutil.move(str(f), str(dst))
|
|
moved += 1
|
|
except FileNotFoundError:
|
|
pass
|
|
# drop now-empty title folders
|
|
for base in SRC_DIRS:
|
|
for d in sorted((p for p in base.rglob("*") if p.is_dir()), key=lambda p: -len(p.parts)):
|
|
if d.parent != base and not any(d.iterdir()):
|
|
d.rmdir()
|
|
return moved, size
|
|
|
|
# ------------------------------------------------------------------ Syncthing (self-service clip intake)
|
|
import urllib.request, urllib.error
|
|
ST_URL = os.environ.get("SYNCTHING_URL", "http://127.0.0.1:8384")
|
|
ST_CONFIG = Path(os.environ.get("SYNCTHING_CONFIG", "/srv/media/.syncthing/config.xml"))
|
|
ST_CLIPS_PREFIX = os.environ.get("SYNCTHING_CLIPS_PREFIX", "/var/syncthing/clips") # path INSIDE the syncthing container
|
|
STIGNORE_SRC = Path(os.environ.get("STIGNORE_SRC", "/etc/vanguard/stignore"))
|
|
MIN_FREE_GB = int(os.environ.get("CLIPS_MIN_FREE_GB", "300"))
|
|
CLIPS_QUOTA_GB = float(os.environ.get("CLIPS_QUOTA_GB", "100")) # default per-user quota for clips/<user>
|
|
CLIPS_QUOTA_FILE = Path(os.environ.get("CLIPS_QUOTA_FILE", "/etc/vanguard/clip-quotas.json")) # {"user": GB, ...} overrides
|
|
QUOTA_CHECK_S = int(os.environ.get("CLIPS_QUOTA_CHECK_S", "300"))
|
|
QUOTA_STATE = Path(os.environ.get("CLIPS_QUOTA_STATE", "/srv/media/.markers/clip-quota-paused.json"))
|
|
|
|
def quota_for(user):
|
|
try:
|
|
q = json.loads(CLIPS_QUOTA_FILE.read_text()).get(user)
|
|
if q is not None: return float(q)
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
pass
|
|
return CLIPS_QUOTA_GB
|
|
|
|
def folder_bytes(path):
|
|
total = 0
|
|
for root, dirs, files in os.walk(path):
|
|
for f in files:
|
|
try: total += os.lstat(os.path.join(root, f)).st_size
|
|
except OSError: pass
|
|
return total
|
|
|
|
def quota_state():
|
|
try: return json.loads(QUOTA_STATE.read_text())
|
|
except (OSError, json.JSONDecodeError): return {}
|
|
|
|
def quota_enforce_once():
|
|
"""Pause a user's Syncthing folder when clips/<user> exceeds their quota; resume when back under 90 %."""
|
|
paused = quota_state(); changed = False
|
|
try:
|
|
folders = st("GET", "/rest/config/folders")
|
|
except Exception as e:
|
|
log("quota_unreachable", level="warn", error=e); return
|
|
for f in folders:
|
|
if not f["id"].startswith("clips-"): continue
|
|
user = f["id"][len("clips-"):]
|
|
path = Path(ROOT) / "clips" / user
|
|
if not path.is_dir(): continue
|
|
used = folder_bytes(path); q = quota_for(user) * 1e9
|
|
over = used > q; under = used < q * 0.9
|
|
if over and not f.get("paused"):
|
|
try: st("POST", f"/rest/db/pause?folder={f['id']}"); paused[user] = {"at": datetime.now(TZ).isoformat(), "used_gb": round(used / 1e9, 1), "quota_gb": q / 1e9}; changed = True
|
|
except Exception as e: log("quota_pause_failed", level="error", user=user, error=e); continue
|
|
log("quota_paused", level="warn", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}", quota_gb=f"{q/1e9:.0f}")
|
|
elif under and f.get("paused") and user in paused:
|
|
try: st("POST", f"/rest/db/resume?folder={f['id']}"); paused.pop(user, None); changed = True
|
|
except Exception as e: log("quota_resume_failed", level="error", user=user, error=e); continue
|
|
log("quota_resumed", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}")
|
|
if changed:
|
|
try: QUOTA_STATE.parent.mkdir(exist_ok=True); QUOTA_STATE.write_text(json.dumps(paused, indent=2))
|
|
except OSError: pass
|
|
|
|
def quota_loop():
|
|
while True:
|
|
try: quota_enforce_once()
|
|
except Exception as e: log("quota_loop_error", level="error", error=e)
|
|
_time.sleep(QUOTA_CHECK_S)
|
|
|
|
def st_apikey():
|
|
try:
|
|
m = re.search(r"<apikey>([^<]+)</apikey>", ST_CONFIG.read_text())
|
|
return m.group(1) if m else None
|
|
except OSError:
|
|
return None
|
|
|
|
def st(method, path, body=None):
|
|
key = st_apikey()
|
|
if not key: raise RuntimeError("syncthing api key not found")
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(ST_URL + path, data=data, method=method, headers={"X-API-Key": key, "Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
raw = r.read()
|
|
return json.loads(raw) if raw else None
|
|
|
|
DEVICE_ID_RE = re.compile(r"^[A-Z2-7]{7}(-[A-Z2-7]{7}){7}$")
|
|
|
|
def clips_status(user):
|
|
out = {"user": user, "folder_id": f"clips-{user}", "server_path": f"{ROOT}/clips/{user}"}
|
|
try:
|
|
out["server_device_id"] = st("GET", "/rest/system/status")["myID"]
|
|
cfg_devices = st("GET", "/rest/config/devices")
|
|
cfg_folders = st("GET", "/rest/config/folders")
|
|
folder = next((f for f in cfg_folders if f["id"] == out["folder_id"]), None)
|
|
out["folder_configured"] = folder is not None
|
|
theirs = [d["deviceID"] for d in (folder or {}).get("devices", []) if d["deviceID"] != out["server_device_id"]]
|
|
out["device_id"] = theirs[0] if theirs else None
|
|
if out["device_id"]:
|
|
conns = st("GET", "/rest/system/connections")["connections"]
|
|
out["device_connected"] = bool(conns.get(out["device_id"], {}).get("connected"))
|
|
out["device_name"] = next((d["name"] for d in cfg_devices if d["deviceID"] == out["device_id"]), None)
|
|
try:
|
|
comp = st("GET", f"/rest/db/completion?folder={out['folder_id']}&device={out['device_id']}")
|
|
out["remote_completion"] = comp.get("completion")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
stt = st("GET", f"/rest/db/status?folder={out['folder_id']}")
|
|
out["files_here"] = stt.get("localFiles"); out["state"] = stt.get("state")
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
out["error"] = f"syncthing: {e}"
|
|
out["folder_exists"] = Path(out["server_path"]).is_dir()
|
|
out["quota_gb"] = quota_for(user)
|
|
out["used_gb"] = round(folder_bytes(out["server_path"]) / 1e9, 2) if out["folder_exists"] else 0.0
|
|
ps = quota_state().get(user)
|
|
out["paused_for_quota"] = bool(ps)
|
|
if ps: out["paused_since"] = ps.get("at")
|
|
return out
|
|
|
|
def clips_setup(user, device_id, device_name=None):
|
|
device_id = device_id.strip().upper()
|
|
if not DEVICE_ID_RE.match(device_id):
|
|
raise ValueError("that does not look like a Syncthing Device ID (8 groups of 7 characters, e.g. ABCDEFG-HIJKLMN-...)")
|
|
# 1. server folder with the video-only ignore list
|
|
d = Path(ROOT) / "clips" / user
|
|
d.mkdir(parents=True, exist_ok=True); shutil.chown(d, 1000, 1000)
|
|
ign = d / ".stignore"
|
|
if not ign.exists() and STIGNORE_SRC.exists():
|
|
ign.write_text(STIGNORE_SRC.read_text()); shutil.chown(ign, 1000, 1000); ign.chmod(0o644)
|
|
loc = d / ".stlocal" # #include'd by .stignore; Syncthing refuses to scan while it is missing
|
|
if not loc.exists():
|
|
loc.write_text(""); shutil.chown(loc, 1000, 1000); loc.chmod(0o644)
|
|
# 2. their device
|
|
my_id = st("GET", "/rest/system/status")["myID"]
|
|
if device_id == my_id: raise ValueError("that is the server's own Device ID — paste yours (your Syncthing → Actions → Show ID)")
|
|
devices = st("GET", "/rest/config/devices")
|
|
if not any(x["deviceID"] == device_id for x in devices):
|
|
st("POST", "/rest/config/devices", {"deviceID": device_id, "name": device_name or user, "addresses": ["dynamic"],
|
|
"autoAcceptFolders": False, "introducer": False, "paused": False})
|
|
# 3. receive-only folder shared with them
|
|
fid = f"clips-{user}"
|
|
folders = st("GET", "/rest/config/folders")
|
|
cur = next((f for f in folders if f["id"] == fid), None)
|
|
spec = {"id": fid, "label": f"Clips — {user}", "path": f"{ST_CLIPS_PREFIX}/{user}", "type": "receiveonly",
|
|
"devices": [{"deviceID": my_id}, {"deviceID": device_id}],
|
|
"minDiskFree": {"value": MIN_FREE_GB, "unit": "GB"}, "fsWatcherEnabled": True, "rescanIntervalS": 3600,
|
|
"ignorePerms": True, "ignoreDelete": False}
|
|
if cur is None:
|
|
st("POST", "/rest/config/folders", spec)
|
|
else:
|
|
if not any(x["deviceID"] == device_id for x in cur.get("devices", [])):
|
|
cur["devices"].append({"deviceID": device_id})
|
|
cur["type"] = "receiveonly"; cur["minDiskFree"] = spec["minDiskFree"]
|
|
st("PUT", f"/rest/config/folders/{fid}", cur)
|
|
log("clips_setup", user=user, device=device_id[:7], folder=fid)
|
|
return clips_status(user)
|
|
|
|
DEPLOY_ENV = Path(os.environ.get("DEPLOY_ENV_FILE", "/etc/vanguard/deploy.env"))
|
|
def deploy_env():
|
|
d = {}
|
|
try:
|
|
for line in DEPLOY_ENV.read_text().splitlines():
|
|
if "=" in line and not line.startswith("#"): k, v = line.split("=", 1); d[k.strip()] = v.strip()
|
|
except OSError:
|
|
pass
|
|
return d
|
|
|
|
BOT_ENV_DEFAULT_KEYS = "DISCORD_BOT_TOKEN,DISCORD_GUILD_ID"
|
|
BOT_ENV_MAX_HEADER = 8192
|
|
_env_write_lock = threading.Lock()
|
|
|
|
def stack_env_path():
|
|
"""The stack's .env — docker compose interpolates ${...} from it, so the bot's settings live there."""
|
|
repo = deploy_env().get("DEPLOY_REPO_DIR", "")
|
|
return (Path(repo) / ".env") if repo else None
|
|
|
|
def apply_bot_env(values):
|
|
"""Merge allowed keys into the stack .env. Returns (applied, refused); never logs a value."""
|
|
allowed = {k.strip() for k in deploy_env().get("BOT_ENV_KEYS", BOT_ENV_DEFAULT_KEYS).split(",") if k.strip()}
|
|
refused = sorted(k for k in values if k not in allowed)
|
|
wanted = {k: str(v) for k, v in values.items() if k in allowed}
|
|
if not wanted:
|
|
return [], refused
|
|
path = stack_env_path()
|
|
if path is None or not path.is_file():
|
|
raise OSError("DEPLOY_REPO_DIR is not set or holds no .env")
|
|
bad = [k for k, v in wanted.items() if not re.fullmatch(r"[A-Z][A-Z0-9_]{1,48}", k) or "\n" in v or "\r" in v]
|
|
if bad:
|
|
raise ValueError(f"not a usable env assignment: {bad}")
|
|
with _env_write_lock:
|
|
lines = path.read_text().splitlines()
|
|
seen = set()
|
|
for i, line in enumerate(lines):
|
|
if "=" in line and not line.lstrip().startswith("#"):
|
|
key = line.split("=", 1)[0].strip()
|
|
if key in wanted:
|
|
lines[i] = f"{key}={wanted[key]}"; seen.add(key)
|
|
for key in wanted:
|
|
if key not in seen: lines.append(f"{key}={wanted[key]}")
|
|
stat = path.stat()
|
|
tmp = path.with_suffix(".env.incoming")
|
|
tmp.write_text("\n".join(lines) + "\n")
|
|
os.chmod(tmp, stat.st_mode & 0o777); os.chown(tmp, stat.st_uid, stat.st_gid)
|
|
os.replace(tmp, path)
|
|
applied = sorted(wanted)
|
|
log("bot_env_applied", keys=",".join(applied), refused=",".join(refused) or None)
|
|
return applied, refused
|
|
|
|
HEARTBEAT_FILE = Path(os.environ.get("ALERT_HEARTBEAT_FILE", "/srv/media/.markers/alert-heartbeat"))
|
|
|
|
_bot_deploy_lock = threading.Lock()
|
|
BOT_DEPLOY_MAX_BYTES = 300 * 1024 * 1024
|
|
BOT_DEPLOY_INSTALLER = os.environ.get("BOT_DEPLOY_INSTALLER", "/usr/local/bin/vanguard-deploy-bot")
|
|
|
|
def install_bot_bundle(raw, reason):
|
|
"""Write the uploaded bundle to a temp file and let vanguard-deploy-bot verify, swap and restart it."""
|
|
import tempfile
|
|
if not _bot_deploy_lock.acquire(blocking=False):
|
|
return 409, {"error": "a bot deploy is already running"}
|
|
tmp = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(prefix="vanguard-bot-bundle.", suffix=".bin", delete=False) as handle:
|
|
handle.write(raw); tmp = handle.name
|
|
log("bot_deploy_start", bytes=len(raw), reason=reason)
|
|
done = subprocess.run([BOT_DEPLOY_INSTALLER, tmp], capture_output=True, text=True, timeout=600)
|
|
output = (done.stdout + done.stderr).strip().splitlines()
|
|
log("bot_deploy_exit", level=("info" if done.returncode == 0 else "error"), rc=done.returncode, reason=reason)
|
|
return (200 if done.returncode == 0 else 500), {"installed": done.returncode == 0, "rc": done.returncode,
|
|
"output": output[-40:]}
|
|
except subprocess.TimeoutExpired:
|
|
log("bot_deploy_timeout", level="error", reason=reason)
|
|
return 504, {"error": "the installer did not finish within 10 minutes"}
|
|
except OSError as error:
|
|
log("bot_deploy_error", level="error", error=error)
|
|
return 500, {"error": str(error)}
|
|
finally:
|
|
if tmp:
|
|
try: os.unlink(tmp)
|
|
except OSError: pass
|
|
_bot_deploy_lock.release()
|
|
|
|
_deploy_lock = {"running": False}
|
|
def run_deploy(reason):
|
|
if _deploy_lock["running"]: return False
|
|
_deploy_lock["running"] = True
|
|
def go():
|
|
try:
|
|
log("deploy_start", reason=reason)
|
|
r = subprocess.run(["/usr/local/bin/vanguard-deploy", "pull"], capture_output=True, text=True, timeout=1800)
|
|
for line in (r.stdout + r.stderr).splitlines()[-40:]: print(f"deploy| {line}", flush=True)
|
|
log("deploy_exit", level=("info" if r.returncode == 0 else "error"), rc=r.returncode)
|
|
except Exception as e:
|
|
log("deploy_error", level="error", error=e)
|
|
finally:
|
|
_deploy_lock["running"] = False
|
|
threading.Thread(target=go, daemon=True).start()
|
|
return True
|
|
|
|
def log(event, level="info", **kv):
|
|
"""One logfmt line per event (see LOGGING.md): ts= level= component=dumpd event= key=value ..."""
|
|
def q(v):
|
|
v = str(v)
|
|
return v if v and not re.search(r'[\s"=]', v) else '"' + v.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
|
print(f"ts={datetime.now(TZ).isoformat(timespec='seconds')} level={level} component=dumpd event={event} "
|
|
+ " ".join(f"{k}={q(v)}" for k, v in kv.items() if v is not None), flush=True)
|
|
|
|
def token_secret():
|
|
try:
|
|
sec = TOKEN_SECRET_FILE.read_bytes().strip()
|
|
if len(sec) >= 16: return sec
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
def mint_token(user):
|
|
"""user|exp signed: <b64url(user|exp)>.<hmac hex[:40]> — URL/RTMP/SRT-safe characters only."""
|
|
sec = token_secret()
|
|
if not sec: raise RuntimeError("no stream token secret")
|
|
exp = int(_time.time() + TOKEN_TTL_H * 3600)
|
|
payload = f"{user}|{exp}".encode()
|
|
sig = hmac.new(sec, payload, hashlib.sha256).hexdigest()[:40]
|
|
return base64.urlsafe_b64encode(payload).decode().rstrip("=") + "." + sig, exp
|
|
|
|
def verify_token(token):
|
|
"""-> username or None"""
|
|
sec = token_secret()
|
|
if not sec or not token or "." not in token: return None
|
|
b64, sig = token.rsplit(".", 1)
|
|
try:
|
|
payload = base64.urlsafe_b64decode(b64 + "=" * (-len(b64) % 4))
|
|
except Exception:
|
|
return None
|
|
if not hmac.compare_digest(hmac.new(sec, payload, hashlib.sha256).hexdigest()[:40], sig): return None
|
|
try:
|
|
user, exp = payload.decode().split("|", 1)
|
|
if int(exp) < _time.time(): return None
|
|
except Exception:
|
|
return None
|
|
return user.lower()
|
|
|
|
|
|
|
|
# ---- stream health: B-frames ------------------------------------------------------------------------
|
|
# MediaMTX refuses to serve H.264 with B-frames over WebRTC ("WebRTC doesn't support H264 streams with
|
|
# B-frames"), so an OBS encoder left on its default of 2 B-frames is unwatchable on the portal. Nothing on the
|
|
# publish path can see this: obs-websocket cannot set encoder settings in OBS's Advanced output mode, and
|
|
# MediaMTX only notices when a viewer tries. The recording can: ffprobe reads the reorder depth off the first
|
|
# fragments of the growing .live file in ~50 ms with a small probesize. Cached per path for 30 s.
|
|
_health_cache = {}
|
|
|
|
def stream_health(path):
|
|
"""{'bframes', 'profile', 'res', 'fps', 'file'} for a live path, or None if nothing is on disk yet."""
|
|
now = _time.time()
|
|
hit = _health_cache.get(path)
|
|
if hit and now - hit["at"] < 30:
|
|
return hit["data"]
|
|
live_dir = ROOT / "streams" / path / ".live"
|
|
try:
|
|
newest = max((f for f in live_dir.glob("*.mp4")), key=lambda f: f.stat().st_mtime, default=None)
|
|
except OSError:
|
|
newest = None
|
|
data = None
|
|
if newest is not None:
|
|
# One probe, fields by name: csv column order follows ffprobe's internal field order, not the list.
|
|
done = subprocess.run(["ffprobe", "-v", "error", "-probesize", "2M", "-analyzeduration", "2M",
|
|
"-select_streams", "v:0",
|
|
"-show_entries", "stream=has_b_frames,profile,width,height,r_frame_rate",
|
|
"-of", "default=noprint_wrappers=1", str(newest)],
|
|
capture_output=True, text=True, timeout=10)
|
|
fields = dict(line.split("=", 1) for line in done.stdout.splitlines() if "=" in line)
|
|
if fields:
|
|
try:
|
|
num, den = (fields.get("r_frame_rate", "0/1").split("/") + ["1"])[:2]
|
|
fps = round(float(num) / float(den), 2) if float(den) else None
|
|
except (ValueError, ZeroDivisionError):
|
|
fps = None
|
|
data = {"bframes": int(fields.get("has_b_frames") or 0), "profile": fields.get("profile", ""),
|
|
"res": f"{fields.get('width', '?')}x{fields.get('height', '?')}", "fps": fps, "file": newest.name}
|
|
_health_cache[path] = {"at": now, "data": data}
|
|
return data
|
|
|
|
# ---- live streaming status ------------------------------------------------------------------------
|
|
# GET /streaming/api/live: what is publishing, from where, how much it has lost on the way in, and why
|
|
# MediaMTX has been opening recording files. Cached 5 s: the Live tab polls every 10 s and several admins
|
|
# may have it open. Reads MediaMTX's API and metrics on loopback, and its container log for the counters.
|
|
_live_cache = {"at": 0.0, "data": None}
|
|
MTX_METRICS = os.environ.get("MTX_METRICS_URL", "http://127.0.0.1:9998/metrics")
|
|
|
|
# WebRTC publish sessions and their transport. MediaMTX's API labels the selected ICE pair as
|
|
# "<type>/<protocol>/<addr>/<port>" (host/udp/… or host/tcp/…), so the protocol a publisher is really on is
|
|
# right there — that is how the Live tab tells a Go Live on TCP (the lossy-uplink switch) from one on UDP.
|
|
# A session history is kept in MARK/webrtc-history.json so "when did he switch" survives the page being closed:
|
|
# it is updated on every Live-tab poll AND every metrics scrape (Metricbeat, every few seconds), and a session
|
|
# that is no longer listed is stamped ended. 24 h, 50 sessions per path.
|
|
HISTORY_FILE = MARK / "webrtc-history.json"
|
|
_history_lock = threading.Lock()
|
|
|
|
def webrtc_sessions():
|
|
with urllib.request.urlopen(f"{MTX_API}/v3/webrtcsessions/list?itemsPerPage=1000", timeout=3) as response:
|
|
return json.load(response).get("items") or []
|
|
|
|
def session_transport(item):
|
|
label = item.get("localCandidate") or item.get("remoteCandidate") or ""
|
|
parts = label.split("/")
|
|
return parts[1].lower() if len(parts) >= 2 and parts[1] else ""
|
|
|
|
def track_sessions(items):
|
|
"""Fold the current publish sessions into the history file; returns the history (path -> sessions)."""
|
|
now = datetime.now(TZ)
|
|
with _history_lock:
|
|
try:
|
|
hist = json.loads(HISTORY_FILE.read_text())
|
|
except (OSError, ValueError):
|
|
hist = {}
|
|
live_ids = set()
|
|
for it in items:
|
|
if str(it.get("state", "")).lower() != "publish":
|
|
continue
|
|
sid = it.get("id", ""); live_ids.add(sid)
|
|
rows = hist.setdefault(it.get("path", "?"), [])
|
|
row = next((r for r in rows if r["id"] == sid), None)
|
|
if row is None:
|
|
row = {"id": sid, "created": it.get("created"), "remote": it.get("remoteAddr"), "transport": "", "ended": None}
|
|
rows.append(row)
|
|
if session_transport(it): row["transport"] = session_transport(it)
|
|
row["received"] = it.get("rtpPacketsReceived", 0); row["lost"] = it.get("rtpPacketsLost", 0)
|
|
row["bytes"] = it.get("bytesReceived", 0); row["last_seen"] = now.isoformat(timespec="seconds")
|
|
cutoff = now - timedelta(hours=24)
|
|
for path in list(hist):
|
|
kept = []
|
|
for r in hist[path]:
|
|
if r["id"] not in live_ids and not r.get("ended"):
|
|
r["ended"] = r.get("last_seen") or now.isoformat(timespec="seconds")
|
|
try:
|
|
when = datetime.fromisoformat((r.get("ended") or r.get("last_seen") or "").replace("Z", "+00:00"))
|
|
if when.tzinfo is None: when = when.replace(tzinfo=TZ)
|
|
except ValueError:
|
|
when = now
|
|
if when >= cutoff: kept.append(r)
|
|
kept.sort(key=lambda r: r.get("created") or "")
|
|
hist[path] = kept[-50:]
|
|
if not hist[path]: del hist[path]
|
|
try:
|
|
MARK.mkdir(exist_ok=True)
|
|
tmp = HISTORY_FILE.with_suffix(".tmp"); tmp.write_text(json.dumps(hist)); os.replace(tmp, HISTORY_FILE)
|
|
except OSError as error:
|
|
log("history_write_failed", level="warn", error=str(error))
|
|
return hist
|
|
|
|
def session_history():
|
|
"""History as the page wants it: one list, newest first, with switch markers."""
|
|
try:
|
|
hist = json.loads(HISTORY_FILE.read_text())
|
|
except (OSError, ValueError):
|
|
return []
|
|
out = []
|
|
for path, rows in hist.items():
|
|
prev = None
|
|
for r in rows:
|
|
entry = {"path": path, **{k: r.get(k) for k in ("id", "created", "ended", "remote", "transport", "received", "lost", "bytes")}}
|
|
entry["id"] = (entry["id"] or "")[:8]
|
|
if prev is not None and prev.get("transport") and r.get("transport") and prev["transport"] != r["transport"]:
|
|
entry["switched_from"] = prev["transport"]
|
|
prev = r
|
|
out.append(entry)
|
|
out.sort(key=lambda e: e.get("created") or "", reverse=True)
|
|
return out
|
|
|
|
# Turning a stream off from the Live tab. Two parts, because a kick alone is a hiccup: OBS reconnects on its
|
|
# own within seconds and Go Live can be restarted with one click. So "stop" = kick the publish session through
|
|
# MediaMTX's API AND put the streamer on a block list that /mtx-auth consults for every publish, for a chosen
|
|
# number of minutes (0 = kick only). Blocks live in MARK/publish-blocks.json, survive a dumpd restart, expire on
|
|
# their own, and the Live tab lists them with an unblock button.
|
|
BLOCKS_FILE = MARK / "publish-blocks.json"
|
|
_blocks_lock = threading.Lock()
|
|
KICK_ENDPOINT = {"webrtc": "webrtcsessions", "rtmp": "rtmpconns", "srt": "srtconns", "rtsp": "rtspsessions"}
|
|
|
|
def read_blocks():
|
|
try:
|
|
blocks = json.loads(BLOCKS_FILE.read_text())
|
|
except (OSError, ValueError):
|
|
blocks = {}
|
|
now = _time.time()
|
|
live = {u: b for u, b in blocks.items() if b.get("until", 0) > now}
|
|
if len(live) != len(blocks):
|
|
write_blocks(live)
|
|
return live
|
|
|
|
def write_blocks(blocks):
|
|
try:
|
|
MARK.mkdir(exist_ok=True)
|
|
tmp = BLOCKS_FILE.with_suffix(".tmp"); tmp.write_text(json.dumps(blocks)); os.replace(tmp, BLOCKS_FILE)
|
|
except OSError as error:
|
|
log("blocks_write_failed", level="error", error=str(error))
|
|
|
|
def block_user(user, minutes, by, reason=""):
|
|
with _blocks_lock:
|
|
blocks = read_blocks()
|
|
blocks[user] = {"until": _time.time() + minutes * 60, "by": by, "reason": reason,
|
|
"since": datetime.now(TZ).isoformat(timespec="seconds")}
|
|
write_blocks(blocks)
|
|
log("publish_block", user=user, minutes=minutes, by=by)
|
|
|
|
def unblock_user(user, by):
|
|
with _blocks_lock:
|
|
blocks = read_blocks()
|
|
if blocks.pop(user, None) is not None:
|
|
write_blocks(blocks); log("publish_unblock", user=user, by=by); return True
|
|
return False
|
|
|
|
def blocked(user):
|
|
return read_blocks().get(user)
|
|
|
|
def kick_publisher(kind, session_id):
|
|
endpoint = KICK_ENDPOINT.get(kind)
|
|
if not endpoint:
|
|
raise ValueError(f"cannot kick a {kind} publisher")
|
|
request = urllib.request.Request(f"{MTX_API}/v3/{endpoint}/kick/{urllib.parse.quote(session_id, safe='')}", method="POST")
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return response.status
|
|
|
|
def stop_stream(path, kind, session_id, minutes, who):
|
|
user = path.split("/", 1)[0].lower()
|
|
out = {"path": path, "user": user, "kicked": False, "blocked_minutes": minutes}
|
|
if minutes > 0:
|
|
block_user(user, minutes, who, reason=f"stopped from the Live tab ({path})") # block first: no reconnect window
|
|
try:
|
|
kick_publisher(kind, session_id); out["kicked"] = True
|
|
except Exception as error:
|
|
out["error"] = f"kick failed: {error}"
|
|
# the Opus mirror of an OBS stream dies with its source (runOnReady exits), so nothing else to stop
|
|
_live_cache["data"] = None
|
|
log("stream_stop", level="info" if out["kicked"] else "error", stream=path, kind=kind, by=who, minutes=minutes,
|
|
result=out.get("error", "kicked"))
|
|
return out
|
|
|
|
def live_status():
|
|
now = _time.time()
|
|
if _live_cache["data"] is not None and now - _live_cache["at"] < 5:
|
|
return _live_cache["data"]
|
|
out = {"paths": [], "publishers": [], "counts": {}, "log": [], "errors": [], "history": []}
|
|
sessions = {}
|
|
try:
|
|
items = webrtc_sessions()
|
|
track_sessions(items)
|
|
for it in items:
|
|
sessions[it.get("id", "")] = {"transport": session_transport(it), "since": it.get("created"),
|
|
"established": it.get("peerConnectionEstablished")}
|
|
out["history"] = session_history()
|
|
except Exception as error:
|
|
out["errors"].append(f"MediaMTX WebRTC sessions: {error}")
|
|
try:
|
|
with urllib.request.urlopen(f"{MTX_API}/v3/paths/list?itemsPerPage=1000", timeout=3) as response:
|
|
for item in json.load(response).get("items") or []:
|
|
if item.get("ready"):
|
|
row = {"name": item["name"], "tracks": item.get("tracks") or [],
|
|
"readers": len(item.get("readers") or []),
|
|
"since": item.get("readyTime"), "bytes": item.get("bytesReceived", 0),
|
|
"mirror": str(item["name"]).endswith(".webrtc")}
|
|
if not row["mirror"]:
|
|
try:
|
|
row["health"] = stream_health(item["name"])
|
|
except Exception as error:
|
|
row["health_error"] = str(error)
|
|
out["paths"].append(row)
|
|
except Exception as error:
|
|
out["errors"].append(f"MediaMTX API: {error}")
|
|
try:
|
|
with urllib.request.urlopen(MTX_METRICS, timeout=3) as response:
|
|
text = response.read().decode(errors="replace")
|
|
rows = {}
|
|
for match in re.finditer(r"^(webrtc_sessions|rtmp_conns|srt_conns)(_rtp_packets_received|_rtp_packets_lost|_bytes_received)?\{([^}]*)\} (\d+)$", text, re.M):
|
|
kind, field, labels_text, value = match.groups()
|
|
labels = dict(re.findall(r'(\w+)="([^"]*)"', labels_text))
|
|
if labels.get("state") != "publish":
|
|
continue
|
|
key = (labels.get("path", "?"), labels.get("id", ""), labels.get("remoteAddr", "?"))
|
|
row = rows.setdefault(key, {"kind": kind.split("_")[0]})
|
|
if field == "_rtp_packets_received": row["received"] = int(value)
|
|
elif field == "_rtp_packets_lost": row["lost"] = int(value)
|
|
elif field == "_bytes_received": row["bytes"] = int(value)
|
|
for (path, sid, addr), row in rows.items():
|
|
out["publishers"].append({"path": path, "id": sid[:8], "session": sid, "from": addr, **row, **sessions.get(sid, {})})
|
|
out["blocks"] = [{"user": u, **b, "until_iso": datetime.fromtimestamp(b["until"], TZ).isoformat(timespec="seconds")}
|
|
for u, b in sorted(read_blocks().items())]
|
|
except Exception as error:
|
|
out["errors"].append(f"MediaMTX metrics: {error}")
|
|
try:
|
|
done = subprocess.run(["docker", "logs", "mediamtx", "--since", "10m"],
|
|
capture_output=True, text=True, timeout=15)
|
|
log_text = done.stdout + done.stderr
|
|
out["counts"] = {
|
|
"publish_sessions": log_text.count("is publishing"),
|
|
"recorder_restarts": len(re.findall(r"recording \d+ tracks", log_text)),
|
|
"processing_errors": log_text.count("processing errors"),
|
|
"access_unit_errors": log_text.count("recorder] access unit"),
|
|
"window": "10m",
|
|
}
|
|
out["log"] = [l for l in log_text.splitlines() if re.search(r"recorder|is publishing|closed|ERR", l)][-14:]
|
|
except Exception as error:
|
|
out["errors"].append(f"MediaMTX log: {error}")
|
|
_live_cache.update(at=now, data=out)
|
|
return out
|
|
|
|
# ---- disk report ------------------------------------------------------------------------------------
|
|
# GET /streaming/api/disk: where the bytes are. Streamers (streams/<user>, month from the recording's own
|
|
# "__YYYY-MM-DD" stamp, else mtime), clippers (clips/<user>, mtime), trash (its night folders), system (appdata
|
|
# per app, /srv/live, /srv/bot, Docker's own accounting) and logs (Elasticsearch indices by the date in their
|
|
# name, the ES data dir as the on-disk truth, journald, Docker's json logs). Walking a few TB of files takes
|
|
# seconds, so the report is cached for ten minutes and ?refresh=1 forces it. Sizes are allocated blocks, i.e.
|
|
# what the disk actually loses, not apparent length.
|
|
_disk_cache = {"at": 0.0, "data": None}
|
|
_disk_lock = threading.Lock()
|
|
DISK_CACHE_S = 600
|
|
APPDATA = Path("/srv/appdata")
|
|
ES_NETRC = Path(os.environ.get("ES_NETRC_FILE", "/etc/vanguard/es-netrc"))
|
|
ES_URL = os.environ.get("ES_URL", "http://127.0.0.1:9200")
|
|
MONTH_IN_NAME = re.compile(r"__(\d{4})-(\d{2})-\d{2}_")
|
|
MONTH_IN_INDEX = re.compile(r"(\d{4})\.(\d{2})\.\d{2}")
|
|
|
|
def _blocks(st):
|
|
return st.st_blocks * 512 if hasattr(st, "st_blocks") else st.st_size
|
|
|
|
def walk_sizes(root, month_of):
|
|
"""{month: bytes} and total for everything under root; month_of(path, stat) -> 'YYYY-MM'."""
|
|
months, total = {}, 0
|
|
stack = [root]
|
|
while stack:
|
|
current = stack.pop()
|
|
try:
|
|
with os.scandir(current) as entries:
|
|
for entry in entries:
|
|
try:
|
|
if entry.is_symlink(): continue
|
|
if entry.is_dir(follow_symlinks=False): stack.append(entry.path); continue
|
|
st = entry.stat(follow_symlinks=False)
|
|
except OSError:
|
|
continue
|
|
size = _blocks(st); total += size
|
|
month = month_of(entry.path, st)
|
|
months[month] = months.get(month, 0) + size
|
|
except OSError:
|
|
continue
|
|
return {"total": total, "months": months}
|
|
|
|
def month_from_mtime(_path, st):
|
|
return datetime.fromtimestamp(st.st_mtime, TZ).strftime("%Y-%m")
|
|
|
|
def month_from_name(path, st):
|
|
m = MONTH_IN_NAME.search(os.path.basename(path))
|
|
return f"{m.group(1)}-{m.group(2)}" if m else month_from_mtime(path, st)
|
|
|
|
def dir_total(path):
|
|
return walk_sizes(path, lambda p, s: "all")["total"] if Path(path).is_dir() else 0
|
|
|
|
def es_indices():
|
|
"""[{index, bytes, month}] from _cat/indices, with the netrc the deadman also uses."""
|
|
import netrc as _netrc
|
|
request = urllib.request.Request(f"{ES_URL}/_cat/indices?bytes=b&format=json&h=index,store.size,creation.date.string&expand_wildcards=all")
|
|
try:
|
|
creds = _netrc.netrc(str(ES_NETRC)).authenticators(urllib.parse.urlparse(ES_URL).hostname or "127.0.0.1")
|
|
except (OSError, _netrc.NetrcParseError):
|
|
creds = None
|
|
if creds:
|
|
request.add_header("Authorization", "Basic " + base64.b64encode(f"{creds[0]}:{creds[2]}".encode()).decode())
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
rows = json.load(response)
|
|
out = []
|
|
for row in rows:
|
|
name = row.get("index", "")
|
|
m = MONTH_IN_INDEX.search(name)
|
|
month = f"{m.group(1)}-{m.group(2)}" if m else (row.get("creation.date.string") or "")[:7] or "other"
|
|
out.append({"index": name, "bytes": int(row.get("store.size") or 0), "month": month})
|
|
return out
|
|
|
|
def docker_df():
|
|
done = subprocess.run(["docker", "system", "df", "--format", "{{json .}}"], capture_output=True, text=True, timeout=60)
|
|
rows = []
|
|
for line in done.stdout.splitlines():
|
|
try: rows.append(json.loads(line))
|
|
except ValueError: pass
|
|
return rows
|
|
|
|
def journal_bytes():
|
|
done = subprocess.run(["journalctl", "--disk-usage"], capture_output=True, text=True, timeout=20)
|
|
m = re.search(r"([\d.]+)\s*([KMGT]?)B?", done.stdout.replace(",", ""))
|
|
if not m: return 0
|
|
return int(float(m.group(1)) * {"": 1, "K": 1e3, "M": 1e6, "G": 1e9, "T": 1e12}[m.group(2)])
|
|
|
|
def disk_report(force=False):
|
|
now = _time.time()
|
|
with _disk_lock:
|
|
if not force and _disk_cache["data"] is not None and now - _disk_cache["at"] < DISK_CACHE_S:
|
|
return _disk_cache["data"]
|
|
t0 = _time.time()
|
|
out = {"generated": datetime.now(TZ).isoformat(timespec="seconds"), "fs": {}, "streamers": {}, "clippers": {},
|
|
"trash": {}, "system": {}, "logs": {}, "errors": []}
|
|
for label, path in (("media", ROOT), ("root", Path("/"))):
|
|
try:
|
|
u = shutil.disk_usage(path); out["fs"][label] = {"path": str(path), "total": u.total, "used": u.used, "free": u.free}
|
|
except OSError as error:
|
|
out["errors"].append(f"{path}: {error}")
|
|
same_fs = False
|
|
try: same_fs = os.stat(ROOT).st_dev == os.stat("/").st_dev
|
|
except OSError: pass
|
|
out["fs"]["same_filesystem"] = same_fs
|
|
for user_dir in sorted(p for p in (ROOT / "streams").glob("*") if p.is_dir()):
|
|
r = walk_sizes(user_dir, month_from_name)
|
|
r["raw"] = dir_total(user_dir / ".src"); r["unpublished"] = sum(dir_total(d) for d in user_dir.glob("*/.live"))
|
|
out["streamers"][user_dir.name] = r
|
|
for user_dir in sorted(p for p in (ROOT / "clips").glob("*") if p.is_dir()):
|
|
out["clippers"][user_dir.name] = walk_sizes(user_dir, month_from_mtime)
|
|
out["trash"] = walk_sizes(ROOT / "trash", lambda p, s: (re.search(r"/trash/(\d{4}-\d{2})-\d{2}/", p) or [None, month_from_mtime(p, s)])[1]) if (ROOT / "trash").is_dir() else {"total": 0, "months": {}}
|
|
other_media = 0
|
|
for entry in ROOT.iterdir():
|
|
if entry.name not in ("streams", "clips", "trash") and entry.is_dir() and not entry.is_symlink():
|
|
other_media += dir_total(entry)
|
|
out["system"]["media_other"] = other_media # .markers, .syncthing, .stversions, …
|
|
for app_dir in sorted(p for p in APPDATA.glob("*") if p.is_dir()) if APPDATA.is_dir() else []:
|
|
size = dir_total(app_dir)
|
|
if app_dir.name == "elasticsearch": out["logs"]["es_data_dir"] = size
|
|
else: out["system"][f"appdata/{app_dir.name}"] = size
|
|
for label, path in (("live portal", "/srv/live"), ("bot", "/srv/bot"), ("etc/vanguard", "/etc/vanguard")):
|
|
out["system"][label] = dir_total(path)
|
|
try:
|
|
for row in docker_df():
|
|
out["system"][f"docker {row.get('Type', '?').lower()}"] = row.get("Size", "?") # docker's own figures (human strings)
|
|
except Exception as error:
|
|
out["errors"].append(f"docker system df: {error}")
|
|
try:
|
|
out["logs"]["indices"] = es_indices()
|
|
out["logs"]["es_total"] = sum(i["bytes"] for i in out["logs"]["indices"])
|
|
by_month = {}
|
|
for i in out["logs"]["indices"]: by_month[i["month"]] = by_month.get(i["month"], 0) + i["bytes"]
|
|
out["logs"]["months"] = by_month
|
|
except Exception as error:
|
|
out["errors"].append(f"Elasticsearch: {error}")
|
|
try: out["logs"]["journald"] = journal_bytes()
|
|
except Exception as error: out["errors"].append(f"journalctl: {error}")
|
|
out["logs"]["docker_json_logs"] = sum(_blocks(p.stat()) for p in Path("/var/lib/docker/containers").glob("*/*-json.log*")) if Path("/var/lib/docker/containers").is_dir() else 0
|
|
out["seconds"] = round(_time.time() - t0, 1)
|
|
_disk_cache.update(at=_time.time(), data=out)
|
|
return out
|
|
|
|
# ---- stuck recordings -----------------------------------------------------------------------------
|
|
# GET /streaming/api/stuck every unpublished recording and why, for anyone the gate let through
|
|
# POST /streaming/api/stuck {"action":"retry"|"drop","src":"<path>"} — admins only
|
|
# The classification and the file moves both live in bin/vgstuck.sh, so the web page and the CLI can never
|
|
# disagree about why something is stuck, and "delete" means the same three-day trash in both.
|
|
STUCK_TOOL = os.environ.get("VGSTUCK_TOOL", "/usr/local/bin/vgstuck.sh")
|
|
|
|
_stuck_cache = {"at": 0.0, "items": None}
|
|
STUCK_CACHE_S = 20 # the page refreshes every 30 s and several admins may have it open
|
|
|
|
def stuck_list():
|
|
"""The survey, cached briefly. It runs ffprobe on every stuck file, so with many of them it is slow, and
|
|
without a cache every open tab re-ran it every 30 s. The cache is also invalidated by any action."""
|
|
now = _time.time()
|
|
if _stuck_cache["items"] is not None and now - _stuck_cache["at"] < STUCK_CACHE_S:
|
|
return _stuck_cache["items"]
|
|
try:
|
|
done = subprocess.run([STUCK_TOOL, "json"], capture_output=True, text=True, timeout=120)
|
|
except subprocess.TimeoutExpired:
|
|
log("stuck_list_timeout", level="error", seconds=120)
|
|
raise RuntimeError("the survey took over 120 s; too many files, or ffprobe is stuck") from None
|
|
if done.returncode != 0:
|
|
log("stuck_list_failed", level="error", rc=done.returncode, err=(done.stderr or "")[:300])
|
|
raise RuntimeError(f"vgstuck.sh exited {done.returncode}: {(done.stderr or '').strip()[-200:]}")
|
|
try:
|
|
items = json.loads(done.stdout or "[]")
|
|
except ValueError as error:
|
|
log("stuck_list_badjson", level="error", err=str(error), head=(done.stdout or "")[:120])
|
|
raise RuntimeError(f"vgstuck.sh did not return JSON: {error}") from None
|
|
_stuck_cache.update(at=now, items=items)
|
|
return items
|
|
|
|
def stuck_target(src):
|
|
"""Resolve what the browser asked us to act on, or raise. This is the only thing between a request body
|
|
and a file move, so it is deliberately strict: inside the streams tree, an unpublished .live recording,
|
|
and really there."""
|
|
path = Path(src).resolve()
|
|
root = (ROOT / "streams").resolve()
|
|
if root != path and root not in path.parents:
|
|
raise ValueError("that path is not inside the streams tree")
|
|
if f"{os.sep}.live{os.sep}" not in str(path) or path.suffix.lower() != ".mp4":
|
|
raise ValueError("that is not an unpublished recording")
|
|
if not path.is_file():
|
|
raise ValueError("no such recording (already published or already binned?)")
|
|
return path
|
|
|
|
def stuck_action(action, src, who):
|
|
path = stuck_target(src)
|
|
done = subprocess.run([STUCK_TOOL, action, str(path)], capture_output=True, text=True, timeout=600)
|
|
_stuck_cache["items"] = None # the list just changed; the next read must re-survey
|
|
lines = [l for l in (done.stdout + done.stderr).splitlines() if l.strip()]
|
|
log(f"stuck_{action}", level="info" if done.returncode == 0 else "error",
|
|
file=str(path), by=who, rc=done.returncode)
|
|
return {"ok": done.returncode == 0, "rc": done.returncode, "output": lines[-8:]}
|
|
|
|
# ---- GET /metrics -----------------------------------------------------------------------------------
|
|
# Prometheus text for Metricbeat's prometheus module. Two sources are merged:
|
|
# 1. counters the media scripts keep (bin/vgmetrics.sh writes MARK/metrics/<component>.counters)
|
|
# 2. gauges computed here at scrape time: unpublished recordings, encodes in flight, failures, MediaMTX
|
|
# Only the loopback listener serves it, and Caddy routes no host to /metrics (every dumpd route over there is
|
|
# an explicit path matcher), so this is reachable from the host and from nowhere else. Nothing here is
|
|
# authenticated and nothing here is allowed to fail: a scrape must never be able to disturb the stack.
|
|
METRICS_DIR = Path(os.environ.get("VG_METRICS_DIR", str(MARK / "metrics")))
|
|
MTX_API = os.environ.get("MTX_API_URL", "http://127.0.0.1:9997")
|
|
IDLE_S = int(os.environ.get("MINS_IDLE", "3")) * 60 # the sweep's own "finished being written" rule
|
|
|
|
# counters key head -> (metric, type, help, the labels the remaining key fields fill, in order)
|
|
COUNTER_MAP = {
|
|
"jobs": ("vanguard_media_jobs_total", "counter", "normalise/deband jobs that finished", ("verdict", "outcome", "kind")),
|
|
"seconds": ("vanguard_media_job_seconds_total", "counter", "wall-clock seconds spent in ffmpeg", ("verdict", "kind")),
|
|
"runs": ("vanguard_media_runs_total", "counter", "times the script has started work", ()),
|
|
"deband": ("vanguard_media_deband_total", "counter", "timecode bands cropped off", ()),
|
|
"last_finish": ("vanguard_media_last_finish_timestamp_seconds", "gauge", "when the script last finished", ()),
|
|
"last_duration": ("vanguard_media_last_run_seconds", "gauge", "how long the last finished run took", ()),
|
|
}
|
|
|
|
def _num(value):
|
|
"""Prometheus wants a plain number: %g would turn a unix timestamp into 1.76e+09."""
|
|
return "%d" % value if float(value).is_integer() else "%.3f" % value
|
|
|
|
def _glob(*patterns):
|
|
found = []
|
|
for pattern in patterns:
|
|
try:
|
|
found.extend(ROOT.glob(pattern))
|
|
except OSError:
|
|
pass
|
|
return found
|
|
|
|
def mtx_paths():
|
|
"""MediaMTX's own view of what is live. None if it did not answer."""
|
|
try:
|
|
with urllib.request.urlopen(f"{MTX_API}/v3/paths/list?itemsPerPage=1000", timeout=1.5) as response:
|
|
return json.load(response).get("items") or []
|
|
except Exception:
|
|
return None
|
|
|
|
def metrics_text():
|
|
order, grouped = [], {}
|
|
def add(metric, typ, helptext, labels, value):
|
|
if metric not in grouped:
|
|
grouped[metric] = (typ, helptext, [])
|
|
order.append(metric)
|
|
grouped[metric][2].append((labels, value))
|
|
|
|
# 1. whatever the scripts have counted
|
|
for counters in sorted(METRICS_DIR.glob("*.counters")):
|
|
component = counters.name[: -len(".counters")]
|
|
try:
|
|
lines = counters.read_text().splitlines()
|
|
except OSError:
|
|
continue
|
|
for line in lines:
|
|
field = line.split()
|
|
if len(field) != 2:
|
|
continue
|
|
try:
|
|
value = float(field[1])
|
|
except ValueError:
|
|
continue
|
|
head, *rest = field[0].split(":")
|
|
if head not in COUNTER_MAP:
|
|
continue
|
|
metric, typ, helptext, names = COUNTER_MAP[head]
|
|
labels = {"component": component}
|
|
labels.update(dict(zip(names, rest)))
|
|
add(metric, typ, helptext, labels, value)
|
|
|
|
# 2. is each script executing right now? (its pid file exists only while it runs; a crash leaves a stale
|
|
# one, so the pid is checked against /proc rather than trusted)
|
|
for pid_file in sorted(METRICS_DIR.glob("*.pid")):
|
|
try:
|
|
pid = int(pid_file.read_text().strip() or 0)
|
|
except (OSError, ValueError):
|
|
pid = 0
|
|
add("vanguard_media_running", "gauge", "1 while the script is executing",
|
|
{"component": pid_file.name[: -len(".pid")]}, 1 if pid and Path(f"/proc/{pid}").is_dir() else 0)
|
|
|
|
# 3. the queue. Depth-limited globs on purpose: never a walk of the whole media tree.
|
|
now = _time.time()
|
|
recording, pending, oldest = 0, 0, 0.0
|
|
for f in _glob("streams/*/*/.live/*"):
|
|
if f.name.startswith(".") or not f.is_file():
|
|
continue
|
|
try:
|
|
age = now - f.stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
# the sweep ignores anything touched in the last MINS_IDLE, so younger files are still being written
|
|
if age <= IDLE_S:
|
|
recording += 1
|
|
else:
|
|
pending += 1
|
|
oldest = max(oldest, age)
|
|
add("vanguard_media_recording", "gauge", "recordings MediaMTX is still writing into .live/", {}, recording)
|
|
add("vanguard_media_pending", "gauge", "finished recordings waiting to be normalised", {"kind": "stream"}, pending)
|
|
add("vanguard_media_pending_oldest_seconds", "gauge", "age of the oldest waiting recording", {}, round(oldest))
|
|
add("vanguard_media_encoding_in_flight", "gauge", "ffmpeg output files open right now", {},
|
|
len(_glob("streams/*/*.part", "streams/*/*/*.part", "streams/*/*/.live/*.part",
|
|
"clips/*/*.part", "clips/*/*/*.part", "clips/*/*/*/*.part")))
|
|
add("vanguard_media_failed_files", "gauge", "files parked with a normalize-failed log", {},
|
|
len(_glob("streams/*/*.normalize-failed.log", "streams/*/*/*.normalize-failed.log",
|
|
"clips/*/*.normalize-failed.log", "clips/*/*/*.normalize-failed.log",
|
|
"clips/*/*/*/*.normalize-failed.log")))
|
|
|
|
# 4. MediaMTX. The Prometheus scrape of :9998 already has per-path detail; this is the count the portal
|
|
# and the dock actually mean by "live", with the derived .webrtc audio mirrors separated out.
|
|
paths = mtx_paths()
|
|
try:
|
|
track_sessions(webrtc_sessions()) # keeps the transport history current between page views
|
|
except Exception:
|
|
pass
|
|
add("vanguard_mediamtx_up", "gauge", "did the MediaMTX API answer this scrape", {}, 0 if paths is None else 1)
|
|
if paths is not None:
|
|
ready = [p for p in paths if p.get("ready")]
|
|
add("vanguard_streams_active", "gauge", "streams publishing right now, mirrors excluded", {},
|
|
sum(1 for p in ready if not str(p.get("name", "")).endswith(".webrtc")))
|
|
add("vanguard_streams_mirrors", "gauge", "derived .webrtc audio mirrors running", {},
|
|
sum(1 for p in ready if str(p.get("name", "")).endswith(".webrtc")))
|
|
add("vanguard_stream_readers", "gauge", "viewers attached across all paths", {},
|
|
sum(len(p.get("readers") or []) for p in ready))
|
|
|
|
out = []
|
|
for metric in order:
|
|
typ, helptext, series = grouped[metric]
|
|
if helptext:
|
|
out.append(f"# HELP {metric} {helptext}")
|
|
out.append(f"# TYPE {metric} {typ}")
|
|
for labels, value in series:
|
|
tags = ",".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
|
|
out.append(f"{metric}{{{tags}}} {_num(value)}" if tags else f"{metric} {_num(value)}")
|
|
return "\n".join(out) + "\n"
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
server_version = "dumpd/1"
|
|
def _json(self, code, obj):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
def _text(self, code, body):
|
|
raw = body.encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
def _deny(self):
|
|
self._json(401, {"error": "not authenticated (no identity header from the proxy)"})
|
|
def _parse(self):
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
body = json.loads(self.rfile.read(n) or b"{}") if n else {}
|
|
# ?user=all&night=YYYY-MM-DD also work (the Dump tab's dry run). GET never moves anything: a link
|
|
# unfurled by Discord must not be able to dump a night.
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
|
|
for key in ("night", "user"):
|
|
if key not in body and query.get(key): body[key] = query[key][0]
|
|
night = body.get("night")
|
|
if night:
|
|
datetime.strptime(night, "%Y-%m-%d") # validate
|
|
return night, body.get("user")
|
|
def _resolve(self, night, tgt):
|
|
"""Explicit night wins. Otherwise: tonight, falling back to last night if tonight is empty."""
|
|
if night:
|
|
return night, candidates(tgt, night)
|
|
n = default_night()
|
|
files = candidates(tgt, n)
|
|
if not files:
|
|
prev = (datetime.strptime(n, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
|
|
pf = candidates(tgt, prev)
|
|
if pf:
|
|
return prev, pf
|
|
return n, files
|
|
def _is_dump(self):
|
|
path = self.path.split("?", 1)[0]
|
|
return path in ("/dump", "/streaming/api/dump") # live. dock button / diag. Dump tab
|
|
def _admin_door(self):
|
|
return self.path.split("?", 1)[0] == "/streaming/api/dump"
|
|
def _target(self, auth_user, is_admin, req_user):
|
|
# Dumping someone else's night is triage, so it is only offered where triage lives: the diag host's
|
|
# streaming page. The dock's Dump tonight (live. /dump) is personal for everyone, admins included.
|
|
if is_admin and self._admin_door():
|
|
return req_user or auth_user # admins default to their own night; pass "all" or a name for others
|
|
return auth_user
|
|
def do_GET(self):
|
|
if self.server.server_port == AUTH_LISTEN[1]:
|
|
return self._json(404, {"error": "not found"}) # the auth port serves POST /mtx-auth only
|
|
if self.path == "/metrics":
|
|
try:
|
|
return self._text(200, metrics_text())
|
|
except Exception as error: # a scrape must never take dumpd down
|
|
log("metrics_failed", level="error", error=str(error))
|
|
return self._text(500, f"# metrics unavailable: {error}\n")
|
|
if self.path.startswith("/time"):
|
|
# server clock for the timecode strip (the /clock/ page and Go Live measure their offset against this)
|
|
return self._json(200, {"now_ms": int(_time.time() * 1000)})
|
|
if self.path == "/me" or self.path == "/me/":
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
groups = sorted({g.strip() for g in re.split(r"[|,]", self.headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()})
|
|
return self._json(200, {"user": who, "groups": groups, "admin": is_admin,
|
|
"name": self.headers.get("X-Authentik-Name") or who, "email": self.headers.get("X-Authentik-Email") or "",
|
|
"discord_id": proxy_claim(self.headers, "discord_id")})
|
|
if self.path == "/me/stream-health":
|
|
# The dock polls this after Go Live. Only the caller's own paths, and only what is live now.
|
|
who, _ = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
streams = []
|
|
try:
|
|
with urllib.request.urlopen(f"{MTX_API}/v3/paths/list?itemsPerPage=1000", timeout=3) as response:
|
|
for item in json.load(response).get("items") or []:
|
|
name = str(item.get("name", ""))
|
|
if item.get("ready") and name.lower().startswith(who.lower() + "/") and not name.endswith(".webrtc"):
|
|
health = stream_health(name)
|
|
streams.append({"path": name, "tracks": item.get("tracks") or [], **(health or {}),
|
|
"webrtc_ok": bool(health) and health["bframes"] == 0})
|
|
except Exception as error:
|
|
return self._json(502, {"error": f"MediaMTX: {error}"})
|
|
# Go Live polls this while live: an empty list means the server no longer has the stream (kicked, or
|
|
# the session died), and "blocked" says why it must not simply reconnect.
|
|
hold = blocked(who.lower())
|
|
return self._json(200, {"user": who, "streams": streams,
|
|
"blocked": ({"until": datetime.fromtimestamp(hold["until"], TZ).isoformat(timespec="seconds"),
|
|
"by": hold.get("by")} if hold else None)})
|
|
if self.path.split("?", 1)[0] == "/streaming/api/disk":
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
force = "refresh=1" in (self.path.split("?", 1) + [""])[1]
|
|
try:
|
|
return self._json(200, {"admin": is_admin, "user": who, **disk_report(force)})
|
|
except Exception as error:
|
|
return self._json(500, {"error": str(error)})
|
|
if self.path == "/streaming/api/live":
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
try:
|
|
return self._json(200, {"admin": is_admin, "user": who, **live_status()})
|
|
except Exception as error:
|
|
return self._json(500, {"error": str(error)})
|
|
if self.path == "/streaming/api/stuck":
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
try:
|
|
return self._json(200, {"admin": is_admin, "user": who, "items": stuck_list()})
|
|
except Exception as error:
|
|
# Always answer. A handler that raises sends nothing, Caddy turns that into an empty 502,
|
|
# and the page can only say "unexpected end of JSON" — which is exactly what happened.
|
|
return self._json(500, {"error": str(error)})
|
|
if self.path.startswith("/me/clips"):
|
|
who, _ = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
return self._json(200, clips_status(who))
|
|
if self.path.startswith("/me/stream-token") or self.path.startswith("/me/stream-key"):
|
|
# Only reachable through Caddy behind Authentik forward-auth, which injects the identity headers.
|
|
who, _ = identity(self.headers)
|
|
if not who:
|
|
return self._json(401, {"error": "not authenticated"})
|
|
groups = {g.strip() for g in re.split(r"[|,]", self.headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()}
|
|
if not (groups & PUBLISH_GROUPS):
|
|
return self._json(403, {"error": f"'{who}' is not in a publishing group ({', '.join(sorted(PUBLISH_GROUPS))}) — ask an admin for the Discord role"})
|
|
try:
|
|
tok, exp = mint_token(who)
|
|
except RuntimeError as e:
|
|
return self._json(500, {"error": str(e)})
|
|
# "pass" kept for the dock/portal: the token IS the password for MediaMTX
|
|
return self._json(200, {"user": who, "pass": tok, "token": tok, "expires": exp, "ttl_hours": TOKEN_TTL_H})
|
|
if not self._is_dump():
|
|
return self._json(404, {"error": "not found"})
|
|
u, is_admin = identity(self.headers)
|
|
if not u:
|
|
return self._deny()
|
|
try:
|
|
night, req_user = self._parse()
|
|
except Exception as e:
|
|
return self._json(400, {"error": str(e)})
|
|
tgt = self._target(u, is_admin, req_user)
|
|
night, files = self._resolve(night, tgt)
|
|
start, end = night_window(night)
|
|
entries = [{"path": str(f.relative_to(ROOT)), "bytes": f.stat().st_size} for f in files]
|
|
self._json(200, {"night": night, "user": tgt, "by": u, "admin": is_admin, "dry_run": True,
|
|
"window": [start.isoformat(timespec="minutes"), end.isoformat(timespec="minutes")],
|
|
"default_night": default_night(),
|
|
"files": [e["path"] for e in entries], "entries": entries,
|
|
"gb": round(sum(e["bytes"] for e in entries) / 1e9, 2)})
|
|
def do_POST(self):
|
|
if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/heartbeat"):
|
|
# Alertmanager posts here every minute because the Watchdog alert always fires. Nothing is read from
|
|
# the body and nothing is authenticated: the port only accepts the Docker bridge, and the only thing
|
|
# this grants is keeping a timestamp fresh. The deadman treats a STALE file as the alarm.
|
|
try:
|
|
self.rfile.read(int(self.headers.get("Content-Length") or 0))
|
|
HEARTBEAT_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
HEARTBEAT_FILE.write_text(f"{int(_time.time())}\n")
|
|
except OSError as error:
|
|
return self._json(500, {"error": str(error)})
|
|
return self._json(200, {"ok": True})
|
|
if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/deploy-bot"):
|
|
# The bot's CI posts its release bundle here (bridge-only port; the firewall allows docker0/br-* alone).
|
|
# The body is the artifact; the signature proves it came from a runner holding BOT_DEPLOY_SECRET.
|
|
env = deploy_env()
|
|
if env.get("BOT_DEPLOY_AUTO", "1") != "1":
|
|
log("bot_deploy_refused", level="warn", reason="BOT_DEPLOY_AUTO=0")
|
|
return self._json(403, {"error": "automatic bot deploys are turned off (BOT_DEPLOY_AUTO=0)"})
|
|
secret = env.get("BOT_DEPLOY_SECRET", "")
|
|
size = int(self.headers.get("Content-Length") or 0)
|
|
if not secret:
|
|
return self._json(503, {"error": "BOT_DEPLOY_SECRET is not set on the server"})
|
|
if not 0 < size <= BOT_DEPLOY_MAX_BYTES:
|
|
return self._json(413, {"error": f"body must be 1..{BOT_DEPLOY_MAX_BYTES} bytes, got {size}"})
|
|
raw = self.rfile.read(size)
|
|
env_b64 = (self.headers.get("X-Vanguard-Env") or "").strip()
|
|
if len(env_b64) > BOT_ENV_MAX_HEADER:
|
|
return self._json(413, {"error": "X-Vanguard-Env is too large"})
|
|
sig = (self.headers.get("X-Vanguard-Signature") or "").replace("sha256=", "").strip()
|
|
signed = env_b64.encode() + raw # the settings are signed with the bundle, not beside it
|
|
if not sig or not hmac.compare_digest(hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest(), sig):
|
|
log("bot_deploy_bad_signature", level="warn", ip=self.client_address[0], bytes=len(raw))
|
|
return self._json(401, {"error": "bad signature"})
|
|
applied, refused = [], []
|
|
if env_b64:
|
|
try:
|
|
values = json.loads(base64.b64decode(env_b64, validate=True))
|
|
if not isinstance(values, dict): raise ValueError("X-Vanguard-Env must be a JSON object")
|
|
applied, refused = apply_bot_env(values)
|
|
except (ValueError, OSError, TypeError) as error:
|
|
log("bot_env_failed", level="error", error=error)
|
|
return self._json(400, {"error": f"settings rejected: {error}"})
|
|
code, body = install_bot_bundle(raw, self.headers.get("X-Vanguard-Reason") or "ci")
|
|
body["settings_applied"] = applied
|
|
if refused: body["settings_refused"] = refused
|
|
return self._json(code, body)
|
|
if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/deploy"):
|
|
# Gitea (container) -> host gateway -> here. Verify HMAC-SHA256 of the raw body with the webhook secret.
|
|
n = int(self.headers.get("Content-Length") or 0); raw = self.rfile.read(n)
|
|
env = deploy_env(); secret = env.get("DEPLOY_WEBHOOK_SECRET", "")
|
|
sig = self.headers.get("X-Gitea-Signature") or self.headers.get("X-Hub-Signature-256", "").replace("sha256=", "")
|
|
if not secret or not sig or not hmac.compare_digest(hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest(), sig):
|
|
log("deploy_webhook_bad_signature", level="warn", ip=self.client_address[0])
|
|
return self._json(401, {"error": "bad signature"})
|
|
try: payload = json.loads(raw or b"{}")
|
|
except json.JSONDecodeError: return self._json(400, {"error": "bad json"})
|
|
ref = payload.get("ref", ""); branch = env.get("DEPLOY_BRANCH", "main")
|
|
after = (payload.get("after") or "")[:10]; pusher = (payload.get("pusher") or {}).get("username", "?")
|
|
if ref != f"refs/heads/{branch}":
|
|
return self._json(200, {"ignored": f"ref {ref} is not {branch}"})
|
|
if env.get("DEPLOY_AUTO", "0") != "1":
|
|
log("deploy_webhook_push", rev=after, pusher=pusher, branch=branch, auto="0")
|
|
return self._json(202, {"queued": False, "note": "auto-deploy disabled (DEPLOY_AUTO=0); run vanguard-deploy pull by hand"})
|
|
started = run_deploy(f"push {after} by {pusher}")
|
|
return self._json(202, {"queued": started, "note": "deploy started" if started else "a deploy is already running"})
|
|
if self.server.server_port == AUTH_LISTEN[1] and not self.path.startswith("/mtx-auth"):
|
|
return self._json(404, {"error": "not found"})
|
|
if self.path.startswith("/mtx-auth"):
|
|
# MediaMTX authHTTP: 200 = allow, anything else = deny. Body: user, password, token, ip, action, path, protocol, id, query
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
try:
|
|
req = json.loads(self.rfile.read(n) or b"{}")
|
|
except json.JSONDecodeError:
|
|
return self._json(400, {"error": "bad json"})
|
|
action = req.get("action", "")
|
|
if action in ("read", "playback"):
|
|
return self._json(200, {"ok": True}) # viewers are gated by Caddy + Authentik
|
|
if action != "publish":
|
|
return self._json(401, {"error": f"action {action} not allowed"})
|
|
path = (req.get("path") or "").lower()
|
|
if path.endswith(".webrtc"):
|
|
# audio mirrors (<user>/<Title>.webrtc) are published by the hook inside the mediamtx container only
|
|
if req.get("ip") in ("127.0.0.1", "::1") and re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_-]+\.webrtc", path):
|
|
return self._json(200, {"ok": True})
|
|
log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="mirror paths are internal")
|
|
return self._json(401, {"error": ".webrtc paths are reserved"})
|
|
cred = req.get("password") or req.get("token") or ""
|
|
if cred.lower().startswith("bearer "): cred = cred[7:]
|
|
if ":" in cred and "." in cred.split(":", 1)[1]: cred = cred.split(":", 1)[1] # "user:token" form
|
|
user = verify_token(cred)
|
|
if not user:
|
|
log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="bad or expired token")
|
|
return self._json(401, {"error": "invalid or expired stream token"})
|
|
# title segment is dot-free (mirrors are <title>.webrtc and must never collide with a title)
|
|
if not re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_-]+", path) or not path.startswith(user + "/"):
|
|
log("publish_deny", level="warn", user=user, stream=path, reason="not their prefix")
|
|
return self._json(401, {"error": "path must be <your-username>/<Title>"})
|
|
hold = blocked(user.lower())
|
|
if hold: # stopped from the Live tab: no reconnecting around it
|
|
log("publish_deny", level="warn", user=user, stream=path, ip=req.get("ip"), reason="blocked", by=hold.get("by"))
|
|
return self._json(401, {"error": "publishing is paused for this account by an admin"})
|
|
log("publish_allow", user=user, stream=path, ip=req.get("ip"))
|
|
return self._json(200, {"ok": True})
|
|
if self.path == "/streaming/api/live":
|
|
# {"action":"stop","path":..,"kind":..,"session":..,"minutes":N} or {"action":"unblock","user":..} — admins only
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._deny()
|
|
if not is_admin:
|
|
return self._json(403, {"error": f"stopping streams is for {', '.join(sorted(ADMIN_GROUPS))}"})
|
|
try:
|
|
body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
|
|
except ValueError as error:
|
|
return self._json(400, {"error": str(error)})
|
|
action = body.get("action")
|
|
if action == "unblock":
|
|
user = str(body.get("user", "")).lower()
|
|
if not re.fullmatch(r"[a-z0-9_.-]+", user): return self._json(400, {"error": "bad user"})
|
|
_live_cache["data"] = None
|
|
return self._json(200, {"ok": unblock_user(user, who), "user": user})
|
|
if action == "stop":
|
|
path, kind, session = str(body.get("path", "")), str(body.get("kind", "")), str(body.get("session", ""))
|
|
if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", path) or kind not in KICK_ENDPOINT or not re.fullmatch(r"[A-Za-z0-9-]{8,64}", session):
|
|
return self._json(400, {"error": "need path, kind and session as the live list reports them"})
|
|
try:
|
|
minutes = max(0, min(int(body.get("minutes", 30)), 24 * 60))
|
|
except (TypeError, ValueError):
|
|
minutes = 30
|
|
return self._json(200, stop_stream(path, kind, session, minutes, who))
|
|
return self._json(400, {"error": "action must be stop or unblock"})
|
|
if self.path == "/streaming/api/stuck":
|
|
who, is_admin = identity(self.headers)
|
|
if not who: return self._deny()
|
|
if not is_admin:
|
|
return self._json(403, {"error": f"retrying and binning recordings is for {', '.join(sorted(ADMIN_GROUPS))}"})
|
|
try:
|
|
body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
|
|
except ValueError as error:
|
|
return self._json(400, {"error": str(error)})
|
|
action = body.get("action")
|
|
if action == "wipe":
|
|
# Everything unprocessed, deliberately (the Dump tab's button): stops the sweep, then every .live/
|
|
# file except one MediaMTX still has open, plus stale failure logs and .part leftovers. vgstuck.sh
|
|
# does the moving, into trash.
|
|
done = subprocess.run([STUCK_TOOL, "wipe"], capture_output=True, text=True, timeout=900)
|
|
_stuck_cache["items"] = None
|
|
lines = [l for l in (done.stdout + done.stderr).splitlines() if l.strip()]
|
|
log("stuck_wipe", level="info" if done.returncode == 0 else "error", by=who, rc=done.returncode,
|
|
result=(lines[-1] if lines else ""))
|
|
return self._json(200, {"ok": done.returncode == 0, "rc": done.returncode, "output": lines[-6:]})
|
|
if action == "drop_under":
|
|
# Bulk: every unpublished STREAM recording smaller than N MB — reconnect stubs. Bounded so a
|
|
# typo cannot bin real recordings; clips never appear here because they never pass through
|
|
# .live/. Still-recording files are skipped by the tool itself.
|
|
try:
|
|
mb = int(body.get("mb", 0))
|
|
except (TypeError, ValueError):
|
|
mb = 0
|
|
if not 1 <= mb <= 50:
|
|
return self._json(400, {"error": "mb must be between 1 and 50"})
|
|
done = subprocess.run([STUCK_TOOL, "drop", "--under", str(mb)], capture_output=True, text=True, timeout=900)
|
|
_stuck_cache["items"] = None
|
|
lines = [l for l in (done.stdout + done.stderr).splitlines() if l.strip()]
|
|
log("stuck_drop_under", level="info" if done.returncode == 0 else "error", mb=mb, by=who, rc=done.returncode)
|
|
return self._json(200, {"ok": done.returncode == 0, "rc": done.returncode, "output": lines[-6:]})
|
|
if action not in ("retry", "drop"):
|
|
return self._json(400, {"error": "action must be retry, drop, drop_under or wipe"})
|
|
try:
|
|
return self._json(200, stuck_action(action, str(body.get("src", "")), who))
|
|
except ValueError as error:
|
|
return self._json(400, {"error": str(error)})
|
|
except Exception as error:
|
|
return self._json(500, {"error": str(error)})
|
|
if self.path.startswith("/me/clips"):
|
|
who, _ = identity(self.headers)
|
|
if not who: return self._json(401, {"error": "not authenticated"})
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
try:
|
|
body = json.loads(self.rfile.read(n) or b"{}")
|
|
return self._json(200, clips_setup(who, str(body.get("device_id", "")), body.get("device_name")))
|
|
except ValueError as e:
|
|
return self._json(400, {"error": str(e)})
|
|
except Exception as e:
|
|
return self._json(502, {"error": f"syncthing: {e}"})
|
|
if not self._is_dump():
|
|
return self._json(404, {"error": "not found"})
|
|
u, is_admin = identity(self.headers)
|
|
if not u:
|
|
return self._deny()
|
|
try:
|
|
night, req_user = self._parse()
|
|
except Exception as e:
|
|
return self._json(400, {"error": str(e)})
|
|
tgt = self._target(u, is_admin, req_user)
|
|
night, files = self._resolve(night, tgt)
|
|
moved, size = do_move(files, night)
|
|
MARK.mkdir(exist_ok=True)
|
|
(MARK / f"{night}-{tgt}.dumped").write_text(f"{datetime.now(TZ).isoformat()} by {u}\n")
|
|
log("dump", night=night, user=tgt, by=u, files=moved, gb=f"{size/1e9:.2f}")
|
|
self._json(200, {"night": night, "user": tgt, "moved": moved, "gb": round(size / 1e9, 2),
|
|
"undo_until": (datetime.now(TZ) + timedelta(days=3)).strftime("%F")})
|
|
def log_message(self, *a): # quiet
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
TRASH.mkdir(exist_ok=True); MARK.mkdir(exist_ok=True)
|
|
print(f"dumpd listening on {LISTEN[0]}:{LISTEN[1]} (dump, me) and {AUTH_LISTEN[0]}:{AUTH_LISTEN[1]} (mtx-auth) tz={TZ.key} night={NIGHT_START:%H:%M}-{NIGHT_END:%H:%M} token-secret={'ok' if token_secret() else 'MISSING'}", flush=True)
|
|
auth_srv = ThreadingHTTPServer(AUTH_LISTEN, H)
|
|
threading.Thread(target=auth_srv.serve_forever, daemon=True).start()
|
|
threading.Thread(target=quota_loop, daemon=True).start() # per-user clip quotas (pause/resume Syncthing folders)
|
|
ThreadingHTTPServer(LISTEN, H).serve_forever()
|