107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
cc-sidepanel bridge — localhost HTTP shim so a browser extension
|
|
(no unix-socket access) can talk to injectd.py's unix socket.
|
|
|
|
Endpoints (GET/POST, JSON):
|
|
GET /seats -> injectd {op: seats}
|
|
GET /screen?seat=X&lines=60 -> injectd {op: screen}
|
|
POST /inject {seat, text} -> injectd {op: inject}
|
|
|
|
CORS is locked to chrome-extension:// origins only (set via env
|
|
CCSIDEPANEL_EXT_ID, no wildcard fallback — error loud if unset).
|
|
"""
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
INJECTD_SOCK = "/tmp/injectd.sock"
|
|
PORT = int(os.environ.get("CCSIDEPANEL_PORT", "8988"))
|
|
EXT_ID = os.environ.get("CCSIDEPANEL_EXT_ID")
|
|
|
|
if not EXT_ID:
|
|
print("CCSIDEPANEL_EXT_ID not set — refusing to guess an allowed origin. "
|
|
"Load the unpacked extension, copy its ID from brave://extensions, "
|
|
"export CCSIDEPANEL_EXT_ID=<id> before starting bridge.py.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
ALLOWED_ORIGIN = "chrome-extension://{}".format(EXT_ID)
|
|
|
|
|
|
def injectd_call(req):
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
try:
|
|
s.connect(INJECTD_SOCK)
|
|
except (FileNotFoundError, ConnectionRefusedError) as exc:
|
|
return {"ok": False, "error": "injectd not reachable at {}: {}".format(INJECTD_SOCK, exc)}
|
|
s.sendall((json.dumps(req) + "\n").encode())
|
|
data = b""
|
|
while not data.endswith(b"\n"):
|
|
chunk = s.recv(65536)
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
s.close()
|
|
return json.loads(data.decode())
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def _cors(self):
|
|
self.send_header("Access-Control-Allow-Origin", ALLOWED_ORIGIN)
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
|
|
def _json(self, obj, status=200):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(status)
|
|
self._cors()
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_OPTIONS(self):
|
|
self.send_response(204)
|
|
self._cors()
|
|
self.end_headers()
|
|
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
qs = parse_qs(parsed.query)
|
|
if parsed.path == "/seats":
|
|
return self._json(injectd_call({"op": "seats"}))
|
|
if parsed.path == "/screen":
|
|
seat = qs.get("seat", [None])[0]
|
|
if not seat:
|
|
return self._json({"ok": False, "error": "seat required"}, 400)
|
|
lines = int(qs.get("lines", ["60"])[0])
|
|
return self._json(injectd_call({"op": "screen", "seat": seat, "lines": lines}))
|
|
self._json({"ok": False, "error": "unknown path"}, 404)
|
|
|
|
def do_POST(self):
|
|
parsed = urlparse(self.path)
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
try:
|
|
payload = json.loads(raw.decode())
|
|
except json.JSONDecodeError:
|
|
return self._json({"ok": False, "error": "bad json"}, 400)
|
|
if parsed.path == "/inject":
|
|
seat, text = payload.get("seat"), payload.get("text")
|
|
if not seat or not text:
|
|
return self._json({"ok": False, "error": "seat and text required"}, 400)
|
|
return self._json(injectd_call({"op": "inject", "seat": seat, "text": text}))
|
|
self._json({"ok": False, "error": "unknown path"}, 404)
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass # keep stdout quiet; run under a real logger if this graduates
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("cc-sidepanel bridge on :{} -> {} (allowed origin: {})".format(
|
|
PORT, INJECTD_SOCK, ALLOWED_ORIGIN))
|
|
ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
|