Rcon client rewrite

rcon/index.js

@@ -1,90 +1,238 @@
-import EventEmiiter from 'events';
+import EventEmitter from 'events';
 import net from 'net';
+import util from 'util';
 
 import Logger from 'core/logger';
 
 const SERVERDATA_EXECCOMMAND = 0x02;
 const SERVERDATA_RESPONSE_VALUE = 0x00;
 const SERVERDATA_AUTH = 0x03;
-// const SERVERDATA_AUTH_RESPONSE = 0x02;
+const SERVERDATA_AUTH_RESPONSE = 0x02;
 const SERVERDATA_CHAT_VALUE = 0x01;
 
 const MID_PACKET_ID = 0x01;
 const END_PACKET_ID = 0x02;
 
-export default class Rcon extends EventEmiiter {
+export default class Rcon extends EventEmitter {
   constructor(options = {}) {
     super();
 
+    // store config
     for (const option of ['host', 'port', 'password'])
       if (!(option in options)) throw new Error(`${option} must be specified.`);
 
     this.host = options.host;
     this.port = options.port;
     this.password = options.password;
+    this.autoReconnectDelay = options.autoReconnectDelay || 5000;
 
-    this.reconnectInterval = null;
+    // bind methods
-    this.autoReconnectInterval = options.autoReconnectInterval || 5000;
+    this.connect = this.connect.bind(this); // we bind this as we call it on the auto reconnect timeout
+    this.onData = this.onData.bind(this);
+    this.onClose = this.onClose.bind(this);
+    this.onError = this.onError.bind(this);
 
+    // setup socket
+    this.client = new net.Socket();
+    this.client.on('data', this.onData);
+    this.client.on('close', this.onClose);
+    this.client.on('error', this.onError);
+
+    // constants
     this.maximumPacketSize = 4096;
 
-    this.client = null;
+    // internal variables
     this.connected = false;
-    this.autoReconnect = true;
+    this.autoReconnect = false;
+    this.autoReconnectTimeout = null;
 
-    this.requestQueue = [];
+    this.responseActionQueue = [];
-    this.currentMultiPacketResponse = [];
+    this.responsePacketQueue = [];
-    this.ignoreNextEndPacket = false;
+  }
 
-    this.onData = this.onData.bind(this);
+  onData(buf) {
+    Logger.verbose('RCON', 4, `Got data: ${this.bufToHexString(buf)}`);
+
+    const packets = this.splitPackets(buf);
+
+    for (const packet of packets) {
+      Logger.verbose('RCON', 4, `Processing packet: ${this.bufToHexString(packet)}`);
+
+      const decodedPacket = this.decodePacket(packet);
+      Logger.verbose(
+        'RCON',
+        3,
+        `Processing decoded packet: ${this.decodedPacketToString(decodedPacket)}`
+      );
+
+      if (decodedPacket.type === SERVERDATA_RESPONSE_VALUE) this.processResponsePacket(decodedPacket);
+      else if (decodedPacket.type === SERVERDATA_AUTH_RESPONSE) this.processAuthPacket(decodedPacket);
+      else if (decodedPacket.type === SERVERDATA_CHAT_VALUE) this.processChatPacket(decodedPacket);
+      else
+        Logger.verbose(
+          'RCON',
+          1,
+          `Unknown packet type ${decodedPacket.type} in: ${this.decodedPacketToString(
+            decodedPacket
+          )}`
+        );
+    }
+  }
+
+  splitPackets(buf) {
+    const packets = [];
+
+    let offset = 0;
+
+    while (offset < buf.byteLength) {
+      const size = buf.readInt32LE(offset);
+
+      const endOfPacket = offset + size + 4;
+
+      // The packet following an empty pocked will appear to be 10 long, it's not.
+      if(size === 10) {
+        // it's 21 bytes long (or 17 when ignoring the 4 size bytes), 7 bytes longer than it should be.
+        const probeEndOfPacket = endOfPacket + 7;
+
+        // check that there is room for the packet to be longer than it claims to be
+        if(probeEndOfPacket <= buf.byteLength) {
+          // it is, so probe that section of the buffer
+          const probeBuf = buf.slice(offset, probeEndOfPacket);
+
+          // we decode to see it's contents
+          const decodedProbePacket = this.decodePacket(probeBuf);
+
+          // if it matches this body then it's the broken length packet
+          if(decodedProbePacket.body === '\x00\x00\x00\x01\x00\x00\x00') {
+            // update the offset with the new correct length, then skip this packet as we don't care about it anyway
+            offset = endOfPacket + 7;
+            Logger.verbose('RCON', 4, `Ignoring some data: ${this.bufToHexString(probeBuf)}`);
+            continue;
+          }
+        }
+      }
+
+      const packet = buf.slice(offset, endOfPacket);
+
+      packets.push(packet);
+
+      offset = endOfPacket;
+    }
+
+    if (packets.length !== 0) {
+      Logger.verbose(
+        'RCON',
+        4,
+        `Split data into packets: ${packets.map(this.bufToHexString).join(', ')}`
+      );
+    }
+
+    return packets;
+  }
+
+  decodePacket(packet) {
+    return {
+      size: packet.readInt32LE(0),
+      id: packet.readInt32LE(4),
+      type: packet.readInt32LE(8),
+      body: packet.toString('utf8', 12, packet.byteLength - 2)
+    };
+  }
+
+  processResponsePacket(decodedPacket) {
+    if (decodedPacket.id === MID_PACKET_ID) {
+      Logger.verbose('RCON', 3, `Pushing packet to queue: ${this.decodedPacketToString(decodedPacket)}`);
+      this.responsePacketQueue.push(decodedPacket);
+    } else if (decodedPacket.id === END_PACKET_ID) {
+      Logger.verbose('RCON', 3, 'Initiating processing of packet queue.');
+      this.processCompleteResponse(this.responsePacketQueue);
+      this.responsePacketQueue = [];
+      this.ignoreNextResponsePacket = true;
+    } else {
+      Logger.verbose(
+        'RCON',
+        1,
+        `Unknown packet id ${decodedPacket.id} in: ${this.decodedPacketToString(decodedPacket)}`
+      );
+    }
+  }
+
+  processCompleteResponse(decodedPackets) {
+    Logger.verbose(
+      'RCON',
+      3,
+      `Processing complete decoded packet response: ${decodedPackets
+        .map(this.decodedPacketToString)
+        .join(', ')}`
+    );
+
+    const response = decodedPackets.map((packet) => packet.body).join();
+
+    this.responseActionQueue.shift()(response);
+  }
+
+  processAuthPacket(decodedPacket) {
+    this.responseActionQueue.shift()(decodedPacket);
+  }
+
+  processChatPacket(decodedPacket) {
+    const match = decodedPacket.body.match(
+      /\[(ChatAll|ChatTeam|ChatSquad|ChatAdmin)] \[SteamID:([0-9]{17})] (.+?) : (.*)/
+    );
+
+    this.emit('CHAT_MESSAGE', {
+      raw: decodedPacket.body,
+      chat: match[1],
+      steamID: match[2],
+      name: match[3],
+      message: match[4],
+      time: new Date()
+    });
+  }
+
+  onClose(hadError) {
+    this.connected = false;
+
+    Logger.verbose('RCON', 1, `Socket closed ${hadError ? 'without' : 'with'} an error.`);
+
+    if(this.autoReconnect) {
+      Logger.verbose('RCON', 1, `Sleeping ${this.autoReconnectDelay}ms before reconnecting.`);
+      setTimeout(this.connect, this.autoReconnectDelay);
+    }
+  }
+
+  onError(err) {
+    Logger.verbose('RCON', 1, `Socket had error:`, err);
+    this.emit('RCON_ERROR', err);
   }
 
   connect() {
-    Logger.verbose('RCON', 1, 'Method Exec: connect()');
     return new Promise((resolve, reject) => {
-      this.autoReconnect = true;
+      Logger.verbose('RCON', 1, `Connecting to: ${this.host}:${this.port}`);
-
-      // setup socket
-      this.client = new net.Socket();
-
-      this.client.on('data', this.onData);
-
-      this.client.on('error', (err) => {
-        Logger.verbose('RCON', 1, `Socket Error: ${err.message}`);
-        this.emit('RCON_ERROR', err);
-      });
-
-      this.client.on('close', async (hadError) => {
-        Logger.verbose('RCON', 1, `Socket Closed. AutoReconnect: ${this.autoReconnect}`);
-        this.connected = false;
-        this.client.removeListener('data', this.onData);
-        if (!this.autoReconnect) return;
-        if (this.reconnectInterval !== null) return;
-        this.reconnectInterval = setInterval(async () => {
-          Logger.verbose('RCON', 1, 'Attempting AutoReconnect.');
-          try {
-            await this.connect();
-            clearInterval(this.reconnectInterval);
-            this.reconnectInterval = null;
-            Logger.verbose('RCON', 1, 'Cleaned AutoReconnect.');
-          } catch (err) {
-            Logger.verbose('RCON', 1, 'AutoReconnect Failed.');
-          }
-        }, this.autoReconnectInterval);
-      });
 
       const onConnect = async () => {
-        Logger.verbose('RCON', 1, 'Socket Opened.');
         this.client.removeListener('error', onError);
         this.connected = true;
-        Logger.verbose('RCON', 1, 'Sending auth packet...');
+
-        await this.write(SERVERDATA_AUTH, this.password);
+        Logger.verbose('RCON', 1, `Connected to: ${this.host}:${this.port}`);
-        resolve();
+
+        try {
+          // connected successfully, now try auth...
+          await this.write(SERVERDATA_AUTH, this.password);
+
+          // connected and authed successfully
+          this.autoReconnect = true;
+          resolve();
+        } catch (err) {
+          reject(err);
+        }
       };
 
       const onError = (err) => {
-        Logger.verbose('RCON', 1, `Error Opening Socket: ${err.message}`);
         this.client.removeListener('connect', onConnect);
+
+        Logger.verbose('RCON', 1, `Failed to connect to: ${this.host}:${this.port}`, err);
+
         reject(err);
       };
 
@@ -95,199 +243,135 @@ export default class Rcon extends EventEmiiter {
     });
   }
 
-  async disconnect(disableAutoReconnect = true) {
+  disconnect() {
-    Logger.verbose('RCON', 1, `Method Exec: disconnect(${disableAutoReconnect})`);
     return new Promise((resolve, reject) => {
-      if (disableAutoReconnect) this.autoReconnect = false;
+      Logger.verbose('RCON', 1, `Disconnecting from: ${this.host}:${this.port}`);
 
       const onClose = () => {
-        Logger.verbose('RCON', 1, 'Disconnect successful.');
         this.client.removeListener('error', onError);
+
+        Logger.verbose('RCON', 1, `Disconnected from: ${this.host}:${this.port}`);
+
         resolve();
       };
 
       const onError = (err) => {
-        Logger.verbose('RCON', 1, `Error disconnecting: ${err.message}`);
         this.client.removeListener('close', onClose);
+
+        Logger.verbose('RCON', 1, `Failed to disconnect from: ${this.host}:${this.port}`, err);
+
         reject(err);
       };
 
       this.client.once('close', onClose);
       this.client.once('error', onError);
 
+      // prevent any auto reconnection happening
+      this.autoReconnect = false;
+      // clear the timeout just in case the socket closed and then we DCed
+      clearTimeout(this.autoReconnectTimeout);
+
       this.client.end();
     });
   }
 
-  decodePacket(buf) {
+  execute(command) {
-    return {
+    return this.write(SERVERDATA_EXECCOMMAND, command);
-      size: buf.readInt32LE(0),
-      id: buf.readInt32LE(4),
-      type: buf.readInt32LE(8),
-      body: buf.toString('utf8', 12, buf.byteLength - 2)
-    };
-  }
-
-  onData(inputBuf) {
-    let offset = 0;
-
-    while (offset < inputBuf.byteLength) {
-      const endOfPacket = offset + inputBuf.readInt32LE(offset) + 4;
-      const packetBuf = inputBuf.slice(offset, endOfPacket);
-      offset = endOfPacket;
-
-      const decodedPacket = this.decodePacket(packetBuf);
-
-      if (decodedPacket.type === SERVERDATA_CHAT_VALUE) {
-        // emit chat messages to own event
-        const message = decodedPacket.body.match(
-          /\[(ChatAll|ChatTeam|ChatSquad|ChatAdmin)] \[SteamID:([0-9]{17})] (.+?) : (.*)/
-        );
-
-        this.emit('CHAT_MESSAGE', {
-          raw: decodedPacket.body,
-          chat: message[1],
-          steamID: message[2],
-          name: message[3],
-          message: message[4],
-          time: new Date()
-        });
-      } else if (decodedPacket.id === END_PACKET_ID) {
-        if (this.ignoreNextEndPacket) {
-          this.ignoreNextEndPacket = false;
-          // boost the offset as the length seems wrong for this response
-          offset += 7;
-          continue;
-        }
-        this.ignoreNextEndPacket = true;
-
-        // at end of multipacket resolve request queue
-        const func = this.requestQueue.shift();
-        func();
-      } else {
-        // push packet to multipacket queue
-        this.currentMultiPacketResponse.push(decodedPacket);
-      }
-    }
-  }
-
-  encodePacket(type, id, body, encoding = 'utf8') {
-    const size = Buffer.byteLength(body) + 14;
-    const buffer = Buffer.alloc(size);
-
-    buffer.writeInt32LE(size - 4, 0);
-    buffer.writeInt32LE(id, 4);
-    buffer.writeInt32LE(type, 8);
-    buffer.write(body, 12, size - 2, encoding);
-    buffer.writeInt16LE(0, size - 2);
-
-    return buffer;
   }
 
   write(type, body) {
     return new Promise((resolve, reject) => {
-      if (!this.client.writable) {
-        reject(new Error('Unable to write to socket'));
-        return;
-      }
       if (!this.connected) {
         reject(new Error('Not connected.'));
         return;
       }
 
-      // prepare packets to send
+      if (!this.client.writable) {
-      const encodedPacket = this.encodePacket(type, MID_PACKET_ID, body);
+        reject(new Error('Unable to write to socket.'));
-      const encodedEmptyPacket = this.encodePacket(SERVERDATA_EXECCOMMAND, END_PACKET_ID, '');
+        return;
+      }
 
-      if (this.maximumPacketSize > 0 && encodedPacket.length > this.maximumPacketSize)
+      Logger.verbose('RCON', 2, `Writing packet with type "${type}" and body "${body}".`);
+
+      const encodedPacket = this.encodePacket(type, type === SERVERDATA_AUTH ? END_PACKET_ID : MID_PACKET_ID, body);
+      const encodedEmptyPacket = this.encodePacket(type, END_PACKET_ID, '');
+
+      if (this.maximumPacketSize < encodedPacket.length) {
         reject(new Error('Packet too long.'));
+        return;
+      }
 
-      // prepare to handle response.
+      let onResponse;
-      const handleAuthMultiPacket = async () => {
+      if (type === SERVERDATA_AUTH) {
-        this.client.removeListener('error', reject);
+        onResponse = (decodedPacket) => {
-
+          this.client.removeListener('error', onError);
-        for (const packet of this.currentMultiPacketResponse) {
+          if (decodedPacket.id === -1) {
-          if (packet.type === SERVERDATA_RESPONSE_VALUE) continue;
+            Logger.verbose('RCON', 1, 'Authentication failed.');
-          if (packet.id !== MID_PACKET_ID) {
+            reject(new Error('Authentication failed.'));
-            Logger.verbose('RCON', 1, 'Unable to authenticate.');
+          } else {
-            await this.disconnect(false);
+            Logger.verbose('RCON', 1, 'Authentication succeeded.');
-            reject(new Error('Unable to authenticate.'));
+            resolve();
           }
+        };
+      } else {
+        onResponse = (response) => {
+          this.client.removeListener('error', onError);
 
-          this.currentMultiPacketResponse = [];
+          Logger.verbose(
+            'RCON',
+            2,
+            `Processing complete response: ${response.replace(/\r\n|\r|\n/g, '\\n')}`
+          );
 
-          Logger.verbose('RCON', 1, 'Authenticated.');
+          resolve(response);
-          resolve();
+        };
-        }
+      }
+
+      const onError = (err) => {
+        Logger.verbose('RCON', 1, 'Error occurred. Wiping response action queue.', err);
+        this.responseActionQueue = [];
+        reject(err);
       };
 
-      const handleMultiPacket = () => {
+      // the auth packet also sends a normal response, so we add an extra empty action to ignore it
-        this.client.removeListener('error', reject);
+      if(type === SERVERDATA_AUTH) this.responseActionQueue.push(() => {});
 
-        let response = '';
+      this.responseActionQueue.push(onResponse);
-        for (const packet of this.currentMultiPacketResponse) {
-          response += packet.body;
-        }
 
-        this.currentMultiPacketResponse = [];
+      this.client.once('error', onError);
 
-        resolve(response);
+      Logger.verbose('RCON', 4, `Sending packet: ${this.bufToHexString(encodedPacket)}`);
-      };
-
-      if (type === SERVERDATA_AUTH) this.requestQueue.push(handleAuthMultiPacket);
-      else this.requestQueue.push(handleMultiPacket);
-
-      this.client.once('error', reject);
-
-      // send packets
       this.client.write(encodedPacket);
-      this.client.write(encodedEmptyPacket);
+
+      if (type !== SERVERDATA_AUTH) {
+        Logger.verbose(
+          'RCON',
+          4,
+          `Sending empty packet: ${this.bufToHexString(encodedEmptyPacket)}`
+        );
+        this.client.write(encodedEmptyPacket);
+      }
     });
   }
 
-  execute(command) {
+  encodePacket(type, id, body, encoding = 'utf8') {
-    Logger.verbose('RCON', 1, `Method Exec: execute(${command})`);
+    const size = Buffer.byteLength(body) + 14;
-    return this.write(SERVERDATA_EXECCOMMAND, command);
+    const buf = Buffer.alloc(size);
+
+    buf.writeInt32LE(size - 4, 0);
+    buf.writeInt32LE(id, 4);
+    buf.writeInt32LE(type, 8);
+    buf.write(body, 12, size - 2, encoding);
+    buf.writeInt16LE(0, size - 2);
+
+    return buf;
   }
 
-  async broadcast(message) {
+  bufToHexString(buf) {
-    await this.execute(`AdminBroadcast ${message}`);
+    return buf.toString('hex').match(/../g).join(' ');
   }
 
-  async getLayerInfo() {
+  decodedPacketToString(decodedPacket) {
-    const response = await this.execute('ShowNextMap');
+    return util.inspect(decodedPacket, { breakLength: Infinity });
-    const match = response.match(/^Current map is (.+), Next map is (.*)/);
-    return { currentLayer: match[1], nextLayer: match[2].length === 0 ? null : match[2] };
-  }
-
-  async getListPlayers() {
-    const response = await this.execute('ListPlayers');
-
-    const players = [];
-
-    for (const line of response.split('\n')) {
-      const match = line.match(
-        /ID: ([0-9]+) \| SteamID: ([0-9]{17}) \| Name: (.+) \| Team ID: ([0-9]+) \| Squad ID: ([0-9]+|N\/A)/
-      );
-      if (!match) continue;
-
-      players.push({
-        playerID: match[1],
-        steamID: match[2],
-        name: match[3],
-        teamID: match[4],
-        squadID: match[5] !== 'N/A' ? match[5] : null
-      });
-    }
-
-    return players;
-  }
-
-  async warn(steamID, message) {
-    await this.execute(`AdminWarn "${steamID}" ${message}`);
-  }
-
-  async switchTeam(steamID) {
-    await this.execute(`AdminForceTeamChange "${steamID}"`);
   }
 }

rcon/package.json

@@ -3,6 +3,7 @@
   "version": "1.0.0",
   "type": "module",
   "exports": {
-    ".": "./index.js"
+    ".": "./index.js",
+    "./squad": "./squad.js"
   }
 }

rcon/squad.js

@@ -0,0 +1,44 @@
+import Rcon from './index.js';
+
+export default class SquadRcon extends Rcon {
+  async broadcast(message) {
+    await this.execute(`AdminBroadcast ${message}`);
+  }
+
+  async getLayerInfo() {
+    const response = await this.execute('ShowNextMap');
+    const match = response.match(/^Current map is (.+), Next map is (.*)/);
+    return { currentLayer: match[1], nextLayer: match[2].length === 0 ? null : match[2] };
+  }
+
+  async getListPlayers() {
+    const response = await this.execute('ListPlayers');
+
+    const players = [];
+
+    for (const line of response.split('\n')) {
+      const match = line.match(
+        /ID: ([0-9]+) \| SteamID: ([0-9]{17}) \| Name: (.+) \| Team ID: ([0-9]+) \| Squad ID: ([0-9]+|N\/A)/
+      );
+      if (!match) continue;
+
+      players.push({
+        playerID: match[1],
+        steamID: match[2],
+        name: match[3],
+        teamID: match[4],
+        squadID: match[5] !== 'N/A' ? match[5] : null
+      });
+    }
+
+    return players;
+  }
+
+  async warn(steamID, message) {
+    await this.execute(`AdminWarn "${steamID}" ${message}`);
+  }
+
+  async switchTeam(steamID) {
+    await this.execute(`AdminForceTeamChange "${steamID}"`);
+  }
+}

squad-server/index.js

@@ -12,7 +12,7 @@ import Logger from 'core/logger';
 import { SQUADJS_API_DOMAIN } from 'core/constants';
 
 import LogParser from 'log-parser';
-import Rcon from 'rcon';
+import Rcon from 'rcon/squad';
 
 import { SQUADJS_VERSION } from './utils/constants.js';
 import { SquadLayers } from './utils/squad-layers.js';