jaredandClaude Opus 5
da2d7a7d4f
CI / Build & Quality Checks (push) Successful in 1m42s
CI / Docker image build & smoke test (push) Skipped
CI / Secret scan (gitleaks) (push) Successful in 6s
CI / Trigger Desktop Build (push) Successful in 7s
CI / Playwright smoke (e2e) (push) Successful in 2m22s
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPmy3tPq869XDW4njjVaKA
53 lines
2.9 KiB
Python
53 lines
2.9 KiB
Python
#!/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()
|