tglock/worker/tglock-worker.test.js

import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import test from "node:test";
import vm from "node:vm";

const source = await readFile(new URL("./tglock-worker.js", import.meta.url), "utf8");
const tick = () => new Promise((resolve) => setImmediate(resolve));
function deferred() {
  let resolve, reject;
  const promise = new Promise((a, b) => { resolve = a; reject = b; });
  return { promise, resolve, reject };
}

// Execute the actual standalone deployment file, replacing only Cloudflare's
// platform primitives. No test-only implementation of the relay is used.
async function fixture({ opened = Promise.resolve(), write, timeout = false } = {}) {
  let readController;
  const readable = new ReadableStream({ start(controller) { readController = controller; } });
  const reads = { reject(error) { readController.error(error); } };
  const closed = deferred();
  const writes = [];
  const events = new Map();
  const state = { connects: 0, accepted: false, socketCloses: 0, sent: [], closes: [], warnings: [] };
  const server = {
    accept() { state.accepted = true; state.binaryTypeOnAccept = this.binaryType; },
    addEventListener(name, callback) { events.set(name, callback); },
    close(code, reason) { state.closes.push({ code, reason }); },
    send(chunk) { state.sent.push(Array.from(chunk)); },
  };
  const socket = {
    opened, closed: closed.promise,
    writable: { getWriter: () => ({
      ready: Promise.resolve(),
      async write(chunk) { writes.push(Array.from(chunk)); if (write) await write(chunk); },
    }) },
    readable,
    async close() {
      state.socketCloses++;
      try { readController.close(); } catch { /* already closed or errored */ }
      closed.resolve();
    },
  };
  const context = vm.createContext({
    URL, Uint8Array, ArrayBuffer, Blob, console: { warn: (line) => state.warnings.push(line) }, setTimeout: timeout ? (fn) => setTimeout(fn, 0) : setTimeout, clearTimeout,
    Response: class {
      constructor(body, options) { this.body = body; Object.assign(this, options); }
    },
    WebSocketPair: class { constructor() { this[0] = {}; this[1] = server; } },
  });
  const sockets = new vm.SyntheticModule(["connect"], function () {
    this.setExport("connect", (address) => {
      state.connects++;
      state.address = address;
      return socket;
    });
  }, { context });
  const module = new vm.SourceTextModule(source, { context });
  await module.link((name) => {
    assert.equal(name, "cloudflare:sockets");
    return sockets;
  });
  await module.evaluate();
  return {
    ...state,
    state, writes, reads, closed, socket,
    download(chunk) { readController.enqueue(chunk); },
    eof() { readController.close(); },
    message(data) { events.get("message")({ data }); },
    event(name) { events.get(name)({}); },
    fetch(path = "/apiws?dst=149.154.167.51&dc=2", headers = {}, env = {}) {
      return module.namespace.default.fetch(new Request(`https://example.workers.dev${path}`, {
        headers: { Upgrade: "websocket", "Sec-WebSocket-Protocol": "binary", ...headers },
      }), env);
    },
  };
}

test("HTTP deployment check does not claim Telegram connectivity", async () => {
  const f = await fixture();
  assert.equal((await f.fetch("/apiws", { Upgrade: "" })).status, 426);
  assert.equal(f.state.connects, 0);
});

test("reject unauthorized destinations, token and subprotocol before opening TCP", async () => {
  const f = await fixture();
  assert.equal((await f.fetch("/wrong")).status, 404);
  assert.equal((await f.fetch("/apiws?dst=127.0.0.1")).status, 403);
  assert.equal((await f.fetch(undefined, {}, { TGLOCK_TOKEN: "test-token" })).status, 403);
  assert.equal((await f.fetch(undefined, { "Sec-WebSocket-Protocol": "chat" })).status, 400);
  assert.equal(f.state.connects, 0);
});

test("wait for upstream TCP before accepting WebSocket", async () => {
  const opened = deferred();
  const f = await fixture({ opened: opened.promise });
  const response = f.fetch();
  await tick();
  assert.equal(f.state.accepted, false);
  opened.resolve();
  const result = await response;
  assert.equal(result.status, 101);
  assert.equal(result.headers["Sec-WebSocket-Protocol"], "binary");
  assert.equal(f.state.accepted, true);
  f.event("close");
});

test("failed upstream returns HTTP 502, never a successful tunnel", async () => {
  const f = await fixture({ opened: Promise.reject(new Error("unreachable")) });
  assert.equal((await f.fetch()).status, 502);
  assert.equal(f.state.accepted, false);
  assert.equal(f.state.socketCloses, 1);
});

test("upstream timeout returns 504 and cancels socket", async () => {
  const f = await fixture({ opened: new Promise(() => {}), timeout: true });
  assert.equal((await f.fetch()).status, 504);
  assert.equal(f.state.accepted, false);
  assert.equal(f.state.socketCloses, 1);
});

test("serialize binary writes, preserve bytes and view offsets", async () => {
  const firstWrite = deferred();
  let calls = 0;
  const f = await fixture({ write: async () => { if (++calls === 1) await firstWrite.promise; } });
  await f.fetch();
  f.message(new Uint8Array([1, 2]).buffer);
  f.message(new Uint8Array([99, 3, 4, 88]).subarray(1, 3));
  await tick();
  assert.deepEqual(f.writes, [[1, 2]]);
  firstWrite.resolve();
  await tick();
  assert.deepEqual(f.writes, [[1, 2], [3, 4]]);
  f.event("close");
});

test("binary frames arrive as ArrayBuffer on new compatibility dates", async () => {
  // With compatibility date 2026-03-17+ Cloudflare delivers Blob unless
  // binaryType is set before accept(); the Worker then closed every tunnel
  // with 1003 right after the handshake (#61).
  const f = await fixture();
  await f.fetch();
  assert.equal(f.state.binaryTypeOnAccept, "arraybuffer");
  f.event("close");
});

test("Blob frames are still relayed in order, never closed as text", async () => {
  const f = await fixture();
  await f.fetch();
  f.message(new Blob([new Uint8Array([7, 8])]));
  f.message(new Uint8Array([9]).buffer);
  for (let i = 0; i < 5; i++) await tick();
  assert.deepEqual(f.writes, [[7, 8], [9]]);
  assert.deepEqual(f.state.closes, []);
  f.event("close");
});

test("overflow closes socket and discards queued writes", async () => {
  const f = await fixture();
  await f.fetch();
  f.message(new Uint8Array(1024 * 1024).buffer);
  f.message(new Uint8Array([1]).buffer);
  await tick();
  assert.equal(f.state.closes[0].code, 1009);
  assert.equal(f.state.socketCloses, 1);
  assert.equal(f.writes.length, 0);
});

test("empty-message floods have a bounded queue too", async () => {
  const f = await fixture();
  await f.fetch();
  for (let i = 0; i < 257; i++) f.message(new ArrayBuffer(0));
  await tick();
  assert.equal(f.state.closes[0].code, 1009);
  assert.equal(f.state.socketCloses, 1);
});

test("text and write failures get diagnostic close codes", async () => {
  const f = await fixture();
  await f.fetch();
  f.message("not binary");
  assert.equal(f.state.closes[0].code, 1003);
  const failing = await fixture({ write: async () => { throw new Error("reset"); } });
  await failing.fetch();
  failing.message(new Uint8Array([1]).buffer);
  await tick();
  assert.equal(failing.state.closes[0].code, 1011);
  assert.equal(failing.state.socketCloses, 1);
});

test("read errors propagate and close both socket directions", async () => {
  const f = await fixture();
  await f.fetch();
  f.reads.reject(new Error("reset"));
  await tick();
  assert.equal(f.state.closes[0].code, 1011);
  assert.equal(f.state.socketCloses, 1);
});

test("Telegram bytes reach WebSocket unchanged and EOF closes both directions", async () => {
  const f = await fixture();
  await f.fetch();
  f.download(new Uint8Array([0, 255, 128]));
  f.download(new Uint8Array([1, 2]));
  f.eof();
  await tick();
  assert.deepEqual(f.state.sent, [[0, 255, 128], [1, 2]]);
  assert.equal(f.state.closes[0].code, 1000);
  assert.equal(f.state.socketCloses, 1);
});

test("Telegram closing TCP before any reply is named, not a silent 1000", async () => {
  // A tunnel that never carried a byte from Telegram used to end as a plain
  // 1000 close, which hid where it broke (#53). The reason also reaches the
  // Worker log in the Cloudflare dashboard.
  const f = await fixture();
  await f.fetch();
  f.message(new Uint8Array([1, 2, 3]).buffer);
  await tick();
  f.eof();
  for (let i = 0; i < 5; i++) await tick();
  assert.deepEqual(f.state.closes[0], { code: 1011, reason: "Telegram closed before replying" });
  assert.match(f.state.warnings[0], /149\.154\.167\.51 closed TCP before replying \(client sent 3 bytes\)/);
});

test("CDN worker keeps the requested DC203 destination", async () => {
  const f = await fixture();
  await f.fetch("/apiws?dst=91.105.192.100&dc=203");
  assert.equal(f.state.address.hostname, "91.105.192.100");
  assert.equal(f.state.address.port, 443);
  f.event("close");
});