dev: one-command local calls stack (LiveKit + JWT issuer + voice-limit guard + https well-known) and a seeded voice room

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPmy3tPq869XDW4njjVaKA

scripts/dev-lk-jwt.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""Minimal lk-jwt-service clone for local testing.
+POST /sfu/get {room, openid_token{access_token, matrix_server_name}, device_id}
+ -> validates the OpenID token against Synapse (federation openid/userinfo),
+    mints a LiveKit JWT for identity "<user_id>:<device_id>" and returns {url, jwt}.
+"""
+import base64, hashlib, hmac, json, os, time, urllib.parse, urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+PORT = int(os.environ.get("PORT", "8071"))
+LK_KEY = os.environ.get("LIVEKIT_KEY", "devkey")
+LK_SECRET = os.environ.get("LIVEKIT_SECRET", "devsecretdevsecretdevsecretdevsecret")
+LK_URL = os.environ.get("LIVEKIT_URL", "ws://127.0.0.1:7880")
+SYNAPSE = os.environ.get("SYNAPSE_API", "http://127.0.0.1:8008")
+
+def b64(b): return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
+def jwt(payload):
+    h = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()); p = b64(json.dumps(payload).encode())
+    sig = b64(hmac.new(LK_SECRET.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest())
+    return f"{h}.{p}.{sig}"
+
+def userinfo(access_token):
+    q = urllib.parse.urlencode({"access_token": access_token})
+    with urllib.request.urlopen(f"{SYNAPSE}/_matrix/federation/v1/openid/userinfo?{q}", timeout=5) as r:
+        return json.load(r)["sub"]
+
+class H(BaseHTTPRequestHandler):
+    def _cors(self):
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Access-Control-Allow-Headers", "*")
+        self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
+    def do_OPTIONS(self):
+        self.send_response(204); self._cors(); self.end_headers()
+    def do_POST(self):
+        n = int(self.headers.get("Content-Length") or 0)
+        data = json.loads(self.rfile.read(n) or b"{}")
+        room = data.get("room") or data.get("room_id") or ""
+        oid = data.get("openid_token") or {}
+        try:
+            user = userinfo(oid.get("access_token", ""))
+        except Exception as e:  # noqa
+            self.send_response(401); self._cors(); self.end_headers(); self.wfile.write(json.dumps({"errcode": "M_LOOKUP_FAILED", "error": str(e)}).encode()); return
+        identity = f"{user}:{data.get('device_id', '')}"
+        now = int(time.time())
+        token = jwt({"iss": LK_KEY, "sub": identity, "name": user, "nbf": now - 10, "exp": now + 3600,
+                     "video": {"room": room, "roomJoin": True, "roomCreate": True, "canPublish": True, "canSubscribe": True, "canPublishData": True}})
+        body = json.dumps({"url": LK_URL, "jwt": token}).encode()
+        self.send_response(200); self._cors(); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
+        print(f"issued {identity} room={room}", flush=True)
+    def log_message(self, *a): pass
+
+ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()