Genesis

src/sniffer/MitmProxy.js

@@ -0,0 +1,256 @@
+const mc = require('minecraft-protocol');
+const { createLogger } = require('../utils/logger');
+const { PacketLog } = require('./PacketLog');
+const { relayPacket, sortLoginPending, relayToJava } = require('./mitmRelay');
+const { enableJavaEncryption } = require('./mitmEncryption');
+const { applyLoginStartIdentity } = require('./mitmLogin');
+const {
+  GATE,
+  canRelayC2S,
+  c2sForwardLabel,
+  hasPendingSuccess,
+  onJavaLoginAcknowledged,
+  onJavaFinishConfiguration,
+  partitionAfterCrypto,
+} = require('./mitmGate');
+const { createMitmSession, createSessionCleanup } = require('./mitmSession');
+const { startStatusPipe, startUpstream } = require('./mitmUpstream');
+
+const log = createLogger('Sniffer');
+const states = mc.states;
+
+/**
+ * MITM sniffer: Java ↔ node server ↔ upstream client ↔ real server.
+ * Each leg is decrypted by minecraft-protocol so packets can be logged by name.
+ */
+class MitmProxy {
+  constructor(config) {
+    this.config = config;
+    this.server = null;
+    this.activeSession = null;
+  }
+
+  start() {
+    const sniffer = this.config.sniffer;
+    this.server = mc.createServer({
+      host: sniffer.host || '0.0.0.0',
+      'online-mode': sniffer.onlineMode === true,
+      port: sniffer.port,
+      version: this.config.server.version,
+      maxPlayers: 1,
+      motd: '§eMITM Sniffer',
+      kickTimeout: 120000,
+      checkTimeoutInterval: 10000,
+      hideErrors: true,
+      errorHandler: (_client, err) => {
+        log.error('Client error:', err.message);
+      },
+    });
+
+    this.server.on('connection', (client) => this._onConnection(client));
+
+    this.server.on('listening', () => {
+      log.info(
+        `MITM sniffer on ${sniffer.host || '0.0.0.0'}:${sniffer.port} → ${this.config.server.host}:${this.config.server.port}`,
+      );
+      log.info(`Upstream auth: ${sniffer.upstreamAuth || 'microsoft'}`);
+      log.info(`Logs: ${sniffer.logDir}`);
+    });
+
+    this.server.on('error', (err) => {
+      log.error('Sniffer listen error:', err.message);
+    });
+  }
+
+  _onConnection(client) {
+    const addr = client.socket?.remoteAddress || '?';
+
+    if (this.activeSession) {
+      log.warn(`Rejecting ${addr} — session active`);
+      client.end('Sniffer allows one client at a time.');
+      return;
+    }
+
+    // Do not run local login — upstream is the real server session.
+    client.removeAllListeners('login_start');
+
+    const packetLog = new PacketLog({
+      logDir: this.config.sniffer.logDir,
+      sessionId: `session-${Date.now()}`,
+      clientUsername: 'unknown',
+      server: `${this.config.server.host}:${this.config.server.port}`,
+      version: this.config.server.version,
+      includePayload: this.config.sniffer.includePayload,
+    });
+
+    const session = createMitmSession(client, packetLog);
+    this.activeSession = session;
+
+    const cleanup = createSessionCleanup(session, packetLog, this);
+
+    client.on('end', () => {
+      log.info(`Client disconnected ${session.username} (${addr})`);
+      cleanup('client_end');
+    });
+    client.on('error', (err) => {
+      log.error(`Client error: ${err.message}`);
+      cleanup('client_error');
+    });
+
+    client.on('packet', (data, meta, buffer) => {
+      packetLog.logPacket('C2S', meta, data, buffer, {
+        forwarded: c2sForwardLabel(session, meta),
+        clientState: client.state,
+        upstreamState: session.upstream?.state,
+        gate: session.gate,
+      });
+
+      if (meta.state === states.HANDSHAKING && meta.name === 'set_protocol' && data.nextState === 1) {
+        startStatusPipe(session, this.config, packetLog, this);
+        return;
+      }
+
+      if (!session.upstream && meta.state === states.LOGIN && meta.name === 'login_start') {
+        try {
+          applyLoginStartIdentity(client, data, this.server, this.server.options);
+        } catch (err) {
+          log.error(`login_start rejected: ${err.message}`);
+          client.end('Invalid login');
+          return;
+        }
+        session.username = data.username;
+        packetLog.writeMeta({ type: 'username', username: data.username });
+        packetLog.writeMeta({ type: 'handshake_intent', mode: 'login' });
+        this._startUpstream(session, cleanup);
+        return;
+      }
+
+      if (meta.name === 'login_acknowledged') {
+        try {
+          if (onJavaLoginAcknowledged(session)) {
+            log.info(`Java login acknowledged → configuration for ${session.username}`);
+          }
+        } catch (err) {
+          log.error(`login_acknowledged error: ${err.message}`);
+        }
+        return;
+      }
+
+      if (meta.name === 'finish_configuration') {
+        try {
+          if (onJavaFinishConfiguration(session, packetLog)) {
+            log.info(`MITM bridge active (play) for ${session.username}`);
+          }
+        } catch (err) {
+          log.error(`finish_configuration error: ${err.message}`);
+        }
+        return;
+      }
+
+      if (session.upstream && canRelayC2S(session, meta)) {
+        try {
+          relayPacket(session.upstream, meta, data, buffer);
+        } catch (err) {
+          log.error(`C2S relay error (${meta.name}):`, err.message);
+        }
+      }
+    });
+
+    log.info(`Client connected ${addr} → ${packetLog.filePath}`);
+  }
+
+  _startUpstream(session, cleanup) {
+    const tryBegin = () => this._tryBeginJavaCrypto(session, cleanup);
+    startUpstream(session, this.config, cleanup, {
+      GATE_LOGIN: GATE.LOGIN,
+      onCompressBeforeCrypto: tryBegin,
+      onEncryptionBegin: tryBegin,
+      onSuccessWhileHeld: tryBegin,
+      onSuccessNoEncryption: (s) => {
+        s.gate = GATE.AWAIT_LOGIN_ACK;
+        log.info(`Login success sent (no upstream encryption) for ${s.username}`);
+      },
+    });
+  }
+
+  _tryBeginJavaCrypto(session, cleanup) {
+    if (!session.waitingJavaCrypto || session.javaCryptoStarting || session.gate !== GATE.LOGIN) return;
+
+    if (!hasPendingSuccess(session)) return;
+
+    this._doJavaCrypto(session, cleanup);
+  }
+
+  async _doJavaCrypto(session, cleanup) {
+    if (session.javaCryptoStarting || session.gate !== GATE.LOGIN) return;
+    session.javaCryptoStarting = true;
+
+    sortLoginPending(session.pendingS2C);
+    const heldLogin = [];
+    for (const item of session.pendingS2C) {
+      const { meta } = item;
+      if (meta.name === 'encryption_begin') continue;
+      if (meta.name === 'compress' && meta.state === states.LOGIN) {
+        session.relayedCompress = true;
+        try {
+          relayToJava(session.client, item.meta, item.data, item.buffer);
+        } catch (err) {
+          log.error(`S2C pre-crypto compress error:`, err.message);
+        }
+        continue;
+      }
+      heldLogin.push(item);
+    }
+    session.pendingS2C.length = 0;
+    const { login: afterCrypto, config: heldConfig } = partitionAfterCrypto(heldLogin);
+    session.pendingConfig.push(...heldConfig);
+
+    try {
+      await enableJavaEncryption(session.client, this.server, this.server.options);
+    } catch (err) {
+      session.javaCryptoStarting = false;
+      log.error('Java encryption setup failed:', err.message);
+      session.client.end('Sniffer encryption setup failed');
+      cleanup('encryption_error');
+      return;
+    }
+
+    session.holdS2C = false;
+    session.waitingJavaCrypto = false;
+    session.gate = GATE.AWAIT_LOGIN_ACK;
+    session.packetLog.writeMeta({ type: 'java_crypto_ready' });
+    log.info(`Java crypto ready for ${session.username}, awaiting login_acknowledged`);
+
+    const successPackets = [
+      ...afterCrypto.filter((p) => p.meta.name === 'success'),
+      ...session.pendingS2C.filter((p) => p.meta.name === 'success'),
+    ];
+    session.pendingS2C = session.pendingS2C.filter((p) => p.meta.name !== 'success');
+
+    for (const { data, meta, buffer } of successPackets) {
+      try {
+        relayToJava(session.client, meta, data, buffer);
+      } catch (err) {
+        log.error(`S2C success flush error:`, err.message);
+      }
+    }
+  }
+
+  stop() {
+    if (this.activeSession) {
+      const session = this.activeSession;
+      this.activeSession = null;
+      try { session.client.end('Sniffer shutting down'); } catch (_) {}
+      if (session.upstream && !session.upstream.ended) {
+        try { session.upstream.end('Sniffer shutting down'); } catch (_) {}
+      }
+      session.packetLog.close('shutdown');
+    }
+    if (this.server) {
+      this.server.close();
+      this.server = null;
+    }
+  }
+}
+
+module.exports = { MitmProxy };

src/sniffer/PacketLog.js

@@ -0,0 +1,250 @@
+const fs = require('fs');
+const path = require('path');
+
+/** JSONL lines longer than this go to sessionDir/line-NNNNNN.jsonl with a short ref in the main log. */
+const MAX_INLINE_LINE = 180;
+
+const LARGE_PACKETS = new Set([
+  'map_chunk',
+  'chunk_data',
+  'level_chunk_with_light',
+  'light_update',
+  'custom_payload',
+]);
+
+/**
+ * Append-only JSONL packet log for login / handoff analysis.
+ */
+class PacketLog {
+  /**
+   * @param {object} opts
+   * @param {string} opts.logDir
+   * @param {string} opts.sessionId
+   * @param {boolean} [opts.includePayload=true]
+   */
+  constructor(opts) {
+    this.includePayload = opts.includePayload !== false;
+    this.sessionId = opts.sessionId;
+    this._seq = 0;
+
+    const dir = path.resolve(opts.logDir);
+    fs.mkdirSync(dir, { recursive: true });
+    const file = path.join(dir, `${opts.sessionId}.jsonl`);
+    this._spillDir = path.join(dir, opts.sessionId);
+    fs.mkdirSync(this._spillDir, { recursive: true });
+    this._spillCount = 0;
+    this._stream = fs.createWriteStream(file, { flags: 'a' });
+    this._closed = false;
+    this.filePath = file;
+    this.spillDir = this._spillDir;
+
+    this.writeMeta({
+      type: 'session_start',
+      sessionId: opts.sessionId,
+      clientUsername: opts.clientUsername,
+      server: opts.server,
+      version: opts.version,
+    });
+  }
+
+  writeMeta(record) {
+    this._write({ ...record, t: new Date().toISOString() });
+  }
+
+  /**
+   * @param {'C2S'|'S2C'} dir
+   * @param {object} meta - minecraft-protocol packet meta
+   * @param {object} data - parsed params
+   * @param {Buffer} [rawBuffer]
+   * @param {object} [extra]
+   */
+  logUnparsed(dir, state, frame, message) {
+    this._write({
+      type: 'parse_error',
+      seq: ++this._seq,
+      t: new Date().toISOString(),
+      dir,
+      state,
+      frameBytes: frame.length,
+      headHex: frame.subarray(0, Math.min(16, frame.length)).toString('hex'),
+      error: message,
+      forwarded: 'tcp',
+    });
+  }
+
+  logOpaque(dir, bytes, extra = {}) {
+    if (extra.encrypted) {
+      this._encryptedOpaque = this._encryptedOpaque || { C2S: 0, S2C: 0, bytes: { C2S: 0, S2C: 0 } };
+      this._encryptedOpaque[dir]++;
+      this._encryptedOpaque.bytes[dir] += bytes;
+      const n = this._encryptedOpaque[dir];
+      if (n !== 1 && n !== 5 && n % 100 !== 0) return;
+      this._write({
+        type: 'opaque_summary',
+        seq: ++this._seq,
+        t: new Date().toISOString(),
+        dir,
+        encryptedChunks: n,
+        encryptedBytes: this._encryptedOpaque.bytes[dir],
+        forwarded: 'tcp',
+        note: 'Encrypted play traffic (map_chunk etc.) is forwarded but not decoded on a transparent pipe.',
+      });
+      return;
+    }
+    this._write({
+      type: 'opaque',
+      seq: ++this._seq,
+      t: new Date().toISOString(),
+      dir,
+      bytes,
+      forwarded: 'tcp',
+      ...extra,
+    });
+  }
+
+  logPacket(dir, meta, data, rawBuffer, extra = {}) {
+    const entry = {
+      type: 'packet',
+      seq: ++this._seq,
+      t: new Date().toISOString(),
+      dir,
+      state: meta.state,
+      name: meta.name,
+      clientState: extra.clientState,
+      upstreamState: extra.upstreamState,
+      forwarded: extra.forwarded ?? null,
+    };
+
+    if (rawBuffer) {
+      entry.rawBytes = rawBuffer.length;
+    }
+
+    if (this.includePayload && !LARGE_PACKETS.has(meta.name)) {
+      entry.data = summarizePacket(meta.name, data);
+    } else {
+      entry.summary = summarizeLargePacket(meta.name, data, rawBuffer);
+    }
+
+    this._write(entry);
+  }
+
+  close(reason) {
+    if (this._closed) return;
+    this.writeMeta({ type: 'session_end', reason: reason || 'closed' });
+    this._closed = true;
+    this._stream.end();
+  }
+
+  _write(obj) {
+    if (this._closed) return;
+    const line = JSON.stringify(obj);
+    if (line.length + 1 <= MAX_INLINE_LINE) {
+      this._writeRaw(line);
+      return;
+    }
+    const spillFile = `line-${String(++this._spillCount).padStart(6, '0')}.jsonl`;
+    fs.writeFileSync(path.join(this._spillDir, spillFile), `${line}\n`);
+    this._writeRaw(JSON.stringify(buildSpillRef(obj, spillFile)));
+  }
+
+  _writeRaw(line) {
+    const ok = this._stream.write(`${line}\n`);
+    if (!ok) {
+      this._stream.once('drain', () => {});
+    }
+  }
+}
+
+function buildPreview(obj) {
+  if (obj.type === 'packet') {
+    const parts = [obj.dir, obj.state, obj.name].filter(Boolean).join(' ');
+    const extra = obj.rawBytes != null ? ` ${obj.rawBytes}b` : '';
+    const fwd = obj.forwarded ? ` →${obj.forwarded}` : '';
+    return `${parts}${extra}${fwd}`.trim();
+  }
+  if (obj.type === 'session_start') {
+    return `${obj.clientUsername ?? '?'} → ${obj.server ?? '?'}`;
+  }
+  if (obj.summary && typeof obj.summary === 'object') {
+    const s = obj.summary;
+    if (s.id) return `${s.name ?? 'packet'} id=${s.id}`;
+    if (s.channel) return `channel=${s.channel}`;
+    return JSON.stringify(s);
+  }
+  if (obj.reason) return String(obj.reason);
+  if (obj.username) return String(obj.username);
+  const compact = JSON.stringify(obj.data ?? obj);
+  return compact.length > 48 ? `${compact.slice(0, 45)}…` : compact;
+}
+
+/** Compact pointer + preview kept in the main log (≤ MAX_INLINE_LINE). */
+function buildSpillRef(obj, spillFile) {
+  const ref = { _spill: spillFile, type: obj.type };
+  if (obj.seq != null) ref.seq = obj.seq;
+  ref.preview = buildPreview(obj);
+  let encoded = JSON.stringify(ref);
+  while (encoded.length > MAX_INLINE_LINE && ref.preview.length > 6) {
+    ref.preview = `${ref.preview.slice(0, ref.preview.length - 3)}…`;
+    encoded = JSON.stringify(ref);
+  }
+  return ref;
+}
+
+function summarizePacket(name, data) {
+  if (data == null) return data;
+  if (Buffer.isBuffer(data)) {
+    return { _type: 'Buffer', length: data.length };
+  }
+  if (typeof data !== 'object') return data;
+
+  try {
+    return JSON.parse(JSON.stringify(data, replacer));
+  } catch {
+    return { _type: 'Unserializable', name };
+  }
+}
+
+function replacer(_key, value) {
+  if (Buffer.isBuffer(value)) {
+    return { _type: 'Buffer', length: value.length };
+  }
+  if (typeof value === 'string' && value.length > 512) {
+    return `${value.slice(0, 512)}…(${value.length} chars)`;
+  }
+  if (Array.isArray(value) && value.length > 32) {
+    return { _type: 'Array', length: value.length, sample: value.slice(0, 3) };
+  }
+  return value;
+}
+
+function summarizeLargePacket(name, data, rawBuffer) {
+  const summary = { name };
+  if (rawBuffer) summary.rawBytes = rawBuffer.length;
+  if (!data || typeof data !== 'object') return summary;
+
+  if (name === 'map_chunk' || name === 'level_chunk_with_light') {
+    summary.x = data.x;
+    summary.z = data.z;
+    if (data.groundUp != null) summary.groundUp = data.groundUp;
+  } else if (name === 'registry_data') {
+    summary.id = data.id;
+    if (data.codec) summary.hasCodec = true;
+  } else if (name === 'player_info') {
+    summary.action = data.action;
+    summary.entryCount = data.data?.length ?? 0;
+  } else if (name === 'custom_payload') {
+    summary.channel = data.channel;
+    if (data.data) {
+      summary.dataLength = Buffer.isBuffer(data.data) ? data.data.length : String(data.data).length;
+    }
+  } else {
+    for (const [k, v] of Object.entries(data)) {
+      if (Buffer.isBuffer(v)) summary[k] = { bytes: v.length };
+      else if (typeof v === 'string' && v.length > 64) summary[k] = `${v.length} chars`;
+      else summary[k] = v;
+    }
+  }
+  return summary;
+}
+
+module.exports = { PacketLog };

src/sniffer/StreamTap.js

@@ -0,0 +1,164 @@
+const mc = require('minecraft-protocol');
+const { createSplitter } = require('minecraft-protocol/src/transforms/framing');
+const { createDeserializer } = require('minecraft-protocol/src/transforms/serializer');
+
+const states = mc.states;
+
+const S2C_CONFIGURATION_PACKETS = new Set([
+  'registry_data',
+  'feature_flags',
+  'tags',
+  'finish_configuration',
+  'custom_payload',
+  'reset_chat',
+  'code_of_conduct',
+  'server_data',
+]);
+
+/**
+ * Parse-only tap: frames are split and parsed for logs; bytes are not modified.
+ */
+class StreamTap {
+  constructor(dir, version, packetLog, hooks = {}) {
+    this.dir = dir;
+    this.version = version;
+    this.packetLog = packetLog;
+    this.hooks = hooks;
+    this.session = hooks.session || {
+      state: states.HANDSHAKING,
+      compressionThreshold: -1,
+      encrypted: false,
+    };
+
+    this.state = this.session.state;
+    this.stats = hooks.stats;
+
+    this.splitter = createSplitter();
+    this._parsers = new Map();
+    this.splitter.on('data', (frame) => this._onFrame(frame));
+  }
+
+  feed(chunk) {
+    if (this.session.encrypted) {
+      this.stats.encryptedBytes[this.dir] += chunk.length;
+      this.stats.encryptedChunks[this.dir] += 1;
+      this.packetLog.logOpaque(this.dir, chunk.length, { encrypted: true });
+      return;
+    }
+    this.stats.rawBytes[this.dir] += chunk.length;
+    this.splitter.write(chunk);
+  }
+
+  _syncState() {
+    this.state = this.session.state;
+  }
+
+  _parser() {
+    this._syncState();
+    const key = `${this.state}:${this.dir}`;
+    if (!this._parsers.has(key)) {
+      this._parsers.set(
+        key,
+        createDeserializer({
+          state: this.state,
+          isServer: this.dir === 'C2S',
+          version: this.version,
+          noErrorLogging: true,
+        }),
+      );
+    }
+    return this._parsers.get(key);
+  }
+
+  _parseFrame(frame) {
+    const payload =
+      this.session.compressionThreshold >= 0 ? this._decompressFrame(frame) : frame;
+    return this._parser().parsePacketBuffer(payload);
+  }
+
+  _decompressFrame(frame) {
+    const { readVarInt } = require('protodef').types.varint;
+    const zlib = require('zlib');
+    const { value, size } = readVarInt(frame, 0);
+    if (value === 0) return frame.slice(size);
+    return zlib.inflateSync(frame.slice(size), { finishFlush: 2 });
+  }
+
+  _onFrame(frame) {
+    if (this.session.encrypted) return;
+
+    this.stats.frames[this.dir] += 1;
+
+    let parsed;
+    try {
+      parsed = this._parseFrame(frame);
+    } catch (err) {
+      this.stats.parseErrors[this.dir] += 1;
+      this.packetLog.logUnparsed(this.dir, this.state, frame, err.message);
+      return;
+    }
+    if (!parsed) return;
+
+    const name = parsed.data.name;
+    const data = parsed.data.params;
+    this.stats.packets[this.dir] += 1;
+
+    this.packetLog.logPacket(this.dir, { state: this.state, name }, data, frame, {
+      forwarded: 'tcp',
+    });
+
+    if (name === 'login_start' && data.username) {
+      this.hooks.onUsername?.(data.username);
+    }
+    if (name === 'set_protocol' && data.serverHost) {
+      const mode = data.nextState === 1 ? 'status_ping' : data.nextState === 2 ? 'login' : `nextState_${data.nextState}`;
+      this.packetLog.writeMeta({
+        type: 'handshake_intent',
+        mode,
+        serverHost: data.serverHost,
+        serverPort: data.serverPort,
+        protocolVersion: data.protocolVersion,
+      });
+    }
+    if ((name === 'set_compression' || name === 'compress') && data.threshold != null) {
+      this.session.compressionThreshold = data.threshold;
+      this._parsers.clear();
+      this.packetLog.writeMeta({ type: 'compression', threshold: data.threshold, dir: this.dir });
+    }
+    // Encryption starts after the client sends its encryption response (C2S), not on S2C offer.
+    if (this.dir === 'C2S' && name === 'encryption_begin') {
+      this.session.encrypted = true;
+      this.packetLog.writeMeta({ type: 'encryption_started', dir: this.dir });
+    }
+
+    this._advanceState(name, data);
+  }
+
+  _setState(next) {
+    if (next === this.session.state) return;
+    this.session.state = next;
+    this.state = next;
+    this._parsers.clear();
+  }
+
+  _advanceState(name, data) {
+    if (this.dir === 'C2S') {
+      if (this.state === states.HANDSHAKING && name === 'set_protocol') {
+        this._setState(data.nextState === 1 ? states.STATUS : states.LOGIN);
+      } else if (this.state === states.LOGIN && name === 'login_acknowledged') {
+        this._setState(states.CONFIGURATION);
+      } else if (this.state === states.CONFIGURATION && name === 'finish_configuration') {
+        this._setState(states.PLAY);
+      }
+      return;
+    }
+
+    if (this.state === states.LOGIN && S2C_CONFIGURATION_PACKETS.has(name)) {
+      this._setState(states.CONFIGURATION);
+    } else if (this.state === states.CONFIGURATION && name === 'finish_configuration') {
+      this._setState(states.PLAY);
+    }
+  }
+}
+
+module.exports = { StreamTap };

src/sniffer/TransparentProxy.js

@@ -0,0 +1,176 @@
+const net = require('net');
+const mc = require('minecraft-protocol');
+const { createLogger } = require('../utils/logger');
+const { PacketLog } = require('./PacketLog');
+const { StreamTap } = require('./StreamTap');
+
+const log = createLogger('Sniffer');
+
+/**
+ * TCP-transparent proxy: bytes forwarded unchanged; tap parses for JSONL only.
+ */
+class TransparentProxy {
+  constructor(config) {
+    this.config = config;
+    this.server = null;
+    this.activeSession = null;
+  }
+
+  start() {
+    const sniffer = this.config.sniffer;
+    const targetHost = this.config.server.host;
+    const targetPort = this.config.server.port;
+
+    this.server = net.createServer((clientSocket) => {
+      this._onClientConnect(clientSocket, targetHost, targetPort, sniffer);
+    });
+
+    this.server.on('error', (err) => {
+      log.error('Sniffer listen error:', err.message);
+    });
+
+    this.server.listen(sniffer.port, sniffer.host || '0.0.0.0', () => {
+      log.info(
+        `TCP sniffer on ${sniffer.host || '0.0.0.0'}:${sniffer.port} → ${targetHost}:${targetPort} (${this.config.server.version})`,
+      );
+      log.info('Join the server in-game (not just refresh the server list)');
+      log.info(`Logs: ${sniffer.logDir}`);
+    });
+  }
+
+  _onClientConnect(clientSocket, targetHost, targetPort, sniffer) {
+    const addr = clientSocket.remoteAddress || '?';
+
+    if (this.activeSession) {
+      log.warn(`Rejecting connection from ${addr} — session already active`);
+      clientSocket.destroy();
+      return;
+    }
+
+    let clientUsername = 'unknown';
+    const packetLog = new PacketLog({
+      logDir: sniffer.logDir,
+      sessionId: `session-${Date.now()}`,
+      clientUsername,
+      server: `${targetHost}:${targetPort}`,
+      version: this.config.server.version,
+      includePayload: sniffer.includePayload,
+    });
+
+    const session = {
+      state: mc.states.HANDSHAKING,
+      compressionThreshold: -1,
+      encrypted: false,
+    };
+    const stats = {
+      rawBytes: { C2S: 0, S2C: 0 },
+      frames: { C2S: 0, S2C: 0 },
+      packets: { C2S: 0, S2C: 0 },
+      parseErrors: { C2S: 0, S2C: 0 },
+      encryptedBytes: { C2S: 0, S2C: 0 },
+      encryptedChunks: { C2S: 0, S2C: 0 },
+    };
+
+    const c2sTap = new StreamTap('C2S', this.config.server.version, packetLog, {
+      session,
+      stats,
+      onUsername: (name) => {
+        clientUsername = name;
+        packetLog.writeMeta({ type: 'username', username: name });
+      },
+    });
+    const s2cTap = new StreamTap('S2C', this.config.server.version, packetLog, { session, stats });
+
+    const upstreamSocket = net.connect({ host: targetHost, port: targetPort });
+    let upstreamReady = false;
+    const pendingToUpstream = [];
+
+    this.activeSession = { clientSocket, upstreamSocket, packetLog };
+
+    let cleaned = false;
+    const cleanup = (reason) => {
+      if (cleaned) return;
+      cleaned = true;
+      packetLog.writeMeta({
+        type: 'session_stats',
+        reason,
+        username: clientUsername,
+        protocolState: session.state,
+        encrypted: session.encrypted,
+        stats,
+      });
+      try { clientSocket.destroy(); } catch (_) {}
+      try { upstreamSocket.destroy(); } catch (_) {}
+      packetLog.close(reason);
+      this.activeSession = null;
+    };
+
+    const flushUpstream = () => {
+      for (const buf of pendingToUpstream) {
+        upstreamSocket.write(buf);
+      }
+      pendingToUpstream.length = 0;
+    };
+
+    clientSocket.on('data', (chunk) => {
+      if (upstreamReady && !upstreamSocket.destroyed) {
+        upstreamSocket.write(chunk);
+      } else {
+        pendingToUpstream.push(chunk);
+      }
+      c2sTap.feed(chunk);
+    });
+
+    upstreamSocket.on('data', (chunk) => {
+      if (!clientSocket.destroyed) clientSocket.write(chunk);
+      s2cTap.feed(chunk);
+    });
+
+    upstreamSocket.on('connect', () => {
+      upstreamReady = true;
+      flushUpstream();
+      log.info(`Upstream TCP connected (${addr} → ${targetHost}:${targetPort})`);
+      packetLog.writeMeta({ type: 'upstream_connect' });
+    });
+
+    upstreamSocket.on('error', (err) => {
+      log.error(`Upstream error: ${err.message}`);
+      cleanup('upstream_error');
+    });
+
+    clientSocket.on('error', (err) => {
+      log.error(`Client socket error: ${err.message}`);
+      cleanup('client_error');
+    });
+
+    upstreamSocket.on('close', () => {
+      log.info(`Session ended (upstream closed) ${clientUsername} (${addr})`);
+      cleanup('upstream_close');
+    });
+
+    clientSocket.on('close', () => {
+      if (!cleaned) {
+        log.info(`Session ended (client closed) ${clientUsername} (${addr})`);
+        cleanup('client_close');
+      }
+    });
+
+    log.info(`Client connected ${addr} — logging to ${packetLog.filePath}`);
+  }
+
+  stop() {
+    if (this.activeSession) {
+      const { clientSocket, upstreamSocket, packetLog } = this.activeSession;
+      try { clientSocket.destroy(); } catch (_) {}
+      try { upstreamSocket.destroy(); } catch (_) {}
+      packetLog.close('shutdown');
+      this.activeSession = null;
+    }
+    if (this.server) {
+      this.server.close();
+      this.server = null;
+    }
+  }
+}
+
+module.exports = { TransparentProxy };

src/sniffer/index.js

@@ -0,0 +1,47 @@
+const path = require('path');
+const { loadConfig } = require('../config');
+const { createLogger } = require('../utils/logger');
+const { MitmProxy } = require('./MitmProxy');
+
+const log = createLogger('SnifferMain');
+
+console.log(`
+\x1b[36m  Packet Sniffer Proxy\x1b[0m
+  MITM mode — decrypt both legs, log packet names
+`);
+
+let config;
+try {
+  config = loadConfig();
+  config.sniffer = Object.assign(
+    {
+      host: '0.0.0.0',
+      port: 25567,
+      onlineMode: false,
+      upstreamAuth: 'microsoft',
+      logDir: path.join(__dirname, '..', '..', 'logs', 'sniffer'),
+      includePayload: true,
+    },
+    config.sniffer,
+  );
+  log.info(`Upstream: ${config.server.host}:${config.server.port} (${config.server.version})`);
+  log.info(`Listen: ${config.sniffer.host || '0.0.0.0'}:${config.sniffer.port}`);
+  log.info(`Client online-mode: ${config.sniffer.onlineMode}`);
+  log.info(`Upstream auth: ${config.sniffer.upstreamAuth}`);
+  log.info(`Logs: ${path.resolve(config.sniffer.logDir)}`);
+} catch (err) {
+  log.error(err.message);
+  process.exit(1);
+}
+
+const proxy = new MitmProxy(config);
+proxy.start();
+
+function shutdown(signal) {
+  log.info(`${signal}, shutting down...`);
+  proxy.stop();
+  setTimeout(() => process.exit(0), 500);
+}
+
+process.on('SIGINT', () => shutdown('SIGINT'));
+process.on('SIGTERM', () => shutdown('SIGTERM'));

src/sniffer/mitmEncryption.js

@@ -0,0 +1,69 @@
+const crypto = require('crypto');
+const NodeRSA = require('node-rsa');
+
+/**
+ * Offer encryption_begin to the Java client using the sniffer's RSA key (MITM leg).
+ * @param {import('minecraft-protocol').Client} client - server-side peer (Java)
+ * @param {import('minecraft-protocol').Server} server
+ * @param {object} options - createServer options
+ * @returns {Promise<void>} resolves when Java leg encryption is active
+ */
+function enableJavaEncryption(client, server, options) {
+  const onlineMode = options['online-mode'] === true;
+
+  return new Promise((resolve, reject) => {
+    const serverId = onlineMode ? crypto.randomBytes(4).toString('hex') : '-';
+    client.verifyToken = crypto.randomBytes(4);
+    const publicKeyStrArr = server.serverKey.exportKey('pkcs8-public-pem').split('\n');
+    let publicKeyStr = '';
+    for (let i = 1; i < publicKeyStrArr.length - 1; i++) {
+      publicKeyStr += publicKeyStrArr[i];
+    }
+    client.publicKey = Buffer.from(publicKeyStr, 'base64');
+
+    client.once('encryption_begin', (packet) => {
+      try {
+        const sharedSecret = decryptSharedSecret(server, client, packet);
+        client.setEncryption(sharedSecret);
+        resolve();
+      } catch (err) {
+        reject(err);
+      }
+    });
+
+    client.write('encryption_begin', {
+      serverId,
+      publicKey: client.publicKey,
+      verifyToken: client.verifyToken,
+      shouldAuthenticate: onlineMode,
+    });
+  });
+}
+
+function decryptSharedSecret(server, client, packet) {
+  const keyRsa = new NodeRSA(server.serverKey.exportKey('pkcs1'), 'private', {
+    encryptionScheme: 'pkcs1',
+  });
+  keyRsa.setOptions({ environment: 'browser' });
+
+  if (packet.hasVerifyToken === false && packet.crypto) {
+    const { concat } = require('minecraft-protocol/src/transforms/binaryStream');
+    const signable = concat('buffer', client.verifyToken, 'i64', packet.crypto.salt);
+    if (
+      client.profileKeys &&
+      !crypto.verify('sha256WithRSAEncryption', signable, client.profileKeys.public, packet.crypto.messageSignature)
+    ) {
+      throw new Error('invalid_public_key_signature');
+    }
+  } else if (packet.verifyToken != null) {
+    const encryptedToken = packet.hasVerifyToken ? packet.crypto?.verifyToken : packet.verifyToken;
+    const decryptedToken = keyRsa.decrypt(encryptedToken);
+    if (!client.verifyToken.equals(decryptedToken)) {
+      throw new Error('invalid_verify_token');
+    }
+  }
+
+  return keyRsa.decrypt(packet.sharedSecret);
+}
+
+module.exports = { enableJavaEncryption };

src/sniffer/mitmGate.js

@@ -0,0 +1,198 @@
+const mc = require('minecraft-protocol');
+const { relayToJava } = require('./mitmRelay');
+
+const states = mc.states;
+
+/** Java client join phases after upstream login (1.20.2+ configuration). */
+const GATE = {
+  LOGIN: 'login',
+  AWAIT_LOGIN_ACK: 'await_login_ack',
+  CONFIGURATION: 'configuration',
+  PLAY: 'play',
+};
+
+const PREBRIDGE_C2S = new Set(['login_plugin_response', 'cookie_response']);
+
+/** Upstream mc client (keepAlive: true) auto-responds; relaying from Java duplicates and kicks. */
+const UPSTREAM_OWNED_C2S = new Set(['keep_alive']);
+
+/**
+ * Upstream is a separate mc client that already completed its own login/config.
+ * Only relay Java C2S once both legs are in play.
+ */
+function canRelayC2S(session, meta) {
+  if (UPSTREAM_OWNED_C2S.has(meta.name)) return false;
+  if (session.gate === GATE.PLAY) return true;
+  return PREBRIDGE_C2S.has(meta.name);
+}
+
+function c2sForwardLabel(session, meta) {
+  if (UPSTREAM_OWNED_C2S.has(meta.name)) return 'blocked';
+  if (canRelayC2S(session, meta)) return 'mitm';
+  return 'pending';
+}
+
+/** How to handle upstream S2C for logging and routing. */
+function classifyS2C(session, meta) {
+  if (session.gate === GATE.PLAY) return 'relay';
+  if (session.gate === GATE.CONFIGURATION && meta.state === states.CONFIGURATION) return 'relay';
+  if (shouldBufferS2C(session, meta)) return 'buffer';
+  return 'hold';
+}
+
+function shouldBufferS2C(session, meta) {
+  if (session.gate === GATE.PLAY) return false;
+  if (session.gate === GATE.CONFIGURATION) return meta.state === states.PLAY;
+  if (session.gate === GATE.AWAIT_LOGIN_ACK) return true;
+  return false;
+}
+
+function flushQueue(session, queue) {
+  for (const { data, meta, buffer } of queue) {
+    try {
+      relayToJava(session.client, meta, data, buffer);
+    } catch (err) {
+      throw new Error(`${meta.state}.${meta.name}: ${err.message}`);
+    }
+  }
+  queue.length = 0;
+}
+
+function flushPendingConfig(session) {
+  flushQueue(session, session.pendingConfig);
+}
+
+/** Join-critical play packets before terrain chunks. */
+const PLAY_JOIN_ORDER = {
+  login: 0,
+  custom_payload: 1,
+  server_data: 2,
+  difficulty: 3,
+  abilities: 4,
+  held_item_slot: 5,
+  recipe_book_settings: 6,
+  recipe_book_add: 7,
+  entity_status: 8,
+  declare_recipes: 9,
+  position: 10,
+  player_info: 11,
+  update_view_distance: 12,
+  simulation_distance: 13,
+  spawn_position: 14,
+  initialize_world_border: 15,
+  update_time: 16,
+  game_state_change: 17,
+  set_ticking_state: 18,
+  step_tick: 19,
+  window_items: 20,
+  set_slot: 21,
+  system_chat: 22,
+  declare_commands: 23,
+  update_health: 24,
+  experience: 25,
+};
+
+const PLAY_CHUNK_PACKETS = new Set([
+  'map_chunk',
+  'update_light',
+  'unload_chunk',
+  'chunk_batch_start',
+  'chunk_batch_finished',
+]);
+
+function sortPlayPending(pending) {
+  pending.sort((a, b) => {
+    const oa = PLAY_JOIN_ORDER[a.meta.name] ?? 100;
+    const ob = PLAY_JOIN_ORDER[b.meta.name] ?? 100;
+    return oa - ob;
+  });
+}
+
+function isStalePlayS2C(meta) {
+  return meta.name === 'keep_alive';
+}
+
+function flushPendingPlay(session) {
+  sortPlayPending(session.pendingPlay);
+  const join = [];
+  const world = [];
+  for (const item of session.pendingPlay) {
+    if (isStalePlayS2C(item.meta)) continue;
+    if (PLAY_CHUNK_PACKETS.has(item.meta.name)) world.push(item);
+    else join.push(item);
+  }
+  session.pendingPlay.length = 0;
+  flushQueue(session, join);
+  if (world.length) {
+    setImmediate(() => {
+      flushQueue(session, world);
+    });
+  }
+}
+
+function queueBufferedS2C(session, data, meta, buffer) {
+  if (isStalePlayS2C(meta)) return;
+  const item = { data, meta, buffer };
+  if (meta.state === states.PLAY) {
+    session.pendingPlay.push(item);
+  } else {
+    session.pendingConfig.push(item);
+  }
+}
+
+function hasPendingSuccess(session) {
+  return session.pendingS2C.some((p) => p.meta.name === 'success');
+}
+
+function onJavaLoginAcknowledged(session) {
+  if (session.gate !== GATE.AWAIT_LOGIN_ACK) return false;
+  session.client.state = states.CONFIGURATION;
+  session.gate = GATE.CONFIGURATION;
+  flushPendingConfig(session);
+  return true;
+}
+
+function onJavaFinishConfiguration(session, packetLog) {
+  if (session.gate !== GATE.CONFIGURATION) return false;
+  session.client.state = states.PLAY;
+  session.gate = GATE.PLAY;
+  session.bridged = true;
+  flushPendingPlay(session);
+  packetLog.writeMeta({ type: 'bridge_active' });
+  return true;
+}
+
+function queueHeldS2C(session, data, meta, buffer) {
+  const item = { data, meta, buffer };
+  if (meta.state === states.CONFIGURATION) {
+    session.pendingConfig.push(item);
+  } else {
+    session.pendingS2C.push(item);
+  }
+}
+
+function partitionAfterCrypto(pendingS2C) {
+  const login = [];
+  const config = [];
+  for (const item of pendingS2C) {
+    if (item.meta.state === states.CONFIGURATION) config.push(item);
+    else login.push(item);
+  }
+  return { login, config };
+}
+
+module.exports = {
+  GATE,
+  canRelayC2S,
+  c2sForwardLabel,
+  classifyS2C,
+  shouldBufferS2C,
+  hasPendingSuccess,
+  flushPendingConfig,
+  flushPendingPlay,
+  onJavaLoginAcknowledged,
+  onJavaFinishConfiguration,
+  queueHeldS2C,
+  queueBufferedS2C,
+  partitionAfterCrypto,
+};

src/sniffer/mitmLogin.js

@@ -0,0 +1,54 @@
+const crypto = require('crypto');
+const uuid = require('minecraft-protocol/src/datatypes/uuid');
+const { concat } = require('minecraft-protocol/src/transforms/binaryStream');
+const { mojangPublicKeyPem } = require('minecraft-protocol/src/server/constants');
+
+/**
+ * Apply login_start fields without completing local login (no success packet).
+ */
+function applyLoginStartIdentity(client, packet, _server, options) {
+  const mcData = require('minecraft-data')(client.version);
+  client.supportFeature = mcData.supportFeature;
+  client.username = packet.username;
+
+  if (packet.playerUUID) {
+    client.uuid = packet.playerUUID;
+  }
+
+  if (packet.signature && mcData.supportFeature('signatureEncryption')) {
+    if (packet.signature.timestamp < BigInt(Date.now())) {
+      throw new Error('expired_public_key');
+    }
+    const publicKey = crypto.createPublicKey({
+      key: packet.signature.publicKey,
+      format: 'der',
+      type: 'spki',
+    });
+    const signable = mcData.supportFeature('profileKeySignatureV2')
+      ? concat('UUID', packet.playerUUID, 'i64', packet.signature.timestamp, 'buffer', publicKey.export({ type: 'spki', format: 'der' }))
+      : Buffer.from(`${packet.signature.timestamp}${publicKeyToPem(packet.signature.publicKey)}`, 'utf8');
+
+    if (!crypto.verify('RSA-SHA1', signable, crypto.createPublicKey(mojangPublicKeyPem), packet.signature.signature)) {
+      throw new Error('invalid_public_key_signature');
+    }
+    client.profileKeys = { public: publicKey };
+  }
+
+  if (options['online-mode'] !== true && !client.uuid) {
+    client.uuid = uuid.nameToMcOfflineUUID(client.username);
+  }
+}
+
+function publicKeyToPem(mcPubKeyBuffer) {
+  let pem = '-----BEGIN RSA PUBLIC KEY-----\n';
+  let base64PubKey = mcPubKeyBuffer.toString('base64');
+  const maxLineLength = 64;
+  while (base64PubKey.length > 0) {
+    pem += `${base64PubKey.substring(0, maxLineLength)}\n`;
+    base64PubKey = base64PubKey.substring(maxLineLength);
+  }
+  pem += '-----END RSA PUBLIC KEY-----\n';
+  return pem;
+}
+
+module.exports = { applyLoginStartIdentity };

src/sniffer/mitmRelay.js

@@ -0,0 +1,108 @@
+const { RAW_FORWARD_PACKETS } = require('../constants/rawPackets');
+
+/** Configuration packets forwarded with writeRaw (byte-identical NBT). */
+const CONFIG_RAW_PACKETS = new Set([
+  'registry_data',
+  'feature_flags',
+  'tags',
+  'finish_configuration',
+  'custom_payload',
+  'reset_chat',
+  'code_of_conduct',
+  'server_data',
+]);
+
+const COMPRESS_PACKETS = new Set(['compress', 'set_compression']);
+
+function shouldWriteRaw(meta, buffer) {
+  if (!buffer || buffer.length === 0) return false;
+  if (meta.state === 'configuration' && CONFIG_RAW_PACKETS.has(meta.name)) return true;
+  if (meta.state === 'play' && RAW_FORWARD_PACKETS.has(meta.name)) return true;
+  return false;
+}
+
+/**
+ * @param {import('minecraft-protocol').Client} target
+ */
+function relayPacket(target, meta, data, buffer) {
+  if (shouldWriteRaw(meta, buffer)) {
+    target.writeRaw(buffer);
+    return 'raw';
+  }
+  // Large login success (skins) must stay byte-identical once compression is negotiated.
+  if (buffer?.length && meta.state === 'login' && meta.name === 'success') {
+    target.writeRaw(buffer);
+    return 'raw';
+  }
+  // Play-phase MITM: prefer wire-identical bytes (movement, chat, etc.).
+  if (buffer?.length && meta.state === 'play') {
+    target.writeRaw(buffer);
+    return 'raw';
+  }
+  target.write(meta.name, data, meta.state);
+  return 'parsed';
+}
+
+function syncCompression(target, name, data) {
+  if (!COMPRESS_PACKETS.has(name) || data.threshold == null) return;
+  target.compressionThreshold = data.threshold;
+}
+
+/** Login-phase S2C order the Java client expects (lower = earlier). */
+const LOGIN_FORWARD_ORDER = {
+  compress: 0,
+  encryption_begin: 1,
+  success: 2,
+  login_plugin_request: 3,
+  cookie_request: 4,
+  disconnect: 99,
+};
+
+function sortLoginPending(pending) {
+  pending.sort((a, b) => {
+    const oa = LOGIN_FORWARD_ORDER[a.meta.name] ?? 50;
+    const ob = LOGIN_FORWARD_ORDER[b.meta.name] ?? 50;
+    return oa - ob;
+  });
+}
+
+/**
+ * Login compress must hit the wire before the compressor is enabled; otherwise
+ * writeRaw adds a spurious 0-length prefix and the client fails to decode.
+ */
+function relayLoginCompressToJava(client, meta, data, buffer) {
+  if (client.cipher != null) {
+    return 'skipped_late_compress';
+  }
+  if (client.compressor != null) {
+    syncCompression(client, meta.name, data);
+    return relayPacket(client, meta, data, buffer);
+  }
+  if (buffer?.length) {
+    client.writeRaw(buffer);
+  } else {
+    client.write(meta.name, data, meta.state);
+  }
+  syncCompression(client, meta.name, data);
+  return 'login_compress';
+}
+
+/**
+ * Forward S2C to Java; skip late login compress after encryption is already on.
+ */
+function relayToJava(client, meta, data, buffer) {
+  if (meta.name === 'compress' && meta.state === 'login') {
+    return relayLoginCompressToJava(client, meta, data, buffer);
+  }
+  return relayPacket(client, meta, data, buffer);
+}
+
+module.exports = {
+  CONFIG_RAW_PACKETS,
+  shouldWriteRaw,
+  relayPacket,
+  syncCompression,
+  sortLoginPending,
+  relayLoginCompressToJava,
+  relayToJava,
+};

src/sniffer/mitmSession.js

@@ -0,0 +1,59 @@
+const { GATE } = require('./mitmGate');
+
+/**
+ * Build the per-connection MITM session state bag.
+ * @param {object} client - minecraft-protocol client
+ * @param {import('./PacketLog').PacketLog} packetLog
+ * @returns {object} session state
+ */
+function createMitmSession(client, packetLog) {
+  return {
+    client,
+    upstream: null,
+    bridged: false,
+    gate: GATE.LOGIN,
+    holdS2C: false,
+    pendingS2C: [],
+    pendingConfig: [],
+    pendingPlay: [],
+    waitingJavaCrypto: false,
+    javaCryptoStarting: false,
+    relayedCompress: false,
+    statusPipe: null,
+    packetLog,
+    username: 'unknown',
+    cleaned: false,
+  };
+}
+
+/**
+ * Build the cleanup closure for a MITM session.
+ * @param {object} session
+ * @param {import('./PacketLog').PacketLog} packetLog
+ * @param {{ activeSession: object|null }} proxy - MitmProxy instance (mutated on cleanup)
+ * @returns {function(string): void}
+ */
+function createSessionCleanup(session, packetLog, proxy) {
+  return (reason) => {
+    if (session.cleaned) return;
+    session.cleaned = true;
+
+    if (session.statusPipe) {
+      try { session.statusPipe.client.destroy(); } catch (_) {}
+      try { session.statusPipe.upstream.destroy(); } catch (_) {}
+    }
+    if (session.upstream && !session.upstream.ended) {
+      try { session.upstream.end(reason); } catch (_) {}
+    }
+    packetLog.writeMeta({
+      type: 'session_stats',
+      reason,
+      username: session.username,
+      bridged: session.bridged,
+    });
+    packetLog.close(reason);
+    if (proxy.activeSession === session) proxy.activeSession = null;
+  };
+}
+
+module.exports = { createMitmSession, createSessionCleanup };

src/sniffer/mitmUpstream.js

@@ -0,0 +1,155 @@
+const net = require('net');
+const mc = require('minecraft-protocol');
+const { createLogger } = require('../utils/logger');
+const { relayToJava, syncCompression } = require('./mitmRelay');
+const { classifyS2C, queueHeldS2C, queueBufferedS2C } = require('./mitmGate');
+
+const log = createLogger('Sniffer');
+const states = mc.states;
+
+/**
+ * Pipe a status/ping handshake directly at the TCP level (no decryption needed).
+ * @param {object} session
+ * @param {{ server: { host: string, port: number } }} config
+ * @param {import('./PacketLog').PacketLog} packetLog
+ * @param {{ activeSession: object|null }} proxy - MitmProxy instance
+ */
+function startStatusPipe(session, config, packetLog, proxy) {
+  const { host, port } = config.server;
+  const upstream = net.connect({ host, port });
+  session.statusPipe = { client: session.client.socket, upstream };
+
+  packetLog.writeMeta({ type: 'handshake_intent', mode: 'status_ping' });
+
+  session.client.socket.pipe(upstream);
+  upstream.pipe(session.client.socket);
+
+  upstream.on('connect', () => packetLog.writeMeta({ type: 'upstream_connect', mode: 'status_tcp' }));
+  let statusDone = false;
+  const endStatus = () => {
+    if (statusDone) return;
+    statusDone = true;
+    if (proxy.activeSession === session) proxy.activeSession = null;
+    packetLog.close('status_done');
+  };
+  upstream.on('close', endStatus);
+  session.client.socket.on('close', endStatus);
+}
+
+/**
+ * Create the upstream mc.createClient and wire the S2C packet relay.
+ * @param {object} session
+ * @param {{ server: { host: string, port: number, version: string }, sniffer: { upstreamAuth?: string } }} config
+ * @param {function(string): void} cleanup
+ * @param {object} callbacks - { onCompressBeforeCrypto, onEncryptionBegin, onSuccessNoEncryption }
+ */
+function startUpstream(session, config, cleanup, callbacks) {
+  const { host, port, version } = config.server;
+  const auth = config.sniffer.upstreamAuth || 'microsoft';
+
+  const upstream = mc.createClient({
+    host,
+    port,
+    username: session.username,
+    version,
+    auth,
+    hideErrors: true,
+    keepAlive: true,
+    checkTimeoutInterval: 60000,
+  });
+  session.upstream = upstream;
+
+  upstream.on('connect', () => {
+    log.info(`Upstream connected for ${session.username}`);
+    session.packetLog.writeMeta({ type: 'upstream_connect' });
+  });
+
+  upstream.on('packet', (data, meta, buffer) => {
+    const s2cAction = classifyS2C(session, meta);
+    session.packetLog.logPacket('S2C', meta, data, buffer, {
+      forwarded: s2cAction,
+      clientState: session.client.state,
+      upstreamState: upstream.state,
+      gate: session.gate,
+    });
+    syncCompression(upstream, meta.name, data);
+
+    if (s2cAction === 'relay') {
+      try {
+        relayToJava(session.client, meta, data, buffer);
+      } catch (err) {
+        log.error(`S2C relay error (${meta.name}):`, err.message);
+      }
+      return;
+    }
+
+    if (s2cAction === 'buffer') {
+      queueBufferedS2C(session, data, meta, buffer);
+      return;
+    }
+
+    if (session.gate !== callbacks.GATE_LOGIN) {
+      return;
+    }
+
+    if (meta.name === 'compress' && meta.state === states.LOGIN) {
+      session.relayedCompress = true;
+      try {
+        relayToJava(session.client, meta, data, buffer);
+      } catch (err) {
+        log.error(`S2C compress error:`, err.message);
+      }
+      callbacks.onCompressBeforeCrypto(session);
+      return;
+    }
+
+    if (meta.name === 'encryption_begin') {
+      session.holdS2C = true;
+      session.waitingJavaCrypto = true;
+      callbacks.onEncryptionBegin(session);
+      return;
+    }
+
+    if (session.holdS2C) {
+      queueHeldS2C(session, data, meta, buffer);
+      if (meta.name === 'success') {
+        callbacks.onSuccessWhileHeld(session);
+      }
+      return;
+    }
+
+    if (meta.name === 'success') {
+      try {
+        relayToJava(session.client, meta, data, buffer);
+        callbacks.onSuccessNoEncryption(session);
+      } catch (err) {
+        log.error(`S2C success error:`, err.message);
+      }
+      return;
+    }
+
+    try {
+      relayToJava(session.client, meta, data, buffer);
+    } catch (err) {
+      log.error(`S2C relay error (${meta.name}):`, err.message);
+    }
+  });
+
+  upstream.on('end', () => {
+    log.info(`Upstream closed for ${session.username}`);
+    if (!session.cleaned && !session.client.ended) {
+      try { session.client.end('Upstream disconnected'); } catch (_) {}
+    }
+    cleanup('upstream_end');
+  });
+
+  upstream.on('error', (err) => {
+    log.error(`Upstream error: ${err.message}`);
+    if (!session.cleaned && !session.client.ended) {
+      try { session.client.end(err.message); } catch (_) {}
+    }
+    cleanup('upstream_error');
+  });
+}
+
+module.exports = { startStatusPipe, startUpstream };