Port chromium message parser and refactor (#13)

* Port chromium message parser and refactor

This commit:
1. Adds sanity tests against chrome gcm. To use the tests copy
   test/keys.template.js to test/keys.js and populate it with gcm
   credentials / sender id. Then do `npm run test`

2. Ports chromium's ConnectionHandlerImpl::WaitForData. This fixes the
   problem of not parsing the wire format correctly. There are two
   more things here that would be nice to port:
   - Add timeouts while waiting for message bytes
   - Error handling around protobufs

   I did make the simplification of not supporting the
   kDefaultDataPacketLimit limit (it shouldn't affect correctness)

   ref: https://cs.chromium.org/chromium/src/google_apis/gcm/engine/connection_handler_impl.cc?rcl=dc7c41bc0ee5fee0ed269495dde6b8c40df43e40&l=178

3. Refactors client and socket into a class. Combining them made it
   easier to cleanup the socket at the end of the test. It also has the
   benefit of making the retrying a bit cleaner.

* Move constants to their own file and use when logging in

* Add return after emitError and update style to make internal things "private"

* Jest + format

* Remove error listener from parser when destroying

* Silently drop notifications that cannot be decrypted

We are unable to decrypt some notifications. When testing in production
we are able to receive future notifications, though. So silently drop
these notifications and add them to the persistentIds so we don't try to
re-process them.

* Clear persistent ids after logging in

This commit adds persistent ids to the login request, which is one way
GCM acks them. The other way GCM acks is by keeping track of stream ids
and sending those back and forth.

That needs to be done too, but at least this should help clear them periodically.

* Code cleanup

1. Remove the test filter regex for jest, it was complaining about it
2. Run the linter
3. Remove comment that is no longer true
4. Make encoding the login request more succint

.gitignore

@@ -6,3 +6,4 @@ coverage
 storage.json
 web/
 .esm-cache
+test/keys.js

package.json

@@ -13,7 +13,16 @@
     "pretty:check":
       "prettier-eslint --single-quote --trailing-comma es5 --list-different --log-level silent \"**/*.js\" \"**/*.json\"",
     "lint": "eslint 'src/**/*.js'",
-    "lint:fix": "eslint 'src/**/*.js' --fix"
+    "lint:fix": "eslint 'src/**/*.js' --fix",
+    "test": "jest",
+    "precommit": "lint-staged"
+  },
+  "lint-staged": {
+    "*.js": [
+      "eslint --fix",
+      "prettier-eslint --single-quote --trailing-comma es5 --write"
+    ],
+    "*.json": ["prettier-eslint --single-quote --trailing-comma es5 --write"]
   },
   "repository": {
     "type": "git",
@@ -40,6 +49,9 @@
     "eslint": "^4.4.1",
     "eslint-plugin-jest": "^20.0.3",
     "http-proxy": "^1.16.2",
+    "husky": "^0.14.3",
+    "jest": "^22.2.2",
+    "lint-staged": "^6.1.0",
     "prettier": "^1.6.1",
     "prettier-eslint": "^7.0.0",
     "prettier-eslint-cli": "^4.3.0",

src/client.js

@@ -0,0 +1,201 @@
+const EventEmitter = require('events');
+const Long = require('long');
+const Parser = require('./parser');
+const decrypt = require('./utils/decrypt');
+const path = require('path');
+const tls = require('tls');
+const { checkIn } = require('./gcm');
+const {
+  kMCSVersion,
+  kLoginRequestTag,
+  kDataMessageStanzaTag,
+  kLoginResponseTag,
+} = require('./constants');
+const { load } = require('protobufjs');
+
+const HOST = 'mtalk.google.com';
+const PORT = 5228;
+const MAX_RETRY_TIMEOUT = 15;
+
+let proto = null;
+
+module.exports = class Client extends EventEmitter {
+  static async init() {
+    if (proto) {
+      return;
+    }
+    proto = await load(path.resolve(__dirname, 'mcs.proto'));
+  }
+
+  constructor(credentials, persistentIds) {
+    super();
+    this._credentials = credentials;
+    this._persistentIds = persistentIds || [];
+    this._retryCount = 0;
+    this._onSocketConnect = this._onSocketConnect.bind(this);
+    this._onSocketClose = this._onSocketClose.bind(this);
+    this._onSocketError = this._onSocketError.bind(this);
+    this._onMessage = this._onMessage.bind(this);
+    this._onParserError = this._onParserError.bind(this);
+  }
+
+  async connect() {
+    await Client.init();
+    await this._checkIn();
+    this._connect();
+    // can happen if the socket immediately closes after being created
+    if (!this._socket) {
+      return;
+    }
+    await Parser.init();
+    // can happen if the socket immediately closes after being created
+    if (!this._socket) {
+      return;
+    }
+    this._parser = new Parser(this._socket);
+    this._parser.on('message', this._onMessage);
+    this._parser.on('error', this._onParserError);
+  }
+
+  destroy() {
+    this._destroy();
+  }
+
+  async _checkIn() {
+    return checkIn(
+      this._credentials.gcm.androidId,
+      this._credentials.gcm.securityToken
+    );
+  }
+
+  _connect() {
+    this._socket = new tls.TLSSocket();
+    this._socket.setKeepAlive(true);
+    this._socket.on('connect', this._onSocketConnect);
+    this._socket.on('close', this._onSocketClose);
+    this._socket.on('error', this._onSocketError);
+    this._socket.connect({ host : HOST, port : PORT });
+    this._socket.write(this._loginBuffer());
+  }
+
+  _destroy() {
+    clearTimeout(this._retryTimeout);
+    if (this._socket) {
+      this._socket.removeListener('connect', this._onSocketConnect);
+      this._socket.removeListener('close', this._onSocketClose);
+      this._socket.removeListener('error', this._onSocketError);
+      this._socket.destroy();
+      this._socket = null;
+    }
+    if (this._parser) {
+      this._parser.removeListener('message', this._onMessage);
+      this._parser.removeListener('error', this._onParserError);
+      this._parser.destroy();
+      this._parser = null;
+    }
+  }
+
+  _loginBuffer() {
+    const LoginRequestType = proto.lookupType('mcs_proto.LoginRequest');
+    const hexAndroidId = Long.fromString(
+      this._credentials.gcm.androidId
+    ).toString(16);
+    const loginRequest = {
+      adaptiveHeartbeat    : false,
+      authService          : 2,
+      authToken            : this._credentials.gcm.securityToken,
+      id                   : 'chrome-63.0.3234.0',
+      domain               : 'mcs.android.com',
+      deviceId             : `android-${hexAndroidId}`,
+      networkType          : 1,
+      resource             : this._credentials.gcm.androidId,
+      user                 : this._credentials.gcm.androidId,
+      useRmq2              : true,
+      setting              : [{ name : 'new_vc', value : '1' }],
+      // Id of the last notification received
+      clientEvent          : [],
+      receivedPersistentId : this._persistentIds,
+    };
+
+    const errorMessage = LoginRequestType.verify(loginRequest);
+    if (errorMessage) {
+      throw new Error(errorMessage);
+    }
+
+    const buffer = LoginRequestType.encodeDelimited(loginRequest).finish();
+
+    return Buffer.concat([
+      Buffer.from([kMCSVersion, kLoginRequestTag]),
+      buffer,
+    ]);
+  }
+
+  _onSocketConnect() {
+    this._retryCount = 0;
+    this.emit('connect');
+  }
+
+  _onSocketClose() {
+    this._retry();
+  }
+
+  _onSocketError(error) {
+    // ignore, the close handler takes care of retry
+    console.error(error);
+  }
+
+  _onParserError(error) {
+    this._retry();
+    console.error(error);
+  }
+
+  _retry() {
+    this._destroy();
+    const timeout = Math.min(++this._retryCount, MAX_RETRY_TIMEOUT) * 1000;
+    this._retryTimeout = setTimeout(this.connect.bind(this), timeout);
+  }
+
+  _onMessage({ tag, object }) {
+    if (tag === kLoginResponseTag) {
+      // clear persistent ids, as we just sent them to the server while logging
+      // in
+      this._persistentIds = [];
+    } else if (tag === kDataMessageStanzaTag) {
+      this._onDataMessage(object);
+    }
+  }
+
+  _onDataMessage(object) {
+    if (this._persistentIds.includes(object.persistentId)) {
+      return;
+    }
+
+    let message;
+    try {
+      message = decrypt(object, this._credentials.keys);
+    } catch (error) {
+      if (
+        error.message.includes(
+          'Unsupported state or unable to authenticate data'
+        )
+      ) {
+        // NOTE(ibash) Periodically we're unable to decrypt notifications. In
+        // all cases we've been able to receive future notifications using the
+        // same keys. So, we sliently drop this notification.
+        this._persistentIds.push(object.persistentId);
+        return;
+      } else {
+        throw error;
+      }
+    }
+
+    // Maintain persistentIds updated with the very last received value
+    this._persistentIds.push(object.persistentId);
+    // Send notification
+    this.emit('ON_NOTIFICATION_RECEIVED', {
+      notification : message,
+      // Needs to be saved by the client
+      persistentId : object.persistentId,
+    });
+  }
+};