482 lines
26 KiB
Python
Executable File
482 lines
26 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 /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, sys
|
|
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:
|
|
print(f"{datetime.now(TZ):%F %T} quota: syncthing unreachable ({e})", flush=True); 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: print(f"{datetime.now(TZ):%F %T} quota: pause {user} failed: {e}", flush=True); continue
|
|
print(f"{datetime.now(TZ):%F %T} quota: PAUSED clips-{user} ({used/1e9:.1f} GB > {q/1e9:.0f} GB)", flush=True)
|
|
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: print(f"{datetime.now(TZ):%F %T} quota: resume {user} failed: {e}", flush=True); continue
|
|
print(f"{datetime.now(TZ):%F %T} quota: resumed clips-{user} ({used/1e9:.1f} GB)", flush=True)
|
|
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: print(f"{datetime.now(TZ):%F %T} quota loop error: {e}", flush=True)
|
|
_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)
|
|
print(f"{datetime.now(TZ):%F %T} clips setup user={user} device={device_id[:7]}… folder={fid}", flush=True)
|
|
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
|
|
|
|
_deploy_lock = {"running": False}
|
|
def run_deploy(reason):
|
|
import subprocess, threading
|
|
if _deploy_lock["running"]: return False
|
|
_deploy_lock["running"] = True
|
|
def go():
|
|
try:
|
|
print(f"{datetime.now(TZ):%F %T} deploy: start ({reason})", flush=True)
|
|
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)
|
|
print(f"{datetime.now(TZ):%F %T} deploy: exit {r.returncode}", flush=True)
|
|
except Exception as e:
|
|
print(f"{datetime.now(TZ):%F %T} deploy: error {e}", flush=True)
|
|
finally:
|
|
_deploy_lock["running"] = False
|
|
threading.Thread(target=go, daemon=True).start()
|
|
return 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"):
|
|
# 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):
|
|
print(f"{datetime.now(TZ):%F %T} deploy webhook: bad signature from {self.client_address[0]}", flush=True)
|
|
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":
|
|
print(f"{datetime.now(TZ):%F %T} deploy webhook: push {after} by {pusher} on {branch} — auto-deploy disabled (DEPLOY_AUTO=0)", flush=True)
|
|
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})
|
|
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish path={path} ip={req.get('ip')} (mirror paths are internal)", flush=True)
|
|
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:
|
|
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish path={path} ip={req.get('ip')} (bad/expired token)", flush=True)
|
|
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 + "/"):
|
|
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish user={user} path={path} (not their prefix)", flush=True)
|
|
return self._json(401, {"error": "path must be <your-username>/<Title>"})
|
|
print(f"{datetime.now(TZ):%F %T} mtx-auth allow publish user={user} path={path} ip={req.get('ip')}", flush=True)
|
|
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")
|
|
print(f"{datetime.now(TZ):%F %T} dump night={night} user={tgt} by={u} files={moved} gb={size/1e9:.2f}", flush=True)
|
|
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)
|
|
import threading
|
|
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()
|