Brandyn / Techy
1e4c0e88dc
POST /deploy-bot may now carry X-Vanguard-Env, base64 JSON of the bot's own settings, and the HMAC covers that header together with the body so the pair cannot be separated or swapped. dumpd merges only the keys BOT_ENV_KEYS allows (DISCORD_BOT_TOKEN, DISCORD_GUILD_ID by default) into the stack .env, atomically and preserving mode and owner, and reports the rest to the CI job as settings_refused. The server keeps what it generates: a CI job cannot touch the database password or the Authentik keys. Verified: allowed keys applied, server-owned keys refused while the rest still applied, a settings swap after signing rejected with 401, and an empty settings header leaving .env untouched. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
599 lines
32 KiB
Python
Executable File
599 lines
32 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)
|
|
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 /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.
|
|
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".
|
|
"""
|
|
import base64, hashlib, hmac, json, os, re, shutil, subprocess, sys, threading
|
|
import time as _time
|
|
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, /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
|
|
|
|
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
|
|
|
|
_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()
|
|
|
|
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 _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 {}
|
|
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 _target(self, auth_user, is_admin, req_user):
|
|
if is_admin:
|
|
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.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 ""})
|
|
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.path.startswith("/dump"):
|
|
return self._json(404, {"error": "not found"})
|
|
u, is_admin = identity(self.headers)
|
|
if not u:
|
|
return self._deny()
|
|
night, req_user = self._parse()
|
|
tgt = self._target(u, is_admin, req_user)
|
|
night, files = self._resolve(night, tgt)
|
|
self._json(200, {"night": night, "user": tgt, "dry_run": True,
|
|
"files": [str(f.relative_to(ROOT)) for f in files],
|
|
"gb": round(sum(f.stat().st_size for f in files) / 1e9, 2)})
|
|
def do_POST(self):
|
|
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"})
|
|
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>"})
|
|
log("publish_allow", user=user, stream=path, ip=req.get("ip"))
|
|
return self._json(200, {"ok": True})
|
|
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.path.startswith("/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()
|