#!/usr/bin/env python3 """ LableBrain static file server + per-artist annotations API. Public reads, gated writes: - GET / -> tries /.html (clean URLs) - GET / -> static file (public, no auth) - GET /annotations/ -> annotations list for artist (public) - POST /annotations/ -> requires Basic Auth : Environment variables: PORT TCP port to listen on (default 8080) RUBY_PASSWORD password required to POST /annotations/ruby ORLANDO_PASSWORD password required to POST /annotations/orlando """ import base64 import json import os import sys import tempfile from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer PORT = int(os.environ.get("PORT", "8080")) ARTIST_PASSWORDS = { "ruby": os.environ.get("RUBY_PASSWORD", ""), "orlando": os.environ.get("ORLANDO_PASSWORD", ""), } def _basic(user: str, pw: str) -> str: return "Basic " + base64.b64encode(f"{user}:{pw}".encode("utf-8")).decode("ascii") EXPECTED_BASIC = { artist: (_basic(artist, pw) if pw else None) for artist, pw in ARTIST_PASSWORDS.items() } for artist, pw in ARTIST_PASSWORDS.items(): if not pw: sys.stderr.write( f"[srv] WARNING: {artist.upper()}_PASSWORD not set -- " f"POST /annotations/{artist} will be refused.\n" ) def ann_path(artist: str) -> str: return os.path.abspath(f"annotations-{artist}.json") def safe_write_json(path: str, data) -> None: d = os.path.dirname(path) or "." os.makedirs(d, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=d, prefix=".ann-", suffix=".tmp") try: with os.fdopen(fd, "w") as f: json.dump(data, f, indent=2) f.flush() os.fsync(f.fileno()) os.replace(tmp, path) except Exception: try: os.unlink(tmp) except OSError: pass raise def safe_read_json(path: str, default): try: with open(path, "r") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return default class Handler(SimpleHTTPRequestHandler): def log_message(self, fmt, *args): sys.stderr.write("[srv] " + fmt % args + "\n") def _json(self, code: int, payload) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _send_401(self, artist: str): self.send_response(401) self.send_header( "WWW-Authenticate", f'Basic realm="LableBrain-{artist}"', ) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"Authentication required") def _redirect(self, location: str, code: int = 301): self.send_response(code) self.send_header("Location", location) self.send_header("Content-Length", "0") self.end_headers() def _match_annotations(self, path: str): p = path.split("?", 1)[0].strip("/") if not p.startswith("annotations/"): return None artist = p[len("annotations/"):].strip("/").lower() if artist in ARTIST_PASSWORDS: return artist return None def _maybe_rewrite_clean_url(self): """ Clean-URL support: - /foo.html -> 301 redirect to /foo (canonical) - /foo -> internally rewrite to /foo.html if foo.html exists Only applies to .html; assets with other extensions pass through. Skips /annotations/* because that's an API. """ raw = self.path if "?" in raw: path, qs = raw.split("?", 1) qs = "?" + qs else: path, qs = raw, "" if path.startswith("/annotations/"): return False # Case 1: explicit .html -> canonical clean URL if path.endswith(".html"): clean = path[:-5] # strip ".html" if clean == "": return False # root, leave alone self._redirect(clean + qs, 301) return True # Case 2: bare clean URL -> serve underlying .html if present # (Skip paths with a file extension, and skip the root "/".) basename = path.rsplit("/", 1)[-1] if path in ("", "/") or "." in basename: return False candidate_rel = path.lstrip("/") + ".html" candidate_abs = os.path.join(os.getcwd(), candidate_rel) if os.path.isfile(candidate_abs): self.path = "/" + candidate_rel + qs return False # let super().do_GET() serve the rewritten path return False def do_GET(self): artist = self._match_annotations(self.path) if artist is not None: data = safe_read_json(ann_path(artist), []) return self._json(200, data) if self._maybe_rewrite_clean_url(): return # redirect already sent return super().do_GET() def do_POST(self): artist = self._match_annotations(self.path) if artist is None: return self._json(404, {"error": "not found"}) expected = EXPECTED_BASIC.get(artist) if not expected: return self._json(503, {"error": f"{artist} password not configured"}) header = self.headers.get("Authorization", "") if header != expected: return self._send_401(artist) length = int(self.headers.get("Content-Length", "0") or "0") if length <= 0 or length > 1_000_000: return self._json(400, {"error": "bad length"}) raw = self.rfile.read(length) try: payload = json.loads(raw.decode("utf-8")) except Exception: return self._json(400, {"error": "bad json"}) if not isinstance(payload, list): return self._json(400, {"error": "payload must be a list"}) cleaned = [] for item in payload: if not isinstance(item, dict): continue cleaned.append({ "id": str(item.get("id", ""))[:64], "timestamp": str(item.get("timestamp", ""))[:40], "tags": [str(t)[:40] for t in (item.get("tags") or [])][:10], "note": str(item.get("note", ""))[:2000], "created": str(item.get("created", ""))[:40], }) try: safe_write_json(ann_path(artist), cleaned) except Exception as e: return self._json(500, {"error": f"write failed: {e}"}) return self._json(200, {"ok": True, "count": len(cleaned)}) def main(): srv = ThreadingHTTPServer((os.environ.get("BIND_HOST", "127.0.0.1"), PORT), Handler) sys.stderr.write(f"[srv] listening on :{PORT} (cwd={os.getcwd()})\n") try: srv.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()