import crypto from "crypto"; import express from "express"; import http from "http"; import { WebSocketServer } from "ws"; const TURN_URLS = String(process.env.TURN_URLS || "") .split(",") .map(value => value.trim()) .filter(Boolean); const TURN_USERNAME = String(process.env.TURN_USERNAME || "").trim(); const TURN_PASSWORD = String(process.env.TURN_PASSWORD || "").trim(); 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(); const CORS_ALLOWED_ORIGINS = String( process.env.CORS_ALLOWED_ORIGINS || "https://angel.33.backend-3.com" ) .split(",") .map(value => value.trim()) .filter(Boolean); const allowedOrigins = new Set(CORS_ALLOWED_ORIGINS); 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 = Math.max(0, allowed.indexOf(HELPER_LOG_LEVEL)); const entryIdx = Math.max(0, 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 isAllowedOrigin(origin) { if (!origin) return false; return allowedOrigins.has(origin); } function applyCors(req, res) { const origin = String(req.headers.origin || "").trim(); if (origin && isAllowedOrigin(origin)) { res.setHeader("Access-Control-Allow-Origin", origin); res.setHeader("Vary", "Origin"); res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS"); res.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization"); } } app.use((req, res, next) => { applyCors(req, res); if (req.method === "OPTIONS") { return res.status(204).end(); } next(); }); 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 getIceServers() { const servers = [ { urls: ["stun:stun.l.google.com:19302"] } ]; if (TURN_URLS.length && TURN_USERNAME && TURN_PASSWORD) { servers.push({ urls: TURN_URLS, username: TURN_USERNAME, credential: TURN_PASSWORD }); } return servers; } 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 } }, turn: { enabled: Boolean(TURN_URLS.length && TURN_USERNAME && TURN_PASSWORD), urls: TURN_URLS }, cors: { allowed_origins: [...allowedOrigins] }, 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) => { applyCors(req, res); res.json(statusSnapshot()); }); app.get("/status", (req, res) => { applyCors(req, res); res.json({ ...statusSnapshot(), calls: [...calls.values()].map(callSnapshot) }); }); app.get("/logs", (req, res) => { applyCors(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) => { applyCors(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, origin: req.headers.origin || null, turnEnabled: Boolean(TURN_URLS.length && TURN_USERNAME && TURN_PASSWORD) }); return res.json({ ok: true, call_id: callId, peer_id: peerId, ws_url: buildWsUrl(ticket), iceServers: getIceServers(), transport_policy: transportPolicy, route_candidates: routeCandidates }); } catch (error) { log("warn", "call session rejected", { error: error.message, origin: req.headers.origin || null }); return res.status(401).json({ ok: false, error: error.message || "invalid call token" }); } }); server.on("upgrade", (req, socket, head) => { const origin = String(req.headers.origin || "").trim(); if (origin && !isAllowedOrigin(origin)) { log("warn", "websocket upgrade rejected by origin", { origin }); socket.destroy(); return; } 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()) { log("warn", "websocket upgrade rejected by invalid/expired ticket", { origin: origin || null }); 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 }, turn: { enabled: Boolean(TURN_URLS.length && TURN_USERNAME && TURN_PASSWORD), urls: TURN_URLS }, corsAllowedOrigins: [...allowedOrigins] }); });