From 0eb3d1eb051033ad21462763ee4778ad53de3d6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Saillant?= Date: Mon, 15 Jun 2026 08:02:50 +0200 Subject: [PATCH] feat(tower): Merlin admin API + basic auth --- tower/lisael_server.py | 243 ++++++++++++++++++++++++++++++----------- 1 file changed, 177 insertions(+), 66 deletions(-) diff --git a/tower/lisael_server.py b/tower/lisael_server.py index 422daf6..705691c 100644 --- a/tower/lisael_server.py +++ b/tower/lisael_server.py @@ -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()