The Tor Browser: comparison dom/push/src/PushService.jsm

--1:000000000000
+:4b9f7f6f530c
+/* This Source Code Form is subject to the terms of the Mozilla Public
+* License, v. 2.0. If a copy of the MPL was not distributed with this file,
+* You can obtain one at http://mozilla.org/MPL/2.0/. */
+"use strict";
+// Don't modify this, instead set services.push.debug.
+let gDebuggingEnabled = false;
+function debug(s) {
+if (gDebuggingEnabled)
+dump("-*- PushService.jsm: " + s + "\n");
+}
+const Cc = Components.classes;
+const Ci = Components.interfaces;
+const Cu = Components.utils;
+const Cr = Components.results;
+Cu.import("resource://gre/modules/XPCOMUtils.jsm");
+Cu.import("resource://gre/modules/Services.jsm");
+Cu.import("resource://gre/modules/IndexedDBHelper.jsm");
+Cu.import("resource://gre/modules/Timer.jsm");
+Cu.import("resource://gre/modules/Preferences.jsm");
+Cu.import("resource://gre/modules/Promise.jsm");
+Cu.importGlobalProperties(["indexedDB"]);
+XPCOMUtils.defineLazyModuleGetter(this, "AlarmService",
+"resource://gre/modules/AlarmService.jsm");
+this.EXPORTED_SYMBOLS = ["PushService"];
+const prefs = new Preferences("services.push.");
+// Set debug first so that all debugging actually works.
+gDebuggingEnabled = prefs.get("debug");
+const kPUSHDB_DB_NAME = "push";
+const kPUSHDB_DB_VERSION = 1; // Change this if the IndexedDB format changes
+const kPUSHDB_STORE_NAME = "push";
+const kUDP_WAKEUP_WS_STATUS_CODE = 4774;  // WebSocket Close status code sent
+// by server to signal that it can
+// wake client up using UDP.
+const kCHILD_PROCESS_MESSAGES = ["Push:Register", "Push:Unregister",
+"Push:Registrations"];
+// This is a singleton
+this.PushDB = function PushDB() {
+debug("PushDB()");
+// set the indexeddb database
+this.initDBHelper(kPUSHDB_DB_NAME, kPUSHDB_DB_VERSION,
+[kPUSHDB_STORE_NAME]);
+};
+this.PushDB.prototype = {
+__proto__: IndexedDBHelper.prototype,
+upgradeSchema: function(aTransaction, aDb, aOldVersion, aNewVersion) {
+debug("PushDB.upgradeSchema()")
+let objectStore = aDb.createObjectStore(kPUSHDB_STORE_NAME,
+{ keyPath: "channelID" });
+// index to fetch records based on endpoints. used by unregister
+objectStore.createIndex("pushEndpoint", "pushEndpoint", { unique: true });
+// index to fetch records per manifest, so we can identify endpoints
+// associated with an app. Since an app can have multiple endpoints
+// uniqueness cannot be enforced
+objectStore.createIndex("manifestURL", "manifestURL", { unique: false });
+},
+/*
+* @param aChannelRecord
+*        The record to be added.
+* @param aSuccessCb
+*        Callback function to invoke with result ID.
+* @param aErrorCb [optional]
+*        Callback function to invoke when there was an error.
+*/
+put: function(aChannelRecord, aSuccessCb, aErrorCb) {
+debug("put()");
+this.newTxn(
+"readwrite",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+debug("Going to put " + aChannelRecord.channelID);
+aStore.put(aChannelRecord).onsuccess = function setTxnResult(aEvent) {
+debug("Request successful. Updated record ID: " +
+aEvent.target.result);
+};
+},
+aSuccessCb,
+aErrorCb
+);
+},
+/*
+* @param aChannelID
+*        The ID of record to be deleted.
+* @param aSuccessCb
+*        Callback function to invoke with result.
+* @param aErrorCb [optional]
+*        Callback function to invoke when there was an error.
+*/
+delete: function(aChannelID, aSuccessCb, aErrorCb) {
+debug("delete()");
+this.newTxn(
+"readwrite",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+debug("Going to delete " + aChannelID);
+aStore.delete(aChannelID);
+},
+aSuccessCb,
+aErrorCb
+);
+},
+getByPushEndpoint: function(aPushEndpoint, aSuccessCb, aErrorCb) {
+debug("getByPushEndpoint()");
+this.newTxn(
+"readonly",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+aTxn.result = undefined;
+let index = aStore.index("pushEndpoint");
+index.get(aPushEndpoint).onsuccess = function setTxnResult(aEvent) {
+aTxn.result = aEvent.target.result;
+debug("Fetch successful " + aEvent.target.result);
+}
+},
+aSuccessCb,
+aErrorCb
+);
+},
+getByChannelID: function(aChannelID, aSuccessCb, aErrorCb) {
+debug("getByChannelID()");
+this.newTxn(
+"readonly",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+aTxn.result = undefined;
+aStore.get(aChannelID).onsuccess = function setTxnResult(aEvent) {
+aTxn.result = aEvent.target.result;
+debug("Fetch successful " + aEvent.target.result);
+}
+},
+aSuccessCb,
+aErrorCb
+);
+},
+getAllByManifestURL: function(aManifestURL, aSuccessCb, aErrorCb) {
+debug("getAllByManifestURL()");
+if (!aManifestURL) {
+if (typeof aErrorCb == "function") {
+aErrorCb("PushDB.getAllByManifestURL: Got undefined aManifestURL");
+}
+return;
+}
+let self = this;
+this.newTxn(
+"readonly",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+let index = aStore.index("manifestURL");
+let range = IDBKeyRange.only(aManifestURL);
+aTxn.result = [];
+index.openCursor(range).onsuccess = function(event) {
+let cursor = event.target.result;
+if (cursor) {
+debug(cursor.value.manifestURL + " " + cursor.value.channelID);
+aTxn.result.push(cursor.value);
+cursor.continue();
+}
+}
+},
+aSuccessCb,
+aErrorCb
+);
+},
+getAllChannelIDs: function(aSuccessCb, aErrorCb) {
+debug("getAllChannelIDs()");
+this.newTxn(
+"readonly",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+aStore.mozGetAll().onsuccess = function(event) {
+aTxn.result = event.target.result;
+}
+},
+aSuccessCb,
+aErrorCb
+);
+},
+drop: function(aSuccessCb, aErrorCb) {
+debug("drop()");
+this.newTxn(
+"readwrite",
+kPUSHDB_STORE_NAME,
+function txnCb(aTxn, aStore) {
+aStore.clear();
+},
+aSuccessCb(),
+aErrorCb()
+);
+}
+};
+/**
+* A proxy between the PushService and the WebSocket. The listener is used so
+* that the PushService can silence messages from the WebSocket by setting
+* PushWebSocketListener._pushService to null. This is required because
+* a WebSocket can continue to send messages or errors after it has been
+* closed but the PushService may not be interested in these. It's easier to
+* stop listening than to have checks at specific points.
+*/
+this.PushWebSocketListener = function(pushService) {
+this._pushService = pushService;
+}
+this.PushWebSocketListener.prototype = {
+onStart: function(context) {
+if (!this._pushService)
+return;
+this._pushService._wsOnStart(context);
+},
+onStop: function(context, statusCode) {
+if (!this._pushService)
+return;
+this._pushService._wsOnStop(context, statusCode);
+},
+onAcknowledge: function(context, size) {
+// EMPTY
+},
+onBinaryMessageAvailable: function(context, message) {
+// EMPTY
+},
+onMessageAvailable: function(context, message) {
+if (!this._pushService)
+return;
+this._pushService._wsOnMessageAvailable(context, message);
+},
+onServerClose: function(context, aStatusCode, aReason) {
+if (!this._pushService)
+return;
+this._pushService._wsOnServerClose(context, aStatusCode, aReason);
+}
+}
+// websocket states
+// websocket is off
+const STATE_SHUT_DOWN = 0;
+// Websocket has been opened on client side, waiting for successful open.
+// (_wsOnStart)
+const STATE_WAITING_FOR_WS_START = 1;
+// Websocket opened, hello sent, waiting for server reply (_handleHelloReply).
+const STATE_WAITING_FOR_HELLO = 2;
+// Websocket operational, handshake completed, begin protocol messaging.
+const STATE_READY = 3;
+/**
+* The implementation of the SimplePush system. This runs in the B2G parent
+* process and is started on boot. It uses WebSockets to communicate with the
+* server and PushDB (IndexedDB) for persistence.
+*/
+this.PushService = {
+observe: function observe(aSubject, aTopic, aData) {
+switch (aTopic) {
+/*
+* We need to call uninit() on shutdown to clean up things that modules aren't very good
+* at automatically cleaning up, so we don't get shutdown leaks on browser shutdown.
+*/
+case "xpcom-shutdown":
+this.uninit();
+case "network-active-changed":         /* On B2G. */
+case "network:offline-status-changed": /* On desktop. */
+// In case of network-active-changed, always disconnect existing
+// connections. In case of offline-status changing from offline to
+// online, it is likely that these statements will be no-ops.
+if (this._udpServer) {
+this._udpServer.close();
+// Set to null since this is checked in _listenForUDPWakeup()
+this._udpServer = null;
+}
+this._shutdownWS();
+// Try to connect if network-active-changed or the offline-status
+// changed to online.
+if (aTopic === "network-active-changed" || aData === "online") {
+this._startListeningIfChannelsPresent();
+}
+break;
+case "nsPref:changed":
+if (aData == "services.push.serverURL") {
+debug("services.push.serverURL changed! websocket. new value " +
+prefs.get("serverURL"));
+this._shutdownWS();
+} else if (aData == "services.push.connection.enabled") {
+if (prefs.get("connection.enabled")) {
+this._startListeningIfChannelsPresent();
+} else {
+this._shutdownWS();
+}
+} else if (aData == "services.push.debug") {
+gDebuggingEnabled = prefs.get("debug");
+}
+break;
+case "timer-callback":
+if (aSubject == this._requestTimeoutTimer) {
+if (Object.keys(this._pendingRequests).length == 0) {
+this._requestTimeoutTimer.cancel();
+}
+// Set to true if at least one request timed out.
+let requestTimedOut = false;
+for (let channelID in this._pendingRequests) {
+let duration = Date.now() - this._pendingRequests[channelID].ctime;
+// If any of the registration requests time out, all the ones after it
+// also made to fail, since we are going to be disconnecting the socket.
+if (requestTimedOut || duration > this._requestTimeout) {
+debug("Request timeout: Removing " + channelID);
+requestTimedOut = true;
+this._pendingRequests[channelID]
+.deferred.reject({status: 0, error: "TimeoutError"});
+delete this._pendingRequests[channelID];
+for (let i = this._requestQueue.length - 1; i >= 0; --i)
+if (this._requestQueue[i].channelID == channelID)
+this._requestQueue.splice(i, 1);
+}
+}
+// The most likely reason for a registration request timing out is
+// that the socket has disconnected. Best to reconnect.
+if (requestTimedOut) {
+this._shutdownWS();
+this._reconnectAfterBackoff();
+}
+}
+break;
+case "webapps-clear-data":
+debug("webapps-clear-data");
+let data = aSubject.QueryInterface(Ci.mozIApplicationClearPrivateDataParams);
+if (!data) {
+debug("webapps-clear-data: Failed to get information about application");
+return;
+}
+// Only remove push registrations for apps.
+if (data.browserOnly) {
+return;
+}
+let appsService = Cc["@mozilla.org/AppsService;1"]
+.getService(Ci.nsIAppsService);
+let manifestURL = appsService.getManifestURLByLocalId(data.appId);
+if (!manifestURL) {
+debug("webapps-clear-data: No manifest URL found for " + data.appId);
+return;
+}
+this._db.getAllByManifestURL(manifestURL, function(records) {
+debug("Got " + records.length);
+for (let i = 0; i < records.length; i++) {
+this._db.delete(records[i].channelID, null, function() {
+debug("webapps-clear-data: " + manifestURL +
+" Could not delete entry " + records[i].channelID);
+});
+// courtesy, but don't establish a connection
+// just for it
+if (this._ws) {
+debug("Had a connection, so telling the server");
+this._send("unregister", {channelID: records[i].channelID});
+}
+}
+}.bind(this), function() {
+debug("webapps-clear-data: Error in getAllByManifestURL(" + manifestURL + ")");
+});
+break;
+}
+},
+get _UAID() {
+return prefs.get("userAgentID");
+},
+set _UAID(newID) {
+if (typeof(newID) !== "string") {
+debug("Got invalid, non-string UAID " + newID +
+". Not updating userAgentID");
+return;
+}
+debug("New _UAID: " + newID);
+prefs.set("userAgentID", newID);
+},
+// keeps requests buffered if the websocket disconnects or is not connected
+_requestQueue: [],
+_ws: null,
+_pendingRequests: {},
+_currentState: STATE_SHUT_DOWN,
+_requestTimeout: 0,
+_requestTimeoutTimer: null,
+_retryFailCount: 0,
+/**
+* According to the WS spec, servers should immediately close the underlying
+* TCP connection after they close a WebSocket. This causes wsOnStop to be
+* called with error NS_BASE_STREAM_CLOSED. Since the client has to keep the
+* WebSocket up, it should try to reconnect. But if the server closes the
+* WebSocket because it will wake up the client via UDP, then the client
+* shouldn't re-establish the connection. If the server says that it will
+* wake up the client over UDP, this is set to true in wsOnServerClose. It is
+* checked in wsOnStop.
+*/
+_willBeWokenUpByUDP: false,
+/**
+* Sends a message to the Push Server through an open websocket.
+* typeof(msg) shall be an object
+*/
+_wsSendMessage: function(msg) {
+if (!this._ws) {
+debug("No WebSocket initialized. Cannot send a message.");
+return;
+}
+msg = JSON.stringify(msg);
+debug("Sending message: " + msg);
+this._ws.sendMsg(msg);
+},
+init: function() {
+debug("init()");
+if (!prefs.get("enabled"))
+return null;
+this._db = new PushDB();
+let ppmm = Cc["@mozilla.org/parentprocessmessagemanager;1"]
+.getService(Ci.nsIMessageBroadcaster);
+kCHILD_PROCESS_MESSAGES.forEach(function addMessage(msgName) {
+ppmm.addMessageListener(msgName, this);
+}.bind(this));
+this._alarmID = null;
+this._requestTimeout = prefs.get("requestTimeout");
+this._startListeningIfChannelsPresent();
+Services.obs.addObserver(this, "xpcom-shutdown", false);
+Services.obs.addObserver(this, "webapps-clear-data", false);
+// On B2G the NetworkManager interface fires a network-active-changed
+// event.
+//
+// The "active network" is based on priority - i.e. Wi-Fi has higher
+// priority than data. The PushService should just use the preferred
+// network, and not care about all interface changes.
+// network-active-changed is not fired when the network goes offline, but
+// socket connections time out. The check for Services.io.offline in
+// _beginWSSetup() prevents unnecessary retries.  When the network comes
+// back online, network-active-changed is fired.
+//
+// On non-B2G platforms, the offline-status-changed event is used to know
+// when to (dis)connect. It may not fire if the underlying OS changes
+// networks; in such a case we rely on timeout.
+//
+// On B2G both events fire, one after the other, when the network goes
+// online, so we explicitly check for the presence of NetworkManager and
+// don't add an observer for offline-status-changed on B2G.
+Services.obs.addObserver(this, this._getNetworkStateChangeEventName(), false);
+// This is only used for testing. Different tests require connecting to
+// slightly different URLs.
+prefs.observe("serverURL", this);
+// Used to monitor if the user wishes to disable Push.
+prefs.observe("connection.enabled", this);
+// Debugging
+prefs.observe("debug", this);
+this._started = true;
+},
+_shutdownWS: function() {
+debug("shutdownWS()");
+this._currentState = STATE_SHUT_DOWN;
+this._willBeWokenUpByUDP = false;
+if (this._wsListener)
+this._wsListener._pushService = null;
+try {
+this._ws.close(0, null);
+} catch (e) {}
+this._ws = null;
+this._waitingForPong = false;
+this._stopAlarm();
+},
+uninit: function() {
+if (!this._started)
+return;
+debug("uninit()");
+prefs.ignore("debug", this);
+prefs.ignore("connection.enabled", this);
+prefs.ignore("serverURL", this);
+Services.obs.removeObserver(this, this._getNetworkStateChangeEventName());
+Services.obs.removeObserver(this, "webapps-clear-data", false);
+Services.obs.removeObserver(this, "xpcom-shutdown", false);
+if (this._db) {
+this._db.close();
+this._db = null;
+}
+if (this._udpServer) {
+this._udpServer.close();
+this._udpServer = null;
+}
+// All pending requests (ideally none) are dropped at this point. We
+// shouldn't have any applications performing registration/unregistration
+// or receiving notifications.
+this._shutdownWS();
+// At this point, profile-change-net-teardown has already fired, so the
+// WebSocket has been closed with NS_ERROR_ABORT (if it was up) and will
+// try to reconnect. Stop the timer.
+this._stopAlarm();
+if (this._requestTimeoutTimer) {
+this._requestTimeoutTimer.cancel();
+}
+debug("shutdown complete!");
+},
+/**
+* How retries work:  The goal is to ensure websocket is always up on
+* networks not supporting UDP. So the websocket should only be shutdown if
+* onServerClose indicates UDP wakeup.  If WS is closed due to socket error,
+* _reconnectAfterBackoff() is called.  The retry alarm is started and when
+* it times out, beginWSSetup() is called again.
+*
+* On a successful connection, the alarm is cancelled in
+* wsOnMessageAvailable() when the ping alarm is started.
+*
+* If we are in the middle of a timeout (i.e. waiting), but
+* a register/unregister is called, we don't want to wait around anymore.
+* _sendRequest will automatically call beginWSSetup(), which will cancel the
+* timer. In addition since the state will have changed, even if a pending
+* timer event comes in (because the timer fired the event before it was
+* cancelled), so the connection won't be reset.
+*/
+_reconnectAfterBackoff: function() {
+debug("reconnectAfterBackoff()");
+// Calculate new timeout, but cap it to pingInterval.
+let retryTimeout = prefs.get("retryBaseInterval") *
+Math.pow(2, this._retryFailCount);
+retryTimeout = Math.min(retryTimeout, prefs.get("pingInterval"));
+this._retryFailCount++;
+debug("Retry in " + retryTimeout + " Try number " + this._retryFailCount);
+this._setAlarm(retryTimeout);
+},
+_beginWSSetup: function() {
+debug("beginWSSetup()");
+if (this._currentState != STATE_SHUT_DOWN) {
+debug("_beginWSSetup: Not in shutdown state! Current state " +
+this._currentState);
+return;
+}
+if (!prefs.get("connection.enabled")) {
+debug("_beginWSSetup: connection.enabled is not set to true. Aborting.");
+return;
+}
+// Stop any pending reconnects scheduled for the near future.
+this._stopAlarm();
+if (Services.io.offline) {
+debug("Network is offline.");
+return;
+}
+let serverURL = prefs.get("serverURL");
+if (!serverURL) {
+debug("No services.push.serverURL found!");
+return;
+}
+let uri;
+try {
+uri = Services.io.newURI(serverURL, null, null);
+} catch(e) {
+debug("Error creating valid URI from services.push.serverURL (" +
+serverURL + ")");
+return;
+}
+if (uri.scheme === "wss") {
+this._ws = Cc["@mozilla.org/network/protocol;1?name=wss"]
+.createInstance(Ci.nsIWebSocketChannel);
+}
+else if (uri.scheme === "ws") {
+debug("Push over an insecure connection (ws://) is not allowed!");
+return;
+}
+else {
+debug("Unsupported websocket scheme " + uri.scheme);
+return;
+}
+debug("serverURL: " + uri.spec);
+this._wsListener = new PushWebSocketListener(this);
+this._ws.protocol = "push-notification";
+this._ws.asyncOpen(uri, serverURL, this._wsListener, null);
+this._currentState = STATE_WAITING_FOR_WS_START;
+},
+_startListeningIfChannelsPresent: function() {
+// Check to see if we need to do anything.
+this._db.getAllChannelIDs(function(channelIDs) {
+if (channelIDs.length > 0) {
+this._beginWSSetup();
+}
+}.bind(this));
+},
+/** |delay| should be in milliseconds. */
+_setAlarm: function(delay) {
+// Bug 909270: Since calls to AlarmService.add() are async, calls must be
+// 'queued' to ensure only one alarm is ever active.
+if (this._settingAlarm) {
+// onSuccess will handle the set. Overwriting the variable enforces the
+// last-writer-wins semantics.
+this._queuedAlarmDelay = delay;
+this._waitingForAlarmSet = true;
+return;
+}
+// Stop any existing alarm.
+this._stopAlarm();
+this._settingAlarm = true;
+AlarmService.add(
+{
+date: new Date(Date.now() + delay),
+ignoreTimezone: true
+},
+this._onAlarmFired.bind(this),
+function onSuccess(alarmID) {
+this._alarmID = alarmID;
+debug("Set alarm " + delay + " in the future " + this._alarmID);
+this._settingAlarm = false;
+if (this._waitingForAlarmSet) {
+this._waitingForAlarmSet = false;
+this._setAlarm(this._queuedAlarmDelay);
+}
+}.bind(this)
+)
+},
+_stopAlarm: function() {
+if (this._alarmID !== null) {
+debug("Stopped existing alarm " + this._alarmID);
+AlarmService.remove(this._alarmID);
+this._alarmID = null;
+}
+},
+/**
+* There is only one alarm active at any time. This alarm has 3 intervals
+* corresponding to 3 tasks.
+*
+* 1) Reconnect on ping timeout.
+*    If we haven't received any messages from the server by the time this
+*    alarm fires, the connection is closed and PushService tries to
+*    reconnect, repurposing the alarm for (3).
+*
+* 2) Send a ping.
+*    The protocol sends a ping ({}) on the wire every pingInterval ms. Once
+*    it sends the ping, the alarm goes to task (1) which is waiting for
+*    a pong. If data is received after the ping is sent,
+*    _wsOnMessageAvailable() will reset the ping alarm (which cancels
+*    waiting for the pong). So as long as the connection is fine, pong alarm
+*    never fires.
+*
+* 3) Reconnect after backoff.
+*    The alarm is set by _reconnectAfterBackoff() and increases in duration
+*    every time we try and fail to connect.  When it triggers, websocket
+*    setup begins again. On successful socket setup, the socket starts
+*    receiving messages. The alarm now goes to (2) where it monitors the
+*    WebSocket by sending a ping.  Since incoming data is a sign of the
+*    connection being up, the ping alarm is reset every time data is
+*    received.
+*/
+_onAlarmFired: function() {
+// Conditions are arranged in decreasing specificity.
+// i.e. when _waitingForPong is true, other conditions are also true.
+if (this._waitingForPong) {
+debug("Did not receive pong in time. Reconnecting WebSocket.");
+this._shutdownWS();
+this._reconnectAfterBackoff();
+}
+else if (this._currentState == STATE_READY) {
+// Send a ping.
+// Bypass the queue; we don't want this to be kept pending.
+// Watch out for exception in case the socket has disconnected.
+// When this happens, we pretend the ping was sent and don't specially
+// handle the exception, as the lack of a pong will lead to the socket
+// being reset.
+try {
+this._wsSendMessage({});
+} catch (e) {
+}
+this._waitingForPong = true;
+this._setAlarm(prefs.get("requestTimeout"));
+}
+else if (this._alarmID !== null) {
+debug("reconnect alarm fired.");
+// Reconnect after back-off.
+// The check for a non-null _alarmID prevents a situation where the alarm
+// fires, but _shutdownWS() is called from another code-path (e.g.
+// network state change) and we don't want to reconnect.
+//
+// It also handles the case where _beginWSSetup() is called from another
+// code-path.
+//
+// alarmID will be non-null only when no shutdown/connect is
+// called between _reconnectAfterBackoff() setting the alarm and the
+// alarm firing.
+// Websocket is shut down. Backoff interval expired, try to connect.
+this._beginWSSetup();
+}
+},
+/**
+* Protocol handler invoked by server message.
+*/
+_handleHelloReply: function(reply) {
+debug("handleHelloReply()");
+if (this._currentState != STATE_WAITING_FOR_HELLO) {
+debug("Unexpected state " + this._currentState +
+"(expected STATE_WAITING_FOR_HELLO)");
+this._shutdownWS();
+return;
+}
+if (typeof reply.uaid !== "string") {
+debug("No UAID received or non string UAID received");
+this._shutdownWS();
+return;
+}
+if (reply.uaid === "") {
+debug("Empty UAID received!");
+this._shutdownWS();
+return;
+}
+// To avoid sticking extra large values sent by an evil server into prefs.
+if (reply.uaid.length > 128) {
+debug("UAID received from server was too long: " +
+reply.uaid);
+this._shutdownWS();
+return;
+}
+function finishHandshake() {
+this._UAID = reply.uaid;
+this._currentState = STATE_READY;
+this._processNextRequestInQueue();
+}
+// By this point we've got a UAID from the server that we are ready to
+// accept.
+//
+// If we already had a valid UAID before, we have to ask apps to
+// re-register.
+if (this._UAID && this._UAID != reply.uaid) {
+debug("got new UAID: all re-register");
+this._notifyAllAppsRegister()
+.then(this._dropRegistrations.bind(this))
+.then(finishHandshake.bind(this));
+return;
+}
+// otherwise we are good to go
+finishHandshake.bind(this)();
+},
+/**
+* Protocol handler invoked by server message.
+*/
+_handleRegisterReply: function(reply) {
+debug("handleRegisterReply()");
+if (typeof reply.channelID !== "string" ||
+typeof this._pendingRequests[reply.channelID] !== "object")
+return;
+let tmp = this._pendingRequests[reply.channelID];
+delete this._pendingRequests[reply.channelID];
+if (Object.keys(this._pendingRequests).length == 0 &&
+this._requestTimeoutTimer)
+this._requestTimeoutTimer.cancel();
+if (reply.status == 200) {
+tmp.deferred.resolve(reply);
+} else {
+tmp.deferred.reject(reply);
+}
+},
+/**
+* Protocol handler invoked by server message.
+*/
+_handleNotificationReply: function(reply) {
+debug("handleNotificationReply()");
+if (typeof reply.updates !== 'object') {
+debug("No 'updates' field in response. Type = " + typeof reply.updates);
+return;
+}
+debug("Reply updates: " + reply.updates.length);
+for (let i = 0; i < reply.updates.length; i++) {
+let update = reply.updates[i];
+debug("Update: " + update.channelID + ": " + update.version);
+if (typeof update.channelID !== "string") {
+debug("Invalid update literal at index " + i);
+continue;
+}
+if (update.version === undefined) {
+debug("update.version does not exist");
+continue;
+}
+let version = update.version;
+if (typeof version === "string") {
+version = parseInt(version, 10);
+}
+if (typeof version === "number" && version >= 0) {
+// FIXME(nsm): this relies on app update notification being infallible!
+// eventually fix this
+this._receivedUpdate(update.channelID, version);
+this._sendAck(update.channelID, version);
+}
+}
+},
+// FIXME(nsm): batch acks for efficiency reasons.
+_sendAck: function(channelID, version) {
+debug("sendAck()");
+this._send('ack', {
+updates: [{channelID: channelID, version: version}]
+});
+},
+/*
+* Must be used only by request/response style calls over the websocket.
+*/
+_sendRequest: function(action, data) {
+debug("sendRequest() " + action);
+if (typeof data.channelID !== "string") {
+debug("Received non-string channelID");
+return Promise.reject("Received non-string channelID");
+}
+let deferred = Promise.defer();
+if (Object.keys(this._pendingRequests).length == 0) {
+// start the timer since we now have at least one request
+if (!this._requestTimeoutTimer)
+this._requestTimeoutTimer = Cc["@mozilla.org/timer;1"]
+.createInstance(Ci.nsITimer);
+this._requestTimeoutTimer.init(this,
+this._requestTimeout,
+Ci.nsITimer.TYPE_REPEATING_SLACK);
+}
+this._pendingRequests[data.channelID] = { deferred: deferred,
+ctime: Date.now() };
+this._send(action, data);
+return deferred.promise;
+},
+_send: function(action, data) {
+debug("send()");
+this._requestQueue.push([action, data]);
+debug("Queued " + action);
+this._processNextRequestInQueue();
+},
+_processNextRequestInQueue: function() {
+debug("_processNextRequestInQueue()");
+if (this._requestQueue.length == 0) {
+debug("Request queue empty");
+return;
+}
+if (this._currentState != STATE_READY) {
+if (!this._ws) {
+// This will end up calling processNextRequestInQueue().
+this._beginWSSetup();
+}
+else {
+// We have a socket open so we are just waiting for hello to finish.
+// That will call processNextRequestInQueue().
+}
+return;
+}
+let [action, data] = this._requestQueue.shift();
+data.messageType = action;
+if (!this._ws) {
+// If our websocket is not ready and our state is STATE_READY we may as
+// well give up all assumptions about the world and start from scratch
+// again.  Discard the message itself, let the timeout notify error to
+// the app.
+debug("This should never happen!");
+this._shutdownWS();
+}
+this._wsSendMessage(data);
+// Process the next one as soon as possible.
+setTimeout(this._processNextRequestInQueue.bind(this), 0);
+},
+_receivedUpdate: function(aChannelID, aLatestVersion) {
+debug("Updating: " + aChannelID + " -> " + aLatestVersion);
+let compareRecordVersionAndNotify = function(aPushRecord) {
+debug("compareRecordVersionAndNotify()");
+if (!aPushRecord) {
+debug("No record for channel ID " + aChannelID);
+return;
+}
+if (aPushRecord.version == null ||
+aPushRecord.version < aLatestVersion) {
+debug("Version changed, notifying app and updating DB");
+aPushRecord.version = aLatestVersion;
+this._notifyApp(aPushRecord);
+this._updatePushRecord(aPushRecord)
+.then(
+null,
+function(e) {
+debug("Error updating push record");
+}
+);
+}
+else {
+debug("No significant version change: " + aLatestVersion);
+}
+}
+let recoverNoSuchChannelID = function(aChannelIDFromServer) {
+debug("Could not get channelID " + aChannelIDFromServer + " from DB");
+}
+this._db.getByChannelID(aChannelID,
+compareRecordVersionAndNotify.bind(this),
+recoverNoSuchChannelID.bind(this));
+},
+// Fires a push-register system message to all applications that have
+// registrations.
+_notifyAllAppsRegister: function() {
+debug("notifyAllAppsRegister()");
+let deferred = Promise.defer();
+// records are objects describing the registrations as stored in IndexedDB.
+function wakeupRegisteredApps(records) {
+// Pages to be notified.
+// wakeupTable[manifestURL] -> [ pageURL ]
+let wakeupTable = {};
+for (let i = 0; i < records.length; i++) {
+let record = records[i];
+if (!(record.manifestURL in wakeupTable))
+wakeupTable[record.manifestURL] = [];
+wakeupTable[record.manifestURL].push(record.pageURL);
+}
+let messenger = Cc["@mozilla.org/system-message-internal;1"]
+.getService(Ci.nsISystemMessagesInternal);
+for (let manifestURL in wakeupTable) {
+wakeupTable[manifestURL].forEach(function(pageURL) {
+messenger.sendMessage('push-register', {},
+Services.io.newURI(pageURL, null, null),
+Services.io.newURI(manifestURL, null, null));
+});
+}
+deferred.resolve();
+}
+this._db.getAllChannelIDs(wakeupRegisteredApps, deferred.reject);
+return deferred.promise;
+},
+_notifyApp: function(aPushRecord) {
+if (!aPushRecord || !aPushRecord.pageURL || !aPushRecord.manifestURL) {
+debug("notifyApp() something is undefined.  Dropping notification");
+return;
+}
+debug("notifyApp() " + aPushRecord.pageURL +
+"  " + aPushRecord.manifestURL);
+let pageURI = Services.io.newURI(aPushRecord.pageURL, null, null);
+let manifestURI = Services.io.newURI(aPushRecord.manifestURL, null, null);
+let message = {
+pushEndpoint: aPushRecord.pushEndpoint,
+version: aPushRecord.version
+};
+let messenger = Cc["@mozilla.org/system-message-internal;1"]
+.getService(Ci.nsISystemMessagesInternal);
+messenger.sendMessage('push', message, pageURI, manifestURI);
+},
+_updatePushRecord: function(aPushRecord) {
+debug("updatePushRecord()");
+let deferred = Promise.defer();
+this._db.put(aPushRecord, deferred.resolve, deferred.reject);
+return deferred.promise;
+},
+_dropRegistrations: function() {
+let deferred = Promise.defer();
+this._db.drop(deferred.resolve, deferred.reject);
+return deferred.promise;
+},
+receiveMessage: function(aMessage) {
+debug("receiveMessage(): " + aMessage.name);
+if (kCHILD_PROCESS_MESSAGES.indexOf(aMessage.name) == -1) {
+debug("Invalid message from child " + aMessage.name);
+return;
+}
+let mm = aMessage.target.QueryInterface(Ci.nsIMessageSender);
+let json = aMessage.data;
+this[aMessage.name.slice("Push:".length).toLowerCase()](json, mm);
+},
+/**
+* Called on message from the child process. aPageRecord is an object sent by
+* navigator.push, identifying the sending page and other fields.
+*/
+register: function(aPageRecord, aMessageManager) {
+debug("register()");
+let uuidGenerator = Cc["@mozilla.org/uuid-generator;1"]
+.getService(Ci.nsIUUIDGenerator);
+// generateUUID() gives a UUID surrounded by {...}, slice them off.
+let channelID = uuidGenerator.generateUUID().toString().slice(1, -1);
+this._sendRequest("register", {channelID: channelID})
+.then(
+this._onRegisterSuccess.bind(this, aPageRecord, channelID),
+this._onRegisterError.bind(this, aPageRecord, aMessageManager)
+)
+.then(
+function(message) {
+aMessageManager.sendAsyncMessage("PushService:Register:OK", message);
+},
+function(message) {
+aMessageManager.sendAsyncMessage("PushService:Register:KO", message);
+});
+},
+/**
+* Exceptions thrown in _onRegisterSuccess are caught by the promise obtained
+* from _sendRequest, causing the promise to be rejected instead.
+*/
+_onRegisterSuccess: function(aPageRecord, generatedChannelID, data) {
+debug("_onRegisterSuccess()");
+let deferred = Promise.defer();
+let message = { requestID: aPageRecord.requestID };
+if (typeof data.channelID !== "string") {
+debug("Invalid channelID " + message);
+message["error"] = "Invalid channelID received";
+throw message;
+}
+else if (data.channelID != generatedChannelID) {
+debug("Server replied with different channelID " + data.channelID +
+" than what UA generated " + generatedChannelID);
+message["error"] = "Server sent 200 status code but different channelID";
+throw message;
+}
+try {
+Services.io.newURI(data.pushEndpoint, null, null);
+}
+catch (e) {
+debug("Invalid pushEndpoint " + data.pushEndpoint);
+message["error"] = "Invalid pushEndpoint " + data.pushEndpoint;
+throw message;
+}
+let record = {
+channelID: data.channelID,
+pushEndpoint: data.pushEndpoint,
+pageURL: aPageRecord.pageURL,
+manifestURL: aPageRecord.manifestURL,
+version: null
+};
+this._updatePushRecord(record)
+.then(
+function() {
+message["pushEndpoint"] = data.pushEndpoint;
+deferred.resolve(message);
+},
+function(error) {
+// Unable to save.
+this._send("unregister", {channelID: record.channelID});
+message["error"] = error;
+deferred.reject(message);
+}
+);
+return deferred.promise;
+},
+/**
+* Exceptions thrown in _onRegisterError are caught by the promise obtained
+* from _sendRequest, causing the promise to be rejected instead.
+*/
+_onRegisterError: function(aPageRecord, aMessageManager, reply) {
+debug("_onRegisterError()");
+if (!reply.error) {
+debug("Called without valid error message!");
+}
+throw { requestID: aPageRecord.requestID, error: reply.error };
+},
+/**
+* Called on message from the child process.
+*
+* Why is the record being deleted from the local database before the server
+* is told?
+*
+* Unregistration is for the benefit of the app and the AppServer
+* so that the AppServer does not keep pinging a channel the UserAgent isn't
+* watching The important part of the transaction in this case is left to the
+* app, to tell its server of the unregistration.  Even if the request to the
+* PushServer were to fail, it would not affect correctness of the protocol,
+* and the server GC would just clean up the channelID eventually.  Since the
+* appserver doesn't ping it, no data is lost.
+*
+* If rather we were to unregister at the server and update the database only
+* on success: If the server receives the unregister, and deletes the
+* channelID, but the response is lost because of network failure, the
+* application is never informed. In addition the application may retry the
+* unregister when it fails due to timeout at which point the server will say
+* it does not know of this unregistration.  We'll have to make the
+* registration/unregistration phases have retries and attempts to resend
+* messages from the server, and have the client acknowledge. On a server,
+* data is cheap, reliable notification is not.
+*/
+unregister: function(aPageRecord, aMessageManager) {
+debug("unregister()");
+let fail = function(error) {
+debug("unregister() fail() error " + error);
+let message = {requestID: aPageRecord.requestID, error: error};
+aMessageManager.sendAsyncMessage("PushService:Unregister:KO", message);
+}
+this._db.getByPushEndpoint(aPageRecord.pushEndpoint, function(record) {
+// If the endpoint didn't exist, let's just fail.
+if (record === undefined) {
+fail("NotFoundError");
+return;
+}
+// Non-owner tried to unregister, say success, but don't do anything.
+if (record.manifestURL !== aPageRecord.manifestURL) {
+aMessageManager.sendAsyncMessage("PushService:Unregister:OK", {
+requestID: aPageRecord.requestID,
+pushEndpoint: aPageRecord.pushEndpoint
+});
+return;
+}
+this._db.delete(record.channelID, function() {
+// Let's be nice to the server and try to inform it, but we don't care
+// about the reply.
+this._send("unregister", {channelID: record.channelID});
+aMessageManager.sendAsyncMessage("PushService:Unregister:OK", {
+requestID: aPageRecord.requestID,
+pushEndpoint: aPageRecord.pushEndpoint
+});
+}.bind(this), fail);
+}.bind(this), fail);
+},
+/**
+* Called on message from the child process
+*/
+registrations: function(aPageRecord, aMessageManager) {
+debug("registrations()");
+if (aPageRecord.manifestURL) {
+this._db.getAllByManifestURL(aPageRecord.manifestURL,
+this._onRegistrationsSuccess.bind(this, aPageRecord, aMessageManager),
+this._onRegistrationsError.bind(this, aPageRecord, aMessageManager));
+}
+else {
+this._onRegistrationsError(aPageRecord, aMessageManager);
+}
+},
+_onRegistrationsSuccess: function(aPageRecord,
+aMessageManager,
+pushRecords) {
+let registrations = [];
+pushRecords.forEach(function(pushRecord) {
+registrations.push({
+__exposedProps__: { pushEndpoint: 'r', version: 'r' },
+pushEndpoint: pushRecord.pushEndpoint,
+version: pushRecord.version
+});
+});
+aMessageManager.sendAsyncMessage("PushService:Registrations:OK", {
+requestID: aPageRecord.requestID,
+registrations: registrations
+});
+},
+_onRegistrationsError: function(aPageRecord, aMessageManager) {
+aMessageManager.sendAsyncMessage("PushService:Registrations:KO", {
+requestID: aPageRecord.requestID,
+error: "Database error"
+});
+},
+// begin Push protocol handshake
+_wsOnStart: function(context) {
+debug("wsOnStart()");
+if (this._currentState != STATE_WAITING_FOR_WS_START) {
+debug("NOT in STATE_WAITING_FOR_WS_START. Current state " +
+this._currentState + ". Skipping");
+return;
+}
+// Since we've had a successful connection reset the retry fail count.
+this._retryFailCount = 0;
+// Openning an available UDP port.
+this._listenForUDPWakeup();
+let data = {
+messageType: "hello",
+}
+if (this._UAID)
+data["uaid"] = this._UAID;
+let networkState = this._getNetworkState();
+if (networkState.ip) {
+// Hostport is apparently a thing.
+data["wakeup_hostport"] = {
+ip: networkState.ip,
+port: this._udpServer && this._udpServer.port
+};
+data["mobilenetwork"] = {
+mcc: networkState.mcc,
+mnc: networkState.mnc
+};
+}
+function sendHelloMessage(ids) {
+// On success, ids is an array, on error its not.
+data["channelIDs"] = ids.map ?
+ids.map(function(el) { return el.channelID; }) : [];
+this._wsSendMessage(data);
+this._currentState = STATE_WAITING_FOR_HELLO;
+}
+this._db.getAllChannelIDs(sendHelloMessage.bind(this),
+sendHelloMessage.bind(this));
+},
+/**
+* This statusCode is not the websocket protocol status code, but the TCP
+* connection close status code.
+*
+* If we do not explicitly call ws.close() then statusCode is always
+* NS_BASE_STREAM_CLOSED, even on a successful close.
+*/
+_wsOnStop: function(context, statusCode) {
+debug("wsOnStop()");
+if (statusCode != Cr.NS_OK &&
+!(statusCode == Cr.NS_BASE_STREAM_CLOSED && this._willBeWokenUpByUDP)) {
+debug("Socket error " + statusCode);
+this._reconnectAfterBackoff();
+}
+// Bug 896919. We always shutdown the WebSocket, even if we need to
+// reconnect. This works because _reconnectAfterBackoff() is "async"
+// (there is a minimum delay of the pref retryBaseInterval, which by default
+// is 5000ms), so that function will open the WebSocket again.
+this._shutdownWS();
+},
+_wsOnMessageAvailable: function(context, message) {
+debug("wsOnMessageAvailable() " + message);
+this._waitingForPong = false;
+// Reset the ping timer.  Note: This path is executed at every step of the
+// handshake, so this alarm does not need to be set explicitly at startup.
+this._setAlarm(prefs.get("pingInterval"));
+let reply = undefined;
+try {
+reply = JSON.parse(message);
+} catch(e) {
+debug("Parsing JSON failed. text : " + message);
+return;
+}
+if (typeof reply.messageType != "string") {
+debug("messageType not a string " + reply.messageType);
+return;
+}
+// A whitelist of protocol handlers. Add to these if new messages are added
+// in the protocol.
+let handlers = ["Hello", "Register", "Notification"];
+// Build up the handler name to call from messageType.
+// e.g. messageType == "register" -> _handleRegisterReply.
+let handlerName = reply.messageType[0].toUpperCase() +
+reply.messageType.slice(1).toLowerCase();
+if (handlers.indexOf(handlerName) == -1) {
+debug("No whitelisted handler " + handlerName + ". messageType: " +
+reply.messageType);
+return;
+}
+let handler = "_handle" + handlerName + "Reply";
+if (typeof this[handler] !== "function") {
+debug("Handler whitelisted but not implemented! " + handler);
+return;
+}
+this[handler](reply);
+},
+/**
+* The websocket should never be closed. Since we don't call ws.close(),
+* _wsOnStop() receives error code NS_BASE_STREAM_CLOSED (see comment in that
+* function), which calls reconnect and re-establishes the WebSocket
+* connection.
+*
+* If the server said it'll use UDP for wakeup, we set _willBeWokenUpByUDP
+* and stop reconnecting in _wsOnStop().
+*/
+_wsOnServerClose: function(context, aStatusCode, aReason) {
+debug("wsOnServerClose() " + aStatusCode + " " + aReason);
+// Switch over to UDP.
+if (aStatusCode == kUDP_WAKEUP_WS_STATUS_CODE) {
+debug("Server closed with promise to wake up");
+this._willBeWokenUpByUDP = true;
+// TODO: there should be no pending requests
+}
+},
+_listenForUDPWakeup: function() {
+debug("listenForUDPWakeup()");
+if (this._udpServer) {
+debug("UDP Server already running");
+return;
+}
+if (!this._getNetworkState().ip) {
+debug("No IP");
+return;
+}
+if (!prefs.get("udp.wakeupEnabled")) {
+debug("UDP support disabled");
+return;
+}
+this._udpServer = Cc["@mozilla.org/network/udp-socket;1"]
+.createInstance(Ci.nsIUDPSocket);
+this._udpServer.init(-1, false);
+this._udpServer.asyncListen(this);
+debug("listenForUDPWakeup listening on " + this._udpServer.port);
+return this._udpServer.port;
+},
+/**
+* Called by UDP Server Socket. As soon as a ping is recieved via UDP,
+* reconnect the WebSocket and get the actual data.
+*/
+onPacketReceived: function(aServ, aMessage) {
+debug("Recv UDP datagram on port: " + this._udpServer.port);
+this._beginWSSetup();
+},
+/**
+* Called by UDP Server Socket if the socket was closed for some reason.
+*
+* If this happens, we reconnect the WebSocket to not miss out on
+* notifications.
+*/
+onStopListening: function(aServ, aStatus) {
+debug("UDP Server socket was shutdown. Status: " + aStatus);
+this._udpServer = undefined;
+this._beginWSSetup();
+},
+/**
+* Get mobile network information to decide if the client is capable of being
+* woken up by UDP (which currently just means having an mcc and mnc along
+* with an IP).
+*/
+_getNetworkState: function() {
+debug("getNetworkState()");
+try {
+if (!prefs.get("udp.wakeupEnabled")) {
+debug("UDP support disabled, we do not send any carrier info");
+throw "UDP disabled";
+}
+let nm = Cc["@mozilla.org/network/manager;1"].getService(Ci.nsINetworkManager);
+if (nm.active && nm.active.type == Ci.nsINetworkInterface.NETWORK_TYPE_MOBILE) {
+let icc = Cc["@mozilla.org/ril/content-helper;1"].getService(Ci.nsIIccProvider);
+// TODO: Bug 927721 - PushService for multi-sim
+// In Multi-sim, there is more than one client in iccProvider. Each
+// client represents a icc service. To maintain backward compatibility
+// with single sim, we always use client 0 for now. Adding support
+// for multiple sim will be addressed in bug 927721, if needed.
+let clientId = 0;
+let iccInfo = icc.getIccInfo(clientId);
+if (iccInfo) {
+debug("Running on mobile data");
+return {
+mcc: iccInfo.mcc,
+mnc: iccInfo.mnc,
+ip:  nm.active.ip
+}
+}
+}
+} catch (e) {}
+debug("Running on wifi");
+return {
+mcc: 0,
+mnc: 0,
+ip: undefined
+};
+},
+// utility function used to add/remove observers in init() and shutdown()
+_getNetworkStateChangeEventName: function() {
+try {
+Cc["@mozilla.org/network/manager;1"].getService(Ci.nsINetworkManager);
+return "network-active-changed";
+} catch (e) {
+return "network:offline-status-changed";
+}
+}
+}

The Tor Browser / file comparison

comparison: dom/push/src/PushService.jsm

dom/push/src/PushService.jsm