A simple Gnutella client for educational purposes. It does not transfer files, just connects to the network and keeps stats.

simple-peer.ts

import net from "net";
import crypto from "crypto";
import { argv, exit } from "process";

interface CLIOpts {
  port: number;
  seeds: { host: string; port: number }[];
}

type GUID = Buffer;

enum Descriptor {
  Ping = 0x00,
  Pong = 0x01,
  Push = 0x40,
  Query = 0x80,
  QueryHit = 0x81,
}

interface ParsedMessage {
  guid: GUID;
  descriptor: Descriptor;
  ttl: number;
  hops: number;
  payload: Buffer;
}

const USER_AGENT = "BunUltra/0.1";
const PING_INTERVAL = 30_000;
const PEER_TIMEOUT = 300_000;
const ADVERTISED_DEGREE = 32;
const VENDOR_ID = "BUN1";
const HANDSHAKE_END = Buffer.from("\r\n\r\n", "utf8");
const OK_HEADER = [
  "GNUTELLA/0.6 200 OK",
  "X-Ultrapeer: True",
  "Vendor-Message: 0.1",
  `X-Vendor-ID: ${VENDOR_ID}`,
  "\r\n",
].join("\r\n");

function parseCLI(): CLIOpts {
  const opts: CLIOpts = { port: 6346, seeds: [] };
  for (let i = 2; i < argv.length; i++) {
    const arg = argv[i];
    switch (arg) {
      case "--port":
      case "-p":
        opts.port = Number(argv[++i]);
        break;
      case "--seed":
      case "-s":
        opts.seeds = argv[++i].split(",").map((s) => {
          const [h, p = "6346"] = s.split(":");
          return { host: h, port: Number(p) };
        });
        break;
      case "--help":
      case "-h":
        console.info(
          "Usage: bun ultrapeer.ts [--port 6346] [--seed h:p[,h:p]]",
        );
        exit(0);
      default:
        console.error(`Unknown argument: ${arg}`);
        console.info(
          "Usage: bun ultrapeer.ts [--port 6346] [--seed h:p[,h:p]]",
        );
        exit(1);
    }
  }
  if (!opts.seeds.length)
    console.warn(
      "[boot] No --seed supplied; waiting for inbound connections.",
    );
  return opts;
}

const { port: LISTEN_PORT, seeds: SEED_HOSTS } = parseCLI();

const newGuid = () => crypto.randomBytes(16);

function buildHeader(
  guid: GUID,
  d: Descriptor,
  ttl: number,
  hops: number,
  len: number,
): Buffer {
  const lenBuf = Buffer.alloc(4);
  lenBuf.writeUInt32LE(len, 0);
  return Buffer.concat([guid, Buffer.from([d, ttl, hops]), lenBuf]);
}

const buildPing = (ttl = 2) =>
  buildHeader(newGuid(), Descriptor.Ping, ttl, 0, 0);

function buildPong(reqGuid: GUID, socket: net.Socket, ttl = 1) {
  const payload = Buffer.alloc(14);
  payload.writeUInt16LE(socket.localPort ?? LISTEN_PORT, 0);
  (socket.localAddress ?? "127.0.0.1").split(".").forEach((n, i) => {
    payload[2 + i] = Number(n);
  });
  const header = buildHeader(
    reqGuid,
    Descriptor.Pong,
    ttl,
    0,
    payload.length,
  );
  return Buffer.concat([header, payload]);
}

function parseMessage(buf: Buffer): ParsedMessage | null {
  if (buf.length < 23) return null;
  const guid = buf.subarray(0, 16);
  const descriptor = buf[16] as Descriptor;
  const ttl = buf[17];
  const hops = buf[18];
  const len = buf.readUInt32LE(19);
  if (buf.length < 23 + len) return null;
  return {
    guid,
    descriptor,
    ttl,
    hops,
    payload: buf.subarray(23, 23 + len),
  };
}

function buildConnectHeader(port: number): string {
  return [
    "GNUTELLA CONNECT/0.6",
    `User-Agent: ${USER_AGENT}`,
    `Listen-IP: 0.0.0.0:${port}`,
    "X-Ultrapeer: True",
    "X-Query-Routing: 0.2",
    `X-Degree: ${ADVERTISED_DEGREE}`,
    "Accept-Encoding: deflate,gzip",
    "\r\n",
  ].join("\r\n");
}

class Peer {
  private lastHeard = Date.now();
  private handshakeDone = false;

  constructor(private socket: net.Socket, outgoing: boolean) {
    if (outgoing)
      socket.once("connect", () =>
        this.socket.write(buildConnectHeader(LISTEN_PORT)),
      );
    this.attachEvents();
  }

  sendPing() {
    if (this.handshakeDone) this.socket.write(buildPing());
  }

  isTimedOut() {
    return Date.now() - this.lastHeard > PEER_TIMEOUT;
  }

  close() {
    this.socket.end();
  }

  private attachEvents() {
    let buf = Buffer.alloc(0);
    this.socket.on("data", (data) => {
      console.log(data.toString("utf8"));
      this.lastHeard = Date.now();
      buf = Buffer.concat([buf, data]);
      if (!this.handshakeDone) {
        const idx = buf.indexOf(HANDSHAKE_END);
        if (idx === -1) return;
        this.handshakeDone = true;
        const txt = buf
          .subarray(0, idx + HANDSHAKE_END.length)
          .toString("utf8");
        if (txt.startsWith("GNUTELLA CONNECT/0.6"))
          this.socket.write(OK_HEADER);
        buf = buf.subarray(idx + HANDSHAKE_END.length);
      }
      while (true) {
        const msg = parseMessage(buf);
        if (!msg) break;
        if (msg.descriptor === Descriptor.Ping)
          this.socket.write(buildPong(msg.guid, this.socket));
        buf = buf.subarray(23 + msg.payload.length);
      }
    });
    this.socket.on("error", () => peers.delete(this));
    this.socket.on("close", () => peers.delete(this));
  }
}

const peers = new Set<Peer>();
const addPeer = (s: net.Socket, outgoing: boolean) =>
  peers.add(new Peer(s, outgoing));

net
  .createServer((s) => addPeer(s, false))
  .listen(LISTEN_PORT, () =>
    console.info(`[boot] Listening on 127.0.0.1:${LISTEN_PORT}`),
  );

SEED_HOSTS.forEach(({ host, port }) =>
  addPeer(
    net
      .connect(port, host)
      .on("error", (e) => console.error(`[seed] ${e.message}`)),
    true,
  ),
);

setInterval(() => {
  peers.forEach((p) =>
    p.isTimedOut() ? (p.close(), peers.delete(p)) : p.sendPing(),
  );
  console.info(`[watchdog] Active peers: ${peers.size}`);
}, PING_INTERVAL);

process.on("SIGINT", () => {
  peers.forEach((p) => p.close());
  process.exit(0);
});