From 17dea74357ee965b9fa4f0adf8814aa8b4d28f4d Mon Sep 17 00:00:00 2001 From: roblabla Date: Thu, 21 May 2015 20:35:03 +0000 Subject: [PATCH] Rewrite of the internal serializer using transform streams --- src/client.js | 211 ++++++++++++++++------------------ src/index.js | 8 +- src/protocol.js | 7 +- src/transforms/compression.js | 69 +++++++++++ src/transforms/framing.js | 47 ++++++++ src/transforms/serializer.js | 59 ++++++++++ 6 files changed, 279 insertions(+), 122 deletions(-) create mode 100644 src/transforms/compression.js create mode 100644 src/transforms/framing.js create mode 100644 src/transforms/serializer.js diff --git a/src/client.js b/src/client.js index 6f3730c..3220283 100644 --- a/src/client.js +++ b/src/client.js @@ -11,6 +11,10 @@ var EventEmitter = require('events').EventEmitter , packetNames = protocol.packetNames , states = protocol.states , debug = require('./debug') + , serializer = require('./transforms/serializer') + , compression = require('./transforms/compression') + , framing = require('./transforms/framing') + , crypto = require('crypto') ; module.exports = Client; @@ -18,23 +22,37 @@ module.exports = Client; function Client(isServer) { EventEmitter.call(this); + var socket; + + this.serializer = serializer.createSerializer({ isServer }); + this.compressor = null; + this.framer = framing.createFramer(); + this.cipher = null; + + this.decipher = null; + this.splitter = framing.createSplitter(); + this.decompressor = null; + this.deserializer = serializer.createDeserializer({ isServer }); + this._state = states.HANDSHAKING; Object.defineProperty(this, "state", { get: function() { - return this._state; + return this.serializer.protocolState; }, set: function(newProperty) { - var oldProperty = this._state; - this._state = newProperty; + var oldProperty = this.serializer.protocolState; + this.serializer.protocolState = newProperty; + this.deserializer.protocolState = newProperty; this.emit('state', newProperty, oldProperty); } }); + Object.defineProperty(this, "compressionThreshold", { + get: () => this.compressor == null ? -2 : this.compressor.compressionThreshold, + set: (threshold) => this.setCompressionThreshold(threshold) + }); + this.isServer = !!isServer; - this.socket = null; - this.encryptionEnabled = false; - this.cipher = null; - this.decipher = null; - this.compressionThreshold = -2; + this.packetsToParse = {}; this.on('newListener', function(event, listener) { var direction = this.isServer ? 'toServer' : 'toClient'; @@ -78,77 +96,46 @@ Client.prototype.onRaw = function(type, func) { }; Client.prototype.setSocket = function(socket) { - var self = this; - - function afterParse(err, parsed) { - if(err || (parsed && parsed.error)) { - self.emit('error', err || parsed.error); - self.end("ProtocolError"); - return; - } - if(!parsed) { - return; - } - var packet = parsed.results; - //incomingBuffer = incomingBuffer.slice(parsed.size); TODO: Already removed in prepare - - var packetName = protocol.packetNames[self.state][self.isServer ? 'toServer' : 'toClient'][packet.id]; - var packetState = self.state; - self.emit(packetName, packet); - self.emit('packet', packet); - self.emit('raw.' + packetName, parsed.buffer, packetState); - self.emit('raw', parsed.buffer, packetState); - prepareParse(); - } - - function prepareParse() { - var packetLengthField = protocol.types["varint"][0](incomingBuffer, 0); - if(packetLengthField && packetLengthField.size + packetLengthField.value <= incomingBuffer.length) { - var buf = incomingBuffer.slice(packetLengthField.size, packetLengthField.size + packetLengthField.value); - // TODO : Slice as early as possible to avoid processing same data twice. - incomingBuffer = incomingBuffer.slice(packetLengthField.size + packetLengthField.value); - if(self.compressionThreshold == -2) { - afterParse(null, parsePacketData(buf, self.state, self.isServer, self.packetsToParse)); - } else { - parseNewStylePacket(buf, self.state, self.isServer, self.packetsToParse, afterParse); - } - } - } - - self.socket = socket; - if(self.socket.setNoDelay) - self.socket.setNoDelay(true); - var incomingBuffer = new Buffer(0); - self.socket.on('data', function(data) { - if(self.encryptionEnabled) data = new Buffer(self.decipher.update(data), 'binary'); - incomingBuffer = Buffer.concat([incomingBuffer, data]); - prepareParse() - }); - - self.socket.on('connect', function() { - self.emit('connect'); - }); - - self.socket.on('error', onError); - self.socket.on('close', endSocket); - self.socket.on('end', endSocket); - self.socket.on('timeout', endSocket); - - function onError(err) { - self.emit('error', err); - endSocket(); - } - var ended = false; - function endSocket() { + // TODO : A lot of other things needs to be done. + var endSocket = () => { if(ended) return; ended = true; - self.socket.removeListener('close', endSocket); - self.socket.removeListener('end', endSocket); - self.socket.removeListener('timeout', endSocket); - self.emit('end', self._endReason); - } + this.socket.removeListener('close', endSocket); + this.socket.removeListener('end', endSocket); + this.socket.removeListener('timeout', endSocket); + this.emit('end', this._endReason); + }; + + var onError = (err) => { + this.emit('error', err); + endSocket(); + }; + + this.socket = socket; + + if(this.socket.setNoDelay) + this.socket.setNoDelay(true); + + this.socket.on('connect', () => this.emit('connect')); + + this.socket.on('error', onError); + this.socket.on('close', endSocket); + this.socket.on('end', endSocket); + this.socket.on('timeout', endSocket); + + this.socket.pipe(this.splitter).pipe(this.deserializer); + this.serializer.pipe(this.framer).pipe(this.socket); + + this.deserializer.on('data', (parsed) => { + var packet = parsed.results; + var packetName = protocol.packetNames[packet.state][this.isServer ? 'toServer' : 'toClient'][packet.id]; + this.emit('packet', packet); + this.emit(packetName, packet); + this.emit('raw.' + packetName, parsed.buffer, packet.state); + this.emit('raw', parsed.buffer, packet.state); + }); }; Client.prototype.end = function(reason) { @@ -156,52 +143,45 @@ Client.prototype.end = function(reason) { this.socket.end(); }; +Client.prototype.setEncryption = function(sharedSecret) { + if (this.cipher != null) + throw new Error("Set encryption twice !"); + this.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret); + this.framer.unpipe(this.socket); + this.framer.pipe(this.cipher).pipe(this.socket); + this.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret); + this.socket.unpipe(this.splitter); + this.socket.pipe(this.decipher).pipe(this.splitter); +} + +Client.prototype.setCompressionThreshold = function(threshold) { + if (this.compressor == null) { + this.compressor = compression.createCompressor(threshold); + this.serializer.unpipe(this.framer); + this.serializer.pipe(this.compressor).pipe(this.framer); + this.decompressor = compression.createDecompressor(threshold); + this.splitter.unpipe(this.deserializer); + this.splitter.pipe(this.decompressor).pipe(this.deserializer); + } else { + this.decompressor.threshold = threshold; + this.compressor.threshold = threshold; + } +} + function noop(err) { if(err) throw err; } -Client.prototype.write = function(packetId, params, cb) { - cb = cb || noop; - if(Array.isArray(packetId)) { - if(packetId[0] !== this.state) - return false; - packetId = packetId[1]; - } - if(typeof packetId === "string") - packetId = packetIds[this.state][this.isServer ? "toClient" : "toServer"][packetId]; - var that = this; - - var finishWriting = function(err, buffer) { - if(err) { - console.log(err); - throw err; // TODO : Handle errors gracefully, if possible - } - var packetName = packetNames[that.state][that.isServer ? "toClient" : "toServer"][packetId]; - debug("writing packetId " + that.state + "." + packetName + " (0x" + packetId.toString(16) + ")"); - debug(params); - var out = that.encryptionEnabled ? new Buffer(that.cipher.update(buffer), 'binary') : buffer; - that.socket.write(out,cb); - return true; - }; - - var buffer = createPacketBuffer(packetId, this.state, params, this.isServer); - if(this.compressionThreshold >= 0 && buffer.length >= this.compressionThreshold) { - debug("Compressing packet"); - compressPacketBuffer(buffer, finishWriting); - } else if(this.compressionThreshold >= -1) { - debug("New-styling packet"); - newStylePacket(buffer, 0, finishWriting); - } else { - debug("Old-styling packet"); - oldStylePacket(buffer, finishWriting); - } +Client.prototype.write = function(packetId, params, cb = noop) { + var packetName = protocol.packetNames[this.state][this.isServer ? "toClient" : "toServer"][packetId]; + debug("writing packetId " + this.state + "." + packetName + " (0x" + packetId.toString(16) + ")"); + debug(params); + this.serializer.write({ packetId, params }, cb); }; -// TODO : Perhaps this should only accept buffers without length, so we can -// handle compression ourself ? Needs to ask peopl who actually use this feature -// like @deathcap +// TODO : Write to the correct stream. This is currently broken. Client.prototype.writeRaw = function(buffer) { - var self = this; + /*var self = this; var finishWriting = function(error, buffer) { if(error) @@ -215,5 +195,6 @@ Client.prototype.writeRaw = function(buffer) { newStylePacket(buffer, 0, finishWriting); } else { oldStylePacket(buffer, finishWriting); - } + }*/ + throw new Error("Pending refactorisation"); }; diff --git a/src/index.js b/src/index.js index ef69bfe..c3aab91 100644 --- a/src/index.js +++ b/src/index.js @@ -183,11 +183,9 @@ function createServer(options) { client.end('DidNotEncryptVerifyTokenProperly'); return; } - client.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret); - client.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret); + client.setEncryption(sharedSecret); hash.update(sharedSecret); hash.update(client.publicKey); - client.encryptionEnabled = true; var isException = !!server.onlineModeExceptions[client.username.toLowerCase()]; var needToVerify = (onlineMode && !isException) || (!onlineMode && isException); @@ -365,16 +363,14 @@ function createClient(options) { var pubKey = mcPubKeyToURsa(packet.publicKey); var encryptedSharedSecretBuffer = pubKey.encrypt(sharedSecret, undefined, undefined, ursa.RSA_PKCS1_PADDING); var encryptedVerifyTokenBuffer = pubKey.encrypt(packet.verifyToken, undefined, undefined, ursa.RSA_PKCS1_PADDING); - client.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret); - client.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret); client.write(0x01, { sharedSecret: encryptedSharedSecretBuffer, verifyToken: encryptedVerifyTokenBuffer, },function(err){ if(err) throw err; - client.encryptionEnabled = true; }); + client.setEncryption(sharedSecret); } } } diff --git a/src/protocol.js b/src/protocol.js index f0ce3b2..52bcad8 100644 --- a/src/protocol.js +++ b/src/protocol.js @@ -161,7 +161,8 @@ function newStylePacket(buffer, dataSize, callback) { callback(null, packet); } -function parsePacketData(buffer, state, isServer, packetsToParse) { +// By default, parse every packets. +function parsePacketData(buffer, state, isServer, packetsToParse = {"packet": true}) { var cursor = 0; var packetIdField = utils.varint[0](buffer, cursor); var packetId = packetIdField.value; @@ -207,6 +208,10 @@ function parsePacketData(buffer, state, isServer, packetsToParse) { results: results }; }*/ + // TODO : investigate readResults returning null : shouldn't happen. + // When there is not enough data to read, we should return an error. + // As a general rule, it would be a good idea to introduce a whole bunch + // of new error classes to differenciate the errors. if(readResults === null || readResults.value == null) continue; if(readResults.error) { return readResults; diff --git a/src/transforms/compression.js b/src/transforms/compression.js new file mode 100644 index 0000000..6a1ded0 --- /dev/null +++ b/src/transforms/compression.js @@ -0,0 +1,69 @@ +var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint; +var zlib = require("zlib"); +var Transform = require("stream").Transform; + +module.exports.createCompressor = function(threshold) { + return new Compressor(threshold); +} + +module.exports.createDecompressor = function(threshold) { + return new Decompressor(threshold); +} + +class Compressor extends Transform { + constructor(compressionThreshold = -1) { + super(); + this.compressionThreshold = compressionThreshold; + } + + _transform(chunk, enc, cb) { + if (chunk.length >= this.compressionThreshold) + { + zlib.deflate(chunk, (err, newChunk) => { + if (err) + return cb(err); + var buf = new Buffer(sizeOfVarInt(chunk.length) + newChunk.length); + var offset = writeVarInt(chunk.length, buf, 0); + newChunk.copy(buf, offset); + this.push(buf); + return cb(); + }); + } + else + { + var buf = new Buffer(sizeOfVarInt(0) + chunk.length); + var offset = writeVarInt(0, buf, 0); + chunk.copy(buf, offset); + this.push(buf); + return cb(); + } + } +} + +class Decompressor extends Transform { + constructor(compressionThreshold = -1) { + super(); + this.compressionThreshold = compressionThreshold; + } + + _transform(chunk, enc, cb) { + var size, value, error; + ({ size, value, error } = readVarInt(chunk, 0)); + if (error) + return cb(error); + if (value === 0) + { + this.push(chunk.slice(size)); + return cb(); + } + else + { + zlib.inflate(chunk.slice(size), (err, newBuf) => { + if (err) + return cb(err); + this.push(newBuf); + return cb(); + }); + } + } +} diff --git a/src/transforms/framing.js b/src/transforms/framing.js new file mode 100644 index 0000000..b6607ad --- /dev/null +++ b/src/transforms/framing.js @@ -0,0 +1,47 @@ +var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint; +var Transform = require("stream").Transform; + +module.exports.createSplitter = function() { + return new Splitter(); +} + +module.exports.createFramer = function() { + return new Framer(); +} + +class Framer extends Transform { + constructor() { + super(); + } + + _transform(chunk, enc, cb) { + var buffer = new Buffer(sizeOfVarInt(chunk.length)); + writeVarInt(chunk.length, buffer, 0); + this.push(buffer); + this.push(chunk); + return cb(); + } +} + +class Splitter extends Transform { + constructor() { + super(); + this.buffer = new Buffer(0); + } + _transform(chunk, enc, cb) { + this.buffer = Buffer.concat([this.buffer, chunk]); + var value, size, error; + var offset = 0; + + ({ value, size, error } = readVarInt(this.buffer, offset) || { error: "Not enough data" }); + while (!error && this.buffer.length >= offset + size + value) + { + this.push(this.buffer.slice(offset + size, offset + size + value)); + offset += size + value; + ({ value, size, error } = readVarInt(this.buffer, offset) || { error: "Not enough data" }); + } + this.buffer = this.buffer.slice(offset); + return cb(); + } +} + diff --git a/src/transforms/serializer.js b/src/transforms/serializer.js new file mode 100644 index 0000000..99136c0 --- /dev/null +++ b/src/transforms/serializer.js @@ -0,0 +1,59 @@ +var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint; +var protocol = require("../protocol"); +var Transform = require("stream").Transform; + +module.exports.createSerializer = function(obj) { + return new Serializer(obj); +} + +module.exports.createDeserializer = function(obj) { + return new Deserializer(obj); +} + +class Serializer extends Transform { + constructor({ state = protocol.states.HANDSHAKING, isServer = false } = {}) { + super({ writableObjectMode: true }); + this.protocolState = state; + this.isServer = isServer; + } + + // TODO : Might make sense to make createPacketBuffer async. + _transform(chunk, enc, cb) { + try { + var buf = protocol.createPacketBuffer(chunk.packetId, this.protocolState, chunk.params, this.isServer); + this.push(buf); + return cb(); + } catch (e) { + return cb(e); + } + } +} + +class Deserializer extends Transform { + constructor({ state = protocol.states.HANDSHAKING, isServer = false } = {}) { + super({ readableObjectMode: true }); + this.protocolState = state; + this.isServer = isServer; + this.calls = 0; + } + + _transform(chunk, enc, cb) { + this.calls++; + var packet; + try { + packet = protocol.parsePacketData(chunk, this.protocolState, this.isServer, this.packetsToParse); + } catch (e) { + return cb(e); + } + if (packet.error) + { + packet.error.packet = packet; + return cb(packet.error) + } + else + { + this.push(packet); + return cb(); + } + } +}