From e67dc50105a904dc39171b539a5f673cd631894d Mon Sep 17 00:00:00 2001 From: dev Date: Fri, 30 Oct 2026 01:23:44 +0000 Subject: [PATCH] Add helper/server.js --- helper/server.js | 661 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 661 insertions(+) create mode 100644 helper/server.js diff --git a/helper/server.js b/helper/server.js new file mode 100644 index 0000000..81c853e --- /dev/null +++ b/helper/server.js @@ -0,0 +1,661 @@ +import crypto from "crypto"; +import express from "express"; +import http from "http"; +import { WebSocketServer } from "ws"; + +const PORT = Number(process.env.PORT || 8080); +const HOST = process.env.HOST || "0.0.0.0"; +const CALL_HELPER_SHARED_SECRET = String( + process.env.CALL_HELPER_SHARED_SECRET || "" +).trim(); +const PUBLIC_BASE = String( + process.env.PUBLIC_BASE || `http://127.0.0.1:${PORT}` +).replace(/\/+$/, ""); + +const HELPER_LOG_LEVEL = String(process.env.HELPER_LOG_LEVEL || "info").toLowerCase(); +const HELPER_DEFAULT_ROUTE_MODE = String( + process.env.HELPER_DEFAULT_ROUTE_MODE || "direct" +).toLowerCase(); + +const ENABLE_OVERLAY_METADATA = String( + process.env.ENABLE_OVERLAY_METADATA || "1" +) === "1"; + +const OVERLAY_YGG_ENABLED = String( + process.env.OVERLAY_YGG_ENABLED || "0" +) === "1"; +const OVERLAY_YGG_BRIDGE_WSS = String( + process.env.OVERLAY_YGG_BRIDGE_WSS || "" +).trim(); +const OVERLAY_YGG_BRIDGE_LABEL = String( + process.env.OVERLAY_YGG_BRIDGE_LABEL || "ygg-bridge" +).trim(); + +const OVERLAY_I2P_ENABLED = String( + process.env.OVERLAY_I2P_ENABLED || "0" +) === "1"; +const OVERLAY_I2P_BRIDGE_HTTPS = String( + process.env.OVERLAY_I2P_BRIDGE_HTTPS || "" +).trim(); +const OVERLAY_I2P_BRIDGE_LABEL = String( + process.env.OVERLAY_I2P_BRIDGE_LABEL || "i2p-bridge" +).trim(); + +const RELAY_ENABLED = String(process.env.RELAY_ENABLED || "1") === "1"; +const RELAY_WSS = String(process.env.RELAY_WSS || "").trim(); + +if (!CALL_HELPER_SHARED_SECRET) { + console.error("missing CALL_HELPER_SHARED_SECRET"); + process.exit(1); +} + +const app = express(); +app.use(express.json({ limit: "64kb" })); + +const server = http.createServer(app); +const wss = new WebSocketServer({ noServer: true }); + +/** + * in-memory state + * + * calls: + * callId -> { + * callId, + * room, + * routeMode, + * participants: Map(peerId -> participant), + * createdAt, + * updatedAt + * } + * + * tickets: + * ticket -> { + * ticket, + * callId, + * peerId, + * sub, + * username, + * role, + * room, + * exp, + * routeProfile + * } + */ +const calls = new Map(); +const tickets = new Map(); +const recentLogs = []; +const MAX_LOGS = 500; + +function nowMs() { + return Date.now(); +} + +function nowSeconds() { + return Math.floor(Date.now() / 1000); +} + +function isoNow() { + return new Date().toISOString(); +} + +function log(level, message, meta = {}) { + const allowed = ["debug", "info", "warn", "error"]; + const currentIdx = allowed.indexOf(HELPER_LOG_LEVEL); + const entryIdx = allowed.indexOf(level); + const item = { + ts: isoNow(), + level, + message, + meta + }; + + recentLogs.push(item); + if (recentLogs.length > MAX_LOGS) { + recentLogs.shift(); + } + + if (entryIdx >= currentIdx) { + const method = + level === "error" ? "error" : + level === "warn" ? "warn" : + "log"; + console[method](`[${item.ts}] [${level}] ${message}`, meta); + } +} + +function b64urlDecode(value) { + const pad = value.length % 4 === 0 ? "" : "=".repeat(4 - (value.length % 4)); + const normalized = value.replace(/-/g, "+").replace(/_/g, "/") + pad; + return Buffer.from(normalized, "base64"); +} + +function verifySignedToken(token) { + if (!token || typeof token !== "string" || !token.includes(".")) { + throw new Error("invalid token format"); + } + + const [payloadPart, sigPart] = token.split(".", 2); + const payloadBytes = b64urlDecode(payloadPart); + const sigBytes = b64urlDecode(sigPart); + + const expected = crypto + .createHmac("sha256", CALL_HELPER_SHARED_SECRET) + .update(payloadBytes) + .digest(); + + if ( + sigBytes.length !== expected.length || + !crypto.timingSafeEqual(sigBytes, expected) + ) { + throw new Error("invalid token signature"); + } + + let payload; + try { + payload = JSON.parse(payloadBytes.toString("utf8")); + } catch { + throw new Error("invalid token payload json"); + } + + if (!payload || typeof payload !== "object") { + throw new Error("invalid token payload"); + } + + if (!payload.exp || Number(payload.exp) <= nowSeconds()) { + throw new Error("token expired"); + } + + if (!payload.can_call) { + throw new Error("token not authorized for calls"); + } + + if (!payload.call_id || !payload.sub) { + throw new Error("token missing required fields"); + } + + return payload; +} + +function randomId(prefix = "") { + return `${prefix}${crypto.randomBytes(8).toString("hex")}`; +} + +function chooseRouteMode(payload) { + const requested = String(payload.route_mode || "").trim().toLowerCase(); + + if (requested === "overlay_ygg" && OVERLAY_YGG_ENABLED) { + return "overlay_ygg"; + } + + if (requested === "overlay_i2p" && OVERLAY_I2P_ENABLED) { + return "overlay_i2p"; + } + + if (HELPER_DEFAULT_ROUTE_MODE === "overlay_ygg" && OVERLAY_YGG_ENABLED) { + return "overlay_ygg"; + } + + if (HELPER_DEFAULT_ROUTE_MODE === "overlay_i2p" && OVERLAY_I2P_ENABLED) { + return "overlay_i2p"; + } + + return "direct"; +} + +function buildRouteCandidates(callId) { + const routes = []; + + routes.push({ + kind: "direct", + target: callId, + priority: 100, + available: true, + label: "direct" + }); + + if (ENABLE_OVERLAY_METADATA && OVERLAY_YGG_ENABLED) { + routes.push({ + kind: "overlay_ygg", + target: OVERLAY_YGG_BRIDGE_WSS || PUBLIC_BASE.replace(/^http/, "ws"), + priority: 90, + available: Boolean(OVERLAY_YGG_BRIDGE_WSS), + label: OVERLAY_YGG_BRIDGE_LABEL + }); + } + + if (ENABLE_OVERLAY_METADATA && OVERLAY_I2P_ENABLED) { + routes.push({ + kind: "overlay_i2p", + target: OVERLAY_I2P_BRIDGE_HTTPS || PUBLIC_BASE, + priority: 40, + available: Boolean(OVERLAY_I2P_BRIDGE_HTTPS), + label: OVERLAY_I2P_BRIDGE_LABEL + }); + } + + if (RELAY_ENABLED) { + routes.push({ + kind: "relay", + target: RELAY_WSS || PUBLIC_BASE.replace(/^http/, "ws"), + priority: 20, + available: true, + label: "relay" + }); + } + + return routes; +} + +function buildTransportPolicy(routeMode) { + const order = []; + if (routeMode === "overlay_ygg") { + order.push("overlay_ygg", "direct", "relay"); + } else if (routeMode === "overlay_i2p") { + order.push("direct", "relay"); + } else { + order.push("direct", "relay"); + } + + return { + preferred_order: order, + allow_relay_fallback: RELAY_ENABLED, + private_service_transport: OVERLAY_I2P_ENABLED ? "overlay_i2p" : null + }; +} + +function getOrCreateCall(callId, room, routeMode = "direct") { + let call = calls.get(callId); + if (!call) { + call = { + callId, + room, + routeMode, + participants: new Map(), + createdAt: nowMs(), + updatedAt: nowMs() + }; + calls.set(callId, call); + log("info", "created call", { callId, room, routeMode }); + } + return call; +} + +function buildWsUrl(ticket) { + const publicBase = PUBLIC_BASE.replace(/^http:/, "ws:").replace(/^https:/, "wss:"); + return `${publicBase}/ws?ticket=${encodeURIComponent(ticket)}`; +} + +function sendJson(ws, payload) { + if (!ws || ws.readyState !== ws.OPEN) return; + ws.send(JSON.stringify(payload)); +} + +function broadcast(call, payload, exceptPeerId = null) { + for (const [peerId, participant] of call.participants.entries()) { + if (exceptPeerId && peerId === exceptPeerId) continue; + sendJson(participant.ws, payload); + } +} + +function callSnapshot(call) { + return { + call_id: call.callId, + room: call.room, + route_mode: call.routeMode, + participant_count: call.participants.size, + created_at: new Date(call.createdAt).toISOString(), + updated_at: new Date(call.updatedAt).toISOString(), + participants: [...call.participants.values()].map(participant => ({ + peer_id: participant.peerId, + sub: participant.sub, + username: participant.username, + role: participant.role, + route_mode: participant.routeMode, + joined_at: new Date(participant.joinedAt).toISOString() + })) + }; +} + +function statusSnapshot() { + return { + ok: true, + ts: isoNow(), + public_base: PUBLIC_BASE, + route_defaults: { + default_route_mode: HELPER_DEFAULT_ROUTE_MODE, + overlay_metadata_enabled: ENABLE_OVERLAY_METADATA, + relay_enabled: RELAY_ENABLED + }, + overlays: { + ygg: { + enabled: OVERLAY_YGG_ENABLED, + bridge_label: OVERLAY_YGG_BRIDGE_LABEL, + bridge_wss: OVERLAY_YGG_BRIDGE_WSS || null + }, + i2p: { + enabled: OVERLAY_I2P_ENABLED, + bridge_label: OVERLAY_I2P_BRIDGE_LABEL, + bridge_https: OVERLAY_I2P_BRIDGE_HTTPS || null + } + }, + counts: { + active_calls: calls.size, + active_tickets: tickets.size + } + }; +} + +function removeParticipant(callId, peerId, notify = true) { + const call = calls.get(callId); + if (!call) return; + + const participant = call.participants.get(peerId); + if (!participant) return; + + call.participants.delete(peerId); + call.updatedAt = nowMs(); + + log("info", "participant removed", { + callId, + peerId, + username: participant.username + }); + + if (notify) { + broadcast(call, { + type: "participant-left", + participantId: peerId, + roomId: callId + }); + } + + if (call.participants.size === 0) { + calls.delete(callId); + log("info", "call removed because empty", { callId }); + } +} + +app.get("/healthz", (_req, res) => { + res.json(statusSnapshot()); +}); + +app.get("/status", (_req, res) => { + res.json({ + ...statusSnapshot(), + calls: [...calls.values()].map(callSnapshot) + }); +}); + +app.get("/logs", (req, res) => { + const limit = Math.max(1, Math.min(200, Number(req.query.limit || 50))); + res.json({ + ok: true, + logs: recentLogs.slice(-limit) + }); +}); + +app.post("/api/call/session", (req, res) => { + try { + const token = req.body?.token; + const payload = verifySignedToken(token); + + const callId = String(payload.call_id); + const room = String(payload.room || callId); + const peerId = randomId("peer_"); + const ticket = randomId("ticket_"); + const routeMode = chooseRouteMode(payload); + const routeCandidates = buildRouteCandidates(callId); + const transportPolicy = buildTransportPolicy(routeMode); + + getOrCreateCall(callId, room, routeMode); + + tickets.set(ticket, { + ticket, + callId, + peerId, + sub: String(payload.sub), + username: String(payload.username || peerId), + role: String(payload.role || "user"), + room, + exp: Number(payload.exp), + routeProfile: { + routeMode, + routeCandidates, + transportPolicy + } + }); + + log("info", "created call session ticket", { + callId, + peerId, + username: payload.username, + routeMode + }); + + return res.json({ + ok: true, + call_id: callId, + peer_id: peerId, + ws_url: buildWsUrl(ticket), + iceServers: [ + { + urls: ["stun:stun.l.google.com:19302"] + } + ], + transport_policy: transportPolicy, + route_candidates: routeCandidates + }); + } catch (error) { + log("warn", "call session rejected", { + error: error.message + }); + + return res.status(401).json({ + ok: false, + error: error.message || "invalid call token" + }); + } +}); + +server.on("upgrade", (req, socket, head) => { + const url = new URL(req.url, `http://${req.headers.host}`); + if (url.pathname !== "/ws") { + socket.destroy(); + return; + } + + const ticket = url.searchParams.get("ticket"); + const ticketData = ticket ? tickets.get(ticket) : null; + + if (!ticketData || ticketData.exp <= nowSeconds()) { + socket.destroy(); + return; + } + + wss.handleUpgrade(req, socket, head, ws => { + ws._ticketData = ticketData; + wss.emit("connection", ws, req); + }); +}); + +wss.on("connection", ws => { + const ticketData = ws._ticketData; + if (!ticketData) { + ws.close(); + return; + } + + tickets.delete(ticketData.ticket); + + const call = getOrCreateCall( + ticketData.callId, + ticketData.room, + ticketData.routeProfile?.routeMode || "direct" + ); + + const participant = { + peerId: ticketData.peerId, + sub: ticketData.sub, + username: ticketData.username, + role: ticketData.role, + room: ticketData.room, + routeMode: ticketData.routeProfile?.routeMode || "direct", + routeCandidates: ticketData.routeProfile?.routeCandidates || [], + transportPolicy: ticketData.routeProfile?.transportPolicy || {}, + ws, + joinedAt: nowMs() + }; + + call.participants.set(ticketData.peerId, participant); + call.updatedAt = nowMs(); + + log("info", "participant joined websocket", { + callId: call.callId, + peerId: participant.peerId, + username: participant.username, + routeMode: participant.routeMode + }); + + sendJson(ws, { + type: "joined", + participantId: ticketData.peerId, + roomId: call.callId, + transportPolicy: participant.transportPolicy, + routeCandidates: participant.routeCandidates + }); + + for (const [peerId, existing] of call.participants.entries()) { + if (peerId === ticketData.peerId) continue; + sendJson(ws, { + type: "participant-joined", + participantId: peerId, + roomId: call.callId, + username: existing.username, + routeMode: existing.routeMode + }); + } + + broadcast( + call, + { + type: "participant-joined", + participantId: ticketData.peerId, + roomId: call.callId, + username: participant.username, + routeMode: participant.routeMode + }, + ticketData.peerId + ); + + ws.on("message", raw => { + let msg; + try { + msg = JSON.parse(String(raw)); + } catch { + log("warn", "invalid ws json from peer", { + callId: call.callId, + peerId: participant.peerId + }); + return; + } + + const senderPeerId = ticketData.peerId; + call.updatedAt = nowMs(); + + if (!msg || typeof msg !== "object" || !msg.type) { + return; + } + + if (msg.type === "hangup") { + log("info", "peer requested hangup", { + callId: call.callId, + peerId: senderPeerId + }); + removeParticipant(call.callId, senderPeerId, true); + try { + ws.close(); + } catch {} + return; + } + + if (msg.type === "offer" || msg.type === "answer") { + const to = String(msg.to || "").trim(); + if (!to) return; + const target = call.participants.get(to); + if (!target) { + log("warn", "signal target missing", { + callId: call.callId, + from: senderPeerId, + to, + type: msg.type + }); + return; + } + + sendJson(target.ws, { + type: msg.type, + from: senderPeerId, + sdp: msg.sdp, + username: participant.username, + roomId: call.callId, + routeMode: participant.routeMode + }); + return; + } + + if (msg.type === "ice") { + const to = String(msg.to || "").trim(); + if (!to) return; + const target = call.participants.get(to); + if (!target) { + log("warn", "ice target missing", { + callId: call.callId, + from: senderPeerId, + to + }); + return; + } + + sendJson(target.ws, { + type: "ice", + from: senderPeerId, + candidate: msg.candidate, + sdpMid: msg.sdpMid, + sdpMLineIndex: msg.sdpMLineIndex, + username: participant.username, + roomId: call.callId, + routeMode: participant.routeMode + }); + return; + } + }); + + ws.on("close", () => { + log("info", "websocket closed", { + callId: call.callId, + peerId: participant.peerId + }); + removeParticipant(call.callId, ticketData.peerId, true); + }); + + ws.on("error", error => { + log("warn", "websocket error", { + callId: call.callId, + peerId: participant.peerId, + error: error?.message || "unknown" + }); + removeParticipant(call.callId, ticketData.peerId, true); + }); +}); + +server.listen(PORT, HOST, () => { + log("info", "calls helper listening", { + host: HOST, + port: PORT, + publicBase: PUBLIC_BASE, + overlays: { + ygg: OVERLAY_YGG_ENABLED, + i2p: OVERLAY_I2P_ENABLED + } + }); +}); \ No newline at end of file