feat(tower): Merlin admin API + basic auth
This commit is contained in:
Clément Saillant
+177
-66
@@ -1,78 +1,189 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Always-on: POST /register {id, have} -> push the staging delta to the box's
|
||||
file-server (:8080). The box IP is the request source address."""
|
||||
import json, os, threading, urllib.request
|
||||
"""Always-on Lisael Box content service on Tower:
|
||||
- POST /register {id,have} (machine: box announces, we push the delta; LAN/tailnet only)
|
||||
- Merlin Studio web UI (GET /) + /api/* (human, behind Basic Auth)
|
||||
"""
|
||||
import base64, ipaddress, json, os, threading, urllib.parse
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
import lisael_content as C
|
||||
|
||||
STAGING = os.environ.get("LISAEL_STAGING", "/home/clems/lisael-content/podcasts")
|
||||
PORT = int(os.environ.get("LISAEL_PORT", "9555"))
|
||||
BOX_PORT = int(os.environ.get("LISAEL_BOX_PORT", "8080"))
|
||||
PORT = int(os.environ.get("LISAEL_PORT", "9555"))
|
||||
UI_HTML = os.path.join(os.path.dirname(os.path.abspath(__file__)), "merlin_ui.html")
|
||||
AUTH = os.environ.get("MERLIN_AUTH", "") # "user:pass"; empty disables auth
|
||||
|
||||
def staging_files():
|
||||
if not os.path.isdir(STAGING):
|
||||
return []
|
||||
return sorted(f for f in os.listdir(STAGING) if not f.startswith("."))
|
||||
|
||||
def compute_delta(staging, have):
|
||||
"""Files the box is missing; manifest.json only when something else changed."""
|
||||
haveset = set(have)
|
||||
files = [f for f in staging if f != "manifest.json" and f not in haveset]
|
||||
if files and "manifest.json" in staging:
|
||||
files.append("manifest.json")
|
||||
return files
|
||||
|
||||
def push_file(ip, name):
|
||||
data = open(os.path.join(STAGING, name), "rb").read()
|
||||
url = f"http://{ip}:{BOX_PORT}/put?name=podcasts/{name}"
|
||||
# Generous timeout: a multi-MB episode written to the box's slow microSD can
|
||||
# take well over a minute, especially during boot when the box is also doing
|
||||
# its first weather TLS fetch.
|
||||
with urllib.request.urlopen(urllib.request.Request(url, data=data, method="POST"),
|
||||
timeout=300) as r:
|
||||
return r.status
|
||||
|
||||
# Pushes run off the request thread (one batch per box at a time) so the box's
|
||||
# short register timeout never waits on a multi-file, multi-minute transfer.
|
||||
_busy = set()
|
||||
_busy_lock = threading.Lock()
|
||||
|
||||
def push_batch(ip, delta):
|
||||
ok = 0
|
||||
for f in delta:
|
||||
try:
|
||||
push_file(ip, f); ok += 1
|
||||
except Exception as e:
|
||||
print("push fail", f, e, flush=True)
|
||||
print(f"push done {ip} {ok}/{len(delta)}", flush=True)
|
||||
with _busy_lock:
|
||||
_busy.discard(ip)
|
||||
def _is_private(ip):
|
||||
try:
|
||||
a = ipaddress.ip_address(ip)
|
||||
return a.is_private or a.is_loopback or ip.startswith("100.") # 100.x = tailnet
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
class H(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
if self.path.split("?")[0] != "/register":
|
||||
self.send_error(404); return
|
||||
ip = self.client_address[0]
|
||||
try:
|
||||
n = int(self.headers.get("Content-Length", 0))
|
||||
body = json.loads(self.rfile.read(n) or b"{}")
|
||||
except Exception:
|
||||
self.send_error(400); return
|
||||
delta = compute_delta(staging_files(), body.get("have", []) or [])
|
||||
print(f"register {ip} id={body.get('id')} have={len(body.get('have',[]) or [])} push={len(delta)}",
|
||||
flush=True)
|
||||
with _busy_lock:
|
||||
already = ip in _busy
|
||||
if delta and not already:
|
||||
_busy.add(ip)
|
||||
if delta and not already:
|
||||
threading.Thread(target=push_batch, args=(ip, delta), daemon=True).start()
|
||||
# Respond immediately; the (possibly slow) push runs in the background.
|
||||
self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
|
||||
self.wfile.write(json.dumps({"queued": len(delta), "busy": already}).encode())
|
||||
# --- helpers ---
|
||||
def _json(self, obj, code=200):
|
||||
b = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||||
self.send_response(code); self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(b))); self.end_headers()
|
||||
self.wfile.write(b)
|
||||
|
||||
def _auth_ok(self):
|
||||
if not AUTH:
|
||||
return True
|
||||
h = self.headers.get("Authorization", "")
|
||||
if h.startswith("Basic "):
|
||||
try:
|
||||
if base64.b64decode(h[6:]).decode() == AUTH:
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
self.send_response(401)
|
||||
self.send_header("WWW-Authenticate", 'Basic realm="Merlin Studio"')
|
||||
self.end_headers()
|
||||
return False
|
||||
|
||||
def _body(self):
|
||||
n = int(self.headers.get("Content-Length", 0) or 0)
|
||||
return self.rfile.read(n) if n else b""
|
||||
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
# --- GET ---
|
||||
def do_GET(self):
|
||||
path = self.path.split("?")[0]
|
||||
if path == "/register":
|
||||
self.send_error(405); return
|
||||
if not self._auth_ok():
|
||||
return
|
||||
if path == "/" or path == "/index.html":
|
||||
try:
|
||||
data = open(UI_HTML, "rb").read()
|
||||
except OSError:
|
||||
self.send_error(500, "UI missing"); return
|
||||
self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(data))); self.end_headers()
|
||||
self.wfile.write(data); return
|
||||
if path == "/api/feeds":
|
||||
self._json(C.load_feeds()); return
|
||||
if path == "/api/episodes":
|
||||
q = urllib.parse.parse_qs(self.path.split("?", 1)[1] if "?" in self.path else "")
|
||||
key = (q.get("feed") or [""])[0]
|
||||
feed = next((f for f in C.load_feeds() if f["key"] == key), None)
|
||||
if not feed:
|
||||
self._json({"error": "unknown feed"}, 404); return
|
||||
try:
|
||||
eps, cov = C.fetch_episodes(feed["rss"])
|
||||
except Exception as e:
|
||||
self._json({"error": str(e)}, 502); return
|
||||
staged = set(C.staging_files())
|
||||
for e in eps:
|
||||
e["staged"] = C.episode_fname(key, e["url"]) in staged
|
||||
self._json({"feed": feed, "cover": cov, "episodes": eps}); return
|
||||
if path == "/api/files":
|
||||
out = []
|
||||
for f in C.staging_files():
|
||||
p = os.path.join(C.STAGING, f)
|
||||
out.append({"name": f, "size": os.path.getsize(p),
|
||||
"type": f.rsplit(".", 1)[-1] if "." in f else ""})
|
||||
self._json(out); return
|
||||
if path == "/api/box":
|
||||
st = C.box_state_load()
|
||||
delta = C.compute_delta(C.staging_files(), st.get("have", []))
|
||||
self._json({"box": st, "delta": delta, "staging_count": len(C.staging_files())}); return
|
||||
self.send_error(404)
|
||||
|
||||
# --- POST ---
|
||||
def do_POST(self):
|
||||
path = self.path.split("?")[0]
|
||||
if path == "/register":
|
||||
ip = self.client_address[0]
|
||||
if not _is_private(ip): # block tunnel-originated abuse
|
||||
self.send_error(403); return
|
||||
try:
|
||||
body = json.loads(self._body() or b"{}")
|
||||
except Exception:
|
||||
self.send_error(400); return
|
||||
have = body.get("have", []) or []
|
||||
C.box_state_save({"ip": ip, "id": body.get("id"),
|
||||
"last_seen": self.date_time_string(), "have": have})
|
||||
delta = C.compute_delta(C.staging_files(), have)
|
||||
print(f"register {ip} id={body.get('id')} have={len(have)} push={len(delta)}", flush=True)
|
||||
started = C.try_push(ip, delta)
|
||||
self._json({"queued": len(delta), "started": started}); return
|
||||
if not self._auth_ok():
|
||||
return
|
||||
if path == "/api/feeds":
|
||||
try:
|
||||
b = json.loads(self._body() or b"{}")
|
||||
C.add_feed(b["key"], b["name"], b["rss"], b.get("n", 2))
|
||||
except Exception as e:
|
||||
self._json({"error": str(e)}, 400); return
|
||||
self._json(C.load_feeds()); return
|
||||
if path == "/api/select":
|
||||
try:
|
||||
b = json.loads(self._body() or b"{}")
|
||||
key = b["feed"]; episodes = b["episodes"]
|
||||
except Exception as e:
|
||||
self._json({"error": str(e)}, 400); return
|
||||
feed = next((f for f in C.load_feeds() if f["key"] == key), None)
|
||||
if not feed:
|
||||
self._json({"error": "unknown feed"}, 404); return
|
||||
def work():
|
||||
cover_bytes = None
|
||||
try:
|
||||
_, cov = C.fetch_episodes(feed["rss"], limit=1)
|
||||
if cov:
|
||||
cover_bytes = C.fetch(cov)
|
||||
except Exception:
|
||||
cover_bytes = None
|
||||
for ep in episodes:
|
||||
try:
|
||||
C.add_episode(key, feed["name"], cover_bytes, ep["url"], ep.get("title", ""))
|
||||
cover_bytes = None
|
||||
print("selected", C.episode_fname(key, ep["url"]), flush=True)
|
||||
except Exception as e:
|
||||
print("select fail", e, flush=True)
|
||||
threading.Thread(target=work, daemon=True).start()
|
||||
self._json({"queued": len(episodes)}, 202); return
|
||||
if path == "/api/upload":
|
||||
q = urllib.parse.parse_qs(self.path.split("?", 1)[1] if "?" in self.path else "")
|
||||
show = (q.get("show") or ["Perso"])[0]
|
||||
title = (q.get("title") or ["Audio"])[0]
|
||||
data = self._body()
|
||||
if not data:
|
||||
self._json({"error": "empty"}, 400); return
|
||||
try:
|
||||
fname = C.add_upload(show, data, title)
|
||||
except Exception as e:
|
||||
self._json({"error": str(e)}, 500); return
|
||||
self._json({"file": fname}); return
|
||||
if path == "/api/push":
|
||||
st = C.box_state_load()
|
||||
ip = st.get("ip")
|
||||
if not ip:
|
||||
self._json({"error": "box never registered"}, 409); return
|
||||
delta = C.compute_delta(C.staging_files(), st.get("have", []))
|
||||
started = C.try_push(ip, delta)
|
||||
self._json({"ip": ip, "delta": len(delta), "started": started}); return
|
||||
self.send_error(404)
|
||||
|
||||
# --- DELETE ---
|
||||
def do_DELETE(self):
|
||||
if not self._auth_ok():
|
||||
return
|
||||
path = self.path.split("?")[0]
|
||||
if path.startswith("/api/feeds/"):
|
||||
C.remove_feed(urllib.parse.unquote(path[len("/api/feeds/"):]))
|
||||
self._json({"ok": True}); return
|
||||
if path.startswith("/api/files/"):
|
||||
name = urllib.parse.unquote(path[len("/api/files/"):])
|
||||
try:
|
||||
C.remove_file(name)
|
||||
except ValueError:
|
||||
self.send_error(400); return
|
||||
self._json({"ok": True}); return
|
||||
self.send_error(404)
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(f"lisael-content on :{PORT} staging={STAGING}", flush=True)
|
||||
print(f"lisael-content on :{PORT} staging={C.STAGING} auth={'on' if AUTH else 'OFF'}", flush=True)
|
||||
ThreadingHTTPServer(("0.0.0.0", PORT), H).serve_forever()
|
||||
|
||||
Reference in New Issue
Block a user