cm_client.js

var Steam = require('../index.js');
var ByteBuffer = require('bytebuffer');
var SteamCrypto = require('@doctormckay/steam-crypto');
var ProxyAgent = require('./proxy-agent.js');
var BufferCRC32 = require('buffer-crc32');
var Zlib = require('zlib');
var HTTPS = require('https');
var Async = require('async');

var TCPConnection = require('./tcp_connection.js');
var UDPConnection = require('./udp_connection.js');
var WebSocketConnection = require('./websocket_connection.js');

var g_WebSocketServers = null;
var g_WebSocketLastPing = 0;
var g_WebSocketPingedServers = null;
var g_WebSocketBootstrapping = false;

var Schema = Steam.Internal;

var EMsg = Steam.EMsg;

const PROTO_MASK = 0x80000000;
const PROTOCOL_VERSION = 65579;

require('util').inherits(CMClient, require('events').EventEmitter);

/**
 * Create a new Steam Client
 * @param {EConnectionProtocol} [protocol=TCP] - The protocol with which we want to connect
 * @augments EventEmitter
 * @constructor
 */
function CMClient(protocol) {
	this.protocol = protocol || Steam.EConnectionProtocol.TCP;
	this.remoteAddress = null;
}

/**
 * Change the local IP/port that will be used to connect (takes effect on next connection)
 * @param {string} [localAddress] - The local IP address (in string format) that will be used to connect
 * @param {int} [localPort] - The local port that will be used to connect
 */
CMClient.prototype.bind = function(localAddress, localPort) {
	if (typeof localAddress === 'number') {
		localPort = localAddress;
		localAddress = undefined;
	}

	this._localAddress = localAddress;
	this._localPort = localPort;
};

/**
 * Set this connection to use an HTTP proxy. Only works with TCP connections.
 * @param {string|null} proxyUrl - The URL of the proxy, e.g. http://user:pass@1.2.3.4; null to disable. Takes effect on next connection.
 */
CMClient.prototype.setHttpProxy = function(proxyUrl) {
	this._httpProxy = proxyUrl;
};


// Methods

try {
	Steam.servers = require('../resources/servers.json'); // try to load the one that's generated by the prepublish script
} catch (e) {
	// It's not there, fallback to the bootstrap one
	Steam.servers = require('../resources/servers_bootstrap.json');
}

/**
 * Connect to Steam.
 * @param {object} [server] - The CM server to which we will connect. If omitted, chosen randomly.
 * @param {string} [server.host] - The IP address or hostname of the server to which we would like to connect
 * @param {int} [server.port] - The port of the server to which we would like to connect
 * @param {boolean} [autoRetry=true] - Should we automatically attempt to reconnect if we can't establish a connection?
 */
CMClient.prototype.connect = function(server, autoRetry) {
	if (typeof server === 'boolean') {
		autoRetry = server;
		server = null;
	}

	delete this._serverLoad;

	this.disconnect();

	this._jobs = {};
	this._currentJobID = 0;

	this._sessionID = 0;

	this._server = server;
	this._autoRetry = typeof autoRetry === 'boolean' ? autoRetry : true;

	this._getServer(server, (server) => {
		server = server || Steam.servers[Math.floor(Math.random() * Steam.servers.length)];
		this.emit('debug', 'connecting to ' + server.host + ':' + server.port);

		switch (this.protocol) {
			case Steam.EConnectionProtocol.TCP:
				this._connection = new TCPConnection();
				break;

			case Steam.EConnectionProtocol.UDP:
				this._connection = new UDPConnection();
				break;

			case Steam.EConnectionProtocol.WebSocket:
				this._connection = new WebSocketConnection();
				break;

			default:
				throw new Error("Unknown connection protocol");
		}

		var self = this;
		this._connection.connect({
			"port": server.port,
			"host": server.host,
			"localAddress": this._localAddress,
			"localPort": this._localPort,
			"httpProxy": this._httpProxy
		}, function(err) {
			// This callback isn't necessarily called when we're connected, but instead when
			// the connection is being established (e.g. we've connected to the proxy if applicable).
			// The err is a fatal connection error (e.g. proxy error).

			if (err) {
				if (err.message == "Proxy connection timed out") {
					self.emit('debug', 'proxy timed out');
					self._disconnected();
				} else {
					err.eresult = Steam.EResult.ConnectFailed;
					self.emit('error', err);
				}

				return;
			}

			self._connection.on('debug', function(msg) {
				self.emit('debug-connection', msg);
			});

			self._connection.on('packet', self._netMsgReceived.bind(self));
			self._connection.on('close', self._disconnected.bind(self));

			self._connection.on('error', function(err) {
				// it's ok, we'll reconnect after 'close'
				self.emit('debug', 'socket error: ' + err);
			});

			self._connection.on('encryptionError', function(err) {
				self.emit('error', err);
				self.disconnect();
			});

			self._connection.on('connect', function(serverLoad) {
				self.emit('debug', 'connected' + (serverLoad && typeof serverLoad !== 'object' ? ', server load ' + serverLoad : ''));
				self._serverLoad = serverLoad;
				delete self._timeout;

				if (self.protocol == Steam.EConnectionProtocol.WebSocket) {
					self._connection.setTimeout(0);
					self.connected = true;
					self.emit('connected');
				}
			});

			self._connection.on('end', function() {
				self.emit('debug', 'socket ended');
			});

			self._connection.setTimeout(1000, function() {
				self.emit('debug', 'socket timed out');
				self._connection.destroy();
			});
		});

		this.remoteAddress = server.host + ":" + server.port;
	});
};

/**
 * Break our current connection without logging off. If we're connecting, cancel the connection.
 * If not connected or connecting, do nothing.
 */
CMClient.prototype.disconnect = function() {
	this.remoteAddress = null;

	if (this._connection) {
		this._connection.end();
		this._connection.removeAllListeners();
		this._connection.on('error', function() { }); // just to prevent crashes
		delete this._connection;

		if (this.loggedOn) {
			this.loggedOn = false;
			clearInterval(this._heartBeatFunc);
		}

		this.connected = false;
	} else if (this._scheduledConnection) {
		// there was an error and we're currently waiting
		clearTimeout(this._scheduledConnection);
		delete this._scheduledConnection;
	}
};

CMClient.prototype._send = function(header, body, callback) {
	if (callback) {
		var sourceJobID = ++this._currentJobID;
		this._jobs[sourceJobID] = callback;
	}

	if (header.msg == EMsg.ClientLogon) {
		// Horrible hack to overwrite the protocol version
		var msg = Schema.CMsgClientLogon.decode(body);
		msg.protocol_version = PROTOCOL_VERSION;
		body = new Schema.CMsgClientLogon(msg).toBuffer();
	}

	if (header.msg == EMsg.ChannelEncryptResponse) {
		header.sourceJobID = sourceJobID;
		header = new Schema.MsgHdr(header);

	} else if (header.proto) {
		header.proto.client_sessionid = this._sessionID;
		header.proto.steamid = this.steamID;
		header.proto.jobid_source = sourceJobID;
		header = new Schema.MsgHdrProtoBuf(header);

	} else {
		header.steamID = this.steamID;
		header.sessionID = this._sessionID;
		header.sourceJobID = sourceJobID;
		header = new Schema.ExtendedClientMsgHdr(header);
	}

	this._connection.send(Buffer.concat([header.toBuffer(), body]));
};

/**
 * Send a logon message to Steam. You will receive the response in the logOnResponse event.
 * @param {object} details - Your logon details. protocol_version will be set for you.
 */
CMClient.prototype.logOn = function(details) {
	details.protocol_version = PROTOCOL_VERSION;
	this.send({"msg": details.game_server_token ? EMsg.ClientLogonGameServer : EMsg.ClientLogon, "proto": {}}, new Schema.CMsgClientLogon(details).toBuffer());
};

/**
 * Send some data to Steam through our connection.
 * @param {object} header - Data to go in the message header
 * @param {Buffer|ByteBuffer} body - The message payload
 * @param {function} [callback] - If you expect a response to this message, a callback to be invoked when that response is received
 */
CMClient.prototype.send = function(header, body, callback) {
	// ignore any target job ID
	if (header.proto) {
		delete header.proto.jobid_target;
	} else {
		delete header.targetJobID;
	}

	if (ByteBuffer.isByteBuffer(body)) {
		body = body.toBuffer();
	}

	this._send(header, body, callback);
};

CMClient.prototype._netMsgReceived = function(data) {
	var rawEMsg = data.readUInt32LE(0);
	var eMsg = rawEMsg & ~PROTO_MASK;

	data = ByteBuffer.wrap(data, ByteBuffer.LITTLE_ENDIAN);

	var header, sourceJobID, targetJobID;
	if (eMsg == EMsg.ChannelEncryptRequest || eMsg == EMsg.ChannelEncryptResult) {
		header = Schema.MsgHdr.decode(data);
		sourceJobID = header.sourceJobID;
		targetJobID = header.targetJobID;

	} else if (rawEMsg & PROTO_MASK) {
		header = Schema.MsgHdrProtoBuf.decode(data);
		header.proto = Steam._processProto(header.proto);
		if (!this._sessionID && header.headerLength > 0) {
			this._sessionID = header.proto.client_sessionid;
			this.steamID = header.proto.steamid;
		}
		sourceJobID = header.proto.jobid_source;
		targetJobID = header.proto.jobid_target;

	} else {
		header = Schema.ExtendedClientMsgHdr.decode(data);
		sourceJobID = header.sourceJobID;
		targetJobID = header.targetJobID;
	}

	var body = data.toBuffer();

	if (eMsg in handlers) {
		handlers[header.msg].call(this, body);
	}

	if (sourceJobID != '18446744073709551615') {
		var callback = function(header, body, callback) {
			if (header.proto) {
				header.proto.jobid_target = sourceJobID;
			} else {
				header.targetJobID = sourceJobID;
			}

			this._send(header, body, callback);
		}.bind(this);
	}

	if (targetJobID in this._jobs) {
		this._jobs[targetJobID](header, body, callback);
	} else {
		this.emit('message', header, body, callback);
	}
};

CMClient.prototype._disconnected = function(had_error) {
	this.emit('debug', 'socket closed' + (had_error ? ' with an error' : ''));
	delete this._connection;

	if (had_error instanceof Error && had_error.proxyConnecting) {
		had_error.eresult = Steam.EResult.ConnectFailed;
		this.emit('error', had_error);
		return;
	}

	if (this.connected) {
		if (this.loggedOn) {
			this.emit('debug', 'unexpected disconnection');
			this.loggedOn = false;
			clearInterval(this._heartBeatFunc);
		}

		this.connected = false;
		this.emit('error', new Error('Disconnected'));
		return;
	}

	if (!this._autoRetry) {
		var err = new Error('Cannot Connect');
		err.hadError = err;

		this.emit('error', err);
		return;
	}

	if (!had_error) {
		this.connect(this._server);
		return;
	}

	var timeout = this._timeout || 1;
	this.emit('debug', 'waiting ' + timeout + ' secs');
	this._scheduledConnection = setTimeout(function() {
		delete this._scheduledConnection;
		this.connect(this._server);
	}.bind(this), timeout * 1000);

	this._timeout = timeout * 2;
};

CMClient.prototype._getServer = function(server, callback) {
	if (server && server.host && server.port) {
		// Make sure it's a valid server for this connection type
		if (this.protocol == Steam.EConnectionProtocol.WebSocket && !server.host.match(/^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/)) {
			callback(server);
			return;
		} else if (this.protocol != Steam.EConnectionProtocol.WebSocket && server.host.match(/^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/)) {
			callback(server);
			return;
		}
	}

	var self = this;

	// Pick a random one
	if (this.protocol == Steam.EConnectionProtocol.WebSocket) {
		if (g_WebSocketBootstrapping && Date.now() - g_WebSocketBootstrapping < 10000) {
			// Some other client is currently downloading the CM list and/or pinging CMs. Let's just try again soon.
			setTimeout(() => {
				this.emit('debug', 'Holding connection as WebSocket is bootstrapping');
				this._getServer(server, callback);
			}, 100);
			return;
		} else {
			g_WebSocketBootstrapping = Date.now();
		}

		if (g_WebSocketServers) {
			chooseWebSocketCM();
		} else {
			// Get em from the WebAPI
			this.emit('debug', 'Fetching WebSocket CM list from WebAPI');
			HTTPS.get("https://api.steampowered.com/ISteamDirectory/GetCMList/v1/?cellid=0", (res) => {
				if (res.statusCode != 200) {
					g_WebSocketBootstrapping = false;
					this.emit('debug', 'Error in GetCMList: HTTP ' + res.statusCode);
					var servers = require('../resources/servers_websocket.json');
					callback(servers[Math.floor(Math.random() * servers.length)]);
					return;
				}

				// parse the response
				var response = "";
				res.on('data', (chunk) => response += chunk.toString('utf8'));
				res.on('end', () => {
					try {
						var json = JSON.parse(response);
						if (!json.response || !json.response.serverlist_websockets || json.response.serverlist_websockets.length == 0) {
							throw new Error("Malformed response");
						}

						g_WebSocketServers = json.response.serverlist_websockets.map((server) => {
							var parts = server.split(':');
							return {
								"host": parts[0],
								"port": parseInt(parts[1], 10)
							};
						});

						chooseWebSocketCM();
					} catch (ex) {
						g_WebSocketBootstrapping = false;
						this.emit('debug', 'Error in GetCMList: ' + ex);
						var servers = require('../resources/servers_websocket.json');
						callback(servers[Math.floor(Math.random() * servers.length)]);
					}
				});
			}).on('error', (err) => {
				// Fallback to built-in list
				g_WebSocketBootstrapping = false;
				this.emit('debug', 'Error in GetCMList: ' + err.message);
				var servers = require('../resources/servers_websocket.json');
				callback(servers[Math.floor(Math.random() * servers.length)]);
			});
		}
	} else {
		callback(Steam.servers[Math.floor(Math.random() * Steam.servers.length)]);
	}

	function pingCM(cm, callback) {
		var addr = cm.host + ":" + cm.port;
		var options = {
			"host": cm.host,
			"port": cm.port,
			"timeout": 700,
			"path": "/cmping/",
			"agent": ProxyAgent.getAgent(true, self._httpProxy)
		};

		// The timeout option seems to not work
		var finished = false;
		var timeout = setTimeout(() => {
			if (finished) {
				return;
			}

			self.emit('debug', 'CM ' + addr + ' timed out');
			callback(null, null);
			finished = true;
		}, 700);

		var start = Date.now();
		HTTPS.get(options, (res) => {
			clearTimeout(timeout);
			if (finished) {
				return;
			}

			var latency = Date.now() - start;

			res.on('data', () => {}); // there is no body, so just throw it away

			if (res.statusCode != 200) {
				// CM is disqualified
				self.emit('debug', 'CM ' + addr + ' disqualified: HTTP error ' + res.statusCode);
				callback(null, null);
				return;
			}

			var load = parseInt(res.headers['x-steam-cmload'], 10) || 999;
			self.emit('debug', 'CM ' + addr + ' latency ' + latency + ' ms + load ' + load);
			callback(null, {"cm": cm, "load": load, "latency": latency});
		}).on('error', (err) => {
			clearTimeout(timeout);
			if (!finished) {
				self.emit('debug', 'CM ' + addr + ' disqualified: ' + err.message);
				callback(null, null); // if error, this CM is disqualified
			}
		});
	}

	function chooseWebSocketCM() {
		// Ping the CMs, if needed
		if (Date.now() - g_WebSocketLastPing < (1000 * 60 * 30)) {
			// Last ping was less than 30 minutes ago
			chooseWeightedWebSocketCM();
			return;
		}

		// Do the ping
		Async.map(g_WebSocketServers, pingCM, (err, cms) => {
			if (err) {
				// shouldn't be possible, but handle it anyway
				g_WebSocketBootstrapping = false;
				callback(g_WebSocketServers[Math.floor(Math.random() * g_WebSocketServers.length)]);
				return;
			}

			var filtered = cms.filter(cm => !!cm);
			if (filtered.length < 20) {
				// less than 20 up? assume steam is down and pick a random one
				g_WebSocketBootstrapping = false;
				self.emit('debug', 'Only ' + filtered.length + ' CMs responded to ping; choosing a random one');
				callback(g_WebSocketServers[Math.floor(Math.random() * g_WebSocketServers.length)]);
				return;
			}

			filtered.sort((a, b) => ((a.load * 2) + a.latency) < ((b.load * 2) + b.latency) ? -1 : 1);
			g_WebSocketPingedServers = filtered;
			g_WebSocketLastPing = Date.now();
			chooseWeightedWebSocketCM();
		});
	}

	function chooseWeightedWebSocketCM() {
		// Pick a random CM from the top 20
		g_WebSocketBootstrapping = false;
		var cms = g_WebSocketPingedServers.slice(0, 20);
		callback(cms[Math.floor(Math.random() * cms.length)].cm);
	}
};


// Handlers

var handlers = {};

handlers[EMsg.ChannelEncryptRequest] = function(data) {
	// assume server isn't dead
	this._connection.setTimeout(0);

	var buffer = ByteBuffer.wrap(data, ByteBuffer.LITTLE_ENDIAN);
	var protocol = buffer.readUint32();
	var universe = buffer.readUint32();
	var nonce = null;

	if (buffer.remaining() >= 16) {
		nonce = buffer.slice(buffer.offset, buffer.offset + 16).toBuffer();
		buffer.skip(16);
	}

	this.emit('debug', 'encrypt request: protocol ' + protocol + ', universe ' + universe + ', ' + (nonce ? 'nonce, ' : '') + (buffer.remaining()) + ' remaining bytes');

	var sessionKey = SteamCrypto.generateSessionKey(nonce);
	this._tempUseHmac = !!nonce;
	this._tempSessionKey = sessionKey.plain;
	var keyCrc = BufferCRC32.signed(sessionKey.encrypted);

	var encResp = new Schema.MsgChannelEncryptResponse().encode();
	var body = new ByteBuffer(encResp.limit + 128 + 4 + 4, ByteBuffer.LITTLE_ENDIAN); // key, crc, trailer

	body.append(encResp);
	body.append(sessionKey.encrypted);
	body.writeInt32(keyCrc);
	body.writeUint32(0); // TODO: check if the trailer is required
	body.flip();

	this.send({"msg": EMsg.ChannelEncryptResponse}, body.toBuffer());
};

handlers[EMsg.ChannelEncryptResult] = function(data) {
	var encResult = Schema.MsgChannelEncryptResult.decode(data);

	if (encResult.result == Steam.EResult.OK) {
		this._connection.sessionKey = this._tempSessionKey;
		this._connection.useHmac = this._tempUseHmac;
	} else {
		this.emit('error', new Error("Encryption fail: " + encResult.result));
		return;
	}

	this.connected = true;
	this.emit('connected', this._serverLoad);
};

handlers[EMsg.Multi] = function(data) {
	var msgMulti = Schema.CMsgMulti.decode(data);

	var payload = msgMulti.message_body.toBuffer();

	if (msgMulti.size_unzipped) {
		payload = Zlib.gunzipSync(payload);
	}

	// stop handling if user disconnected
	while (payload.length && this.connected) {
		var subSize = payload.readUInt32LE(0);
		this._netMsgReceived(payload.slice(4, 4 + subSize));
		payload = payload.slice(4 + subSize);
	}
};

handlers[EMsg.ClientLogOnResponse] = function(data) {
	var logonResp = Schema.CMsgClientLogonResponse.decode(data);
	var eresult = logonResp.eresult;

	if (eresult == Steam.EResult.OK) {
		var hbDelay = logonResp.out_of_game_heartbeat_seconds;

		this._heartBeatFunc = setInterval(function() {
			this.send({
				"msg": EMsg.ClientHeartBeat,
				"proto": {}
			}, new Schema.CMsgClientHeartBeat().toBuffer());
		}.bind(this), hbDelay * 1000);

		this.loggedOn = true;
	}

	this.emit('logOnResponse', Steam._processProto(logonResp));
};

handlers[EMsg.ClientLoggedOff] = function(data) {
	this.loggedOn = false;
	clearInterval(this._heartBeatFunc);

	var eresult = Schema.CMsgClientLoggedOff.decode(data).eresult;

	this.emit('loggedOff', eresult);
};

handlers[EMsg.ClientCMList] = function(data) {
	var list = Schema.CMsgClientCMList.decode(data);
	var servers = list.cm_addresses.map(function(number, index) {
		var buf = new Buffer(4);
		buf.writeUInt32BE(number, 0);
		return {
			host: [].join.call(buf, '.'),
			port: list.cm_ports[index]
		};
	});

	this.emit('servers', servers);
	Steam.servers = servers;
};


Steam.CMClient = CMClient;