Bug 578671 - Sync which engines are enabled across clients, wipe data for disabled engines [r=mconnor]

services/sync/modules/engines.js

@@ -406,8 +406,7 @@ SyncEngine.prototype = {
 
     // Delete any existing data and reupload on bad version or missing meta
     if (meta == null) {
-      new Resource(this.engineURL).delete();
-      this._resetClient();
+      this.wipeServer(true);
 
       // Generate a new crypto record
       let symkey = Svc.Crypto.generateRandomKey();
@@ -783,5 +782,12 @@ SyncEngine.prototype = {
   _resetClient: function SyncEngine__resetClient() {
     this.resetLastSync();
     this.toFetch = [];
+  },
+
+  wipeServer: function wipeServer(ignoreCrypto) {
+    new Resource(this.engineURL).delete();
+    if (!ignoreCrypto)
+      new Resource(this.cryptoMetaURL).delete();
+    this._resetClient();
   }
 };

services/sync/modules/ext/Preferences.js

@@ -499,11 +499,11 @@ PrefObserver.prototype = {
     // The pref service only observes whole branches, but we only observe
     // individual preferences, so we check here that the pref that changed
     // is the exact one we're observing (and not some sub-pref on the branch).
-    if (data != this.prefName)
+    if (data.indexOf(this.prefName) != 0)
       return;
 
     if (typeof this.callback == "function") {
-      let prefValue = Preferences.get(this.prefName);
+      let prefValue = Preferences.get(data);
 
       if (this.thisObject)
         this.callback.call(this.thisObject, prefValue);

services/sync/modules/service.js

@@ -262,6 +262,7 @@ WeaveSvc.prototype = {
     Svc.Obs.add("weave:service:backoff:interval", this);
     Svc.Obs.add("weave:engine:score:updated", this);
     Svc.Obs.add("weave:resource:status:401", this);
+    Svc.Prefs.observe("engine.", this);
 
     if (!this.enabled)
       this._log.info("Weave Sync disabled");
@@ -419,6 +420,12 @@ WeaveSvc.prototype = {
         this._idleTime = 0;
         Utils.delay(function() this.sync(false), 0, this);
         break;
+      case "nsPref:changed":
+        if (this._ignorePrefObserver)
+          return;
+        let engine = data.slice((PREFS_BRANCH + "engine.").length);
+        this._handleEngineStatusChanged(engine);
+        break;
     }
   },
 
@@ -439,6 +446,17 @@ WeaveSvc.prototype = {
     this._checkSyncStatus();
   },
 
+  _handleEngineStatusChanged: function handleEngineDisabled(engine) {
+    this._log.trace("Status for " + engine + " engine changed.");
+    if (Svc.Prefs.get("engineStatusChanged." + engine, false)) {
+      // The enabled status being changed back to what it was before.
+      Svc.Prefs.reset("engineStatusChanged." + engine);
+    } else {
+      // Remember that the engine status changed locally until the next sync.
+      Svc.Prefs.set("engineStatusChanged." + engine, true);
+    }
+  },
+
   _handleResource401: function _handleResource401(request) {
     // Only handle 401s that are hitting the current cluster
     let spec = request.resource.spec;
@@ -676,7 +694,9 @@ WeaveSvc.prototype = {
     // Reset all engines
     this.resetClient();
     // Reset Weave prefs
+    this._ignorePrefObserver = true;
     Svc.Prefs.resetBranch("");
+    this._ignorePrefObserver = false;
     // set lastversion pref
     Svc.Prefs.set("lastversion", WEAVE_VERSION);
     // Find weave logins and remove them.
@@ -1392,8 +1412,9 @@ WeaveSvc.prototype = {
       }
     }
 
-    // Update the client mode now because it might change what we sync
+    // Update the client mode and engines because it might change what we sync.
     this._updateClientMode();
+    this._updateEnabledEngines();
 
     try {
       for each (let engine in Engines.getEnabled()) {
@@ -1445,6 +1466,56 @@ WeaveSvc.prototype = {
     }
   },
 
+  _updateEnabledEngines: function _updateEnabledEngines() {
+    let meta = Records.get(this.metaURL);
+    if (!meta.payload.engines)
+      return;
+
+    this._ignorePrefObserver = true;
+
+    let enabled = [eng.name for each (eng in Engines.getEnabled())];
+    for (let engineName in meta.payload.engines) {
+      let index = enabled.indexOf(engineName);
+      if (index != -1) {
+        // The engine is enabled locally. Nothing to do.
+        enabled.splice(index, 1);
+        continue;
+      }
+      let engine = Engines.get(engineName);
+      if (!engine) {
+        // The engine doesn't exist locally. Nothing to do.
+        continue;
+      }
+
+      if (Svc.Prefs.get("engineStatusChanged." + engineName, false)) {
+        // The engine was disabled locally. Wipe server data and
+        // disable it everywhere.
+        this._log.trace("Wiping data for " + engineName + " engine.");
+        engine.wipeServer();
+        delete meta.payload.engines[engineName];
+        meta.changed = true;
+        Svc.Prefs.reset("engineStatusChanged." + engineName);
+      } else {
+        // The engine was enabled remotely. Enable it locally.
+        this._log.trace(engineName + " engine was enabled remotely.");
+        engine.enabled = true;
+      }
+    }
+
+    // Any remaining engines were either enabled locally or disabled remotely.
+    for each (engineName in enabled) {
+      if (Svc.Prefs.get("engineStatusChanged." + engineName, false)) {
+        this._log.trace("The " + engineName + " engine was enabled locally.");
+        Svc.Prefs.reset("engineStatusChanged." + engineName);
+      } else {
+        this._log.trace("The " + engineName + " engine was disabled remotely.");
+        Engines.get(engineName).enabled = false;
+      }
+    }
+
+    this._ignorePrefObserver = false;
+  },
+
   // returns true if sync should proceed
   // false / no return value means sync should be aborted
   _syncEngine: function WeaveSvc__syncEngine(engine) {

services/sync/tests/unit/head_http_server.js

@@ -36,7 +36,33 @@ function readBytesFromInputStream(inputStream, count) {
   return new BinaryInputStream(inputStream).readBytes(count);
 }
 
+/*
+ * Create and upload public + private key pair. You probably want to enable
+ * FakeCryptoService first, otherwise this will be very expensive.
+ */
+function createAndUploadKeypair() {
+  let storageURL = Svc.Prefs.get("clusterURL") + Svc.Prefs.get("storageAPI")
+                   + "/" + ID.get("WeaveID").username + "/storage/";
 
+  PubKeys.defaultKeyUri = storageURL + "keys/pubkey";
+  PrivKeys.defaultKeyUri = storageURL + "keys/privkey";
+  let keys = PubKeys.createKeypair(ID.get("WeaveCryptoID"),
+                                   PubKeys.defaultKeyUri,
+                                   PrivKeys.defaultKeyUri);
+  PubKeys.uploadKeypair(keys);
+}
+
+/*
+ * Create and upload an engine's symmetric key.
+ */
+function createAndUploadSymKey(url) {
+  let symkey = Svc.Crypto.generateRandomKey();
+  let pubkey = PubKeys.getDefaultKey();
+  let meta = new CryptoMeta(url);
+  meta.addUnwrappedKey(pubkey, symkey);
+  let res = new Resource(meta.uri);
+  res.put(meta);
+}
 
 /*
  * Represent a WBO on the server

services/sync/tests/unit/test_service_sync_updateEnabledEngines.js

@@ -0,0 +1,206 @@
+Cu.import("resource://services-sync/service.js");
+Cu.import("resource://services-sync/engines.js");
+Cu.import("resource://services-sync/util.js");
+Cu.import("resource://services-sync/constants.js");
+Cu.import("resource://services-sync/base_records/crypto.js");
+Cu.import("resource://services-sync/base_records/keys.js");
+Cu.import("resource://services-sync/base_records/wbo.js");
+
+function SteamEngine() {
+  SyncEngine.call(this, "Steam");
+}
+SteamEngine.prototype = {
+  __proto__: SyncEngine.prototype,
+  // We're not interested in engine sync but what the service does.
+  _sync: function _sync() {
+    this._syncStartup();
+  }
+};
+Engines.register(SteamEngine);
+
+function sync_httpd_setup(handlers) {
+  handlers["/1.0/johndoe/info/collections"]
+      = (new ServerWBO("collections", {})).handler(),
+  handlers["/1.0/johndoe/storage/keys/pubkey"]
+      = (new ServerWBO("pubkey")).handler();
+  handlers["/1.0/johndoe/storage/keys/privkey"]
+      = (new ServerWBO("privkey")).handler();
+  handlers["/1.0/johndoe/storage/clients"]
+      = (new ServerCollection()).handler();
+  handlers["/1.0/johndoe/storage/crypto"]
+      = (new ServerCollection()).handler();
+  handlers["/1.0/johndoe/storage/crypto/clients"]
+      = (new ServerWBO("clients", {})).handler();
+  return httpd_setup(handlers);
+}
+
+function setUp() {
+  Service.username = "johndoe";
+  Service.password = "ilovejane";
+  Service.passphrase = "sekrit";
+  Service.clusterURL = "http://localhost:8080/";
+  new FakeCryptoService();
+  createAndUploadKeypair();
+}
+
+const PAYLOAD = 42;
+
+function test_enabledLocally() {
+  _("Test: Engine is disabled on remote clients and enabled locally");
+  Service.syncID = "abcdefghij";
+  let engine = Engines.get("steam");
+  let metaWBO = new ServerWBO("global", {syncID: Service.syncID,
+                                         storageVersion: STORAGE_VERSION,
+                                         engines: {}});
+  let server = sync_httpd_setup({
+    "/1.0/johndoe/storage/meta/global": metaWBO.handler(),
+    "/1.0/johndoe/storage/crypto/steam": new ServerWBO("steam", {}).handler(),
+    "/1.0/johndoe/storage/steam": new ServerWBO("steam", {}).handler()
+  });
+  do_test_pending();
+  setUp();
+
+  try {
+    _("Enable engine locally.");
+    engine.enabled = true;
+
+    _("Sync.");
+    Weave.Service.login();
+    Weave.Service.sync();
+
+    _("Meta record now contains the new engine.");
+    do_check_true(!!metaWBO.data.engines.steam);
+
+    _("Engine continues to be enabled.");
+    do_check_true(engine.enabled);
+  } finally {
+    server.stop(do_test_finished);
+    Service.startOver();
+  }
+}
+
+function test_disabledLocally() {
+  _("Test: Engine is enabled on remote clients and disabled locally");
+  Service.syncID = "abcdefghij";
+  let engine = Engines.get("steam");
+  let metaWBO = new ServerWBO("global", {
+    syncID: Service.syncID,
+    storageVersion: STORAGE_VERSION,
+    engines: {steam: {syncID: engine.syncID,
+                      version: engine.version}}
+  });
+  let steamCrypto = new ServerWBO("steam", PAYLOAD);
+  let steamCollection = new ServerWBO("steam", PAYLOAD);
+  let server = sync_httpd_setup({
+    "/1.0/johndoe/storage/meta/global": metaWBO.handler(),
+    "/1.0/johndoe/storage/crypto/steam": steamCrypto.handler(),
+    "/1.0/johndoe/storage/steam": steamCollection.handler()
+  });
+  do_test_pending();
+  setUp();
+
+  try {
+    _("Disable engine locally.");
+    Service._ignorePrefObserver = true;
+    engine.enabled = true;
+    Service._ignorePrefObserver = false;
+    engine.enabled = false;
+
+    _("Sync.");
+    Weave.Service.login();
+    Weave.Service.sync();
+
+    _("Meta record no longer contains engine.");
+    do_check_false(!!metaWBO.data.engines.steam);
+
+    _("Server records are wiped.");
+    do_check_eq(steamCollection.payload, undefined);
+    do_check_eq(steamCrypto.payload, undefined);
+
+    _("Engine continues to be disabled.");
+    do_check_false(engine.enabled);
+  } finally {
+    server.stop(do_test_finished);
+    Service.startOver();
+  }
+}
+
+function test_enabledRemotely() {
+  _("Test: Engine is disabled locally and enabled on a remote client");
+  Service.syncID = "abcdefghij";
+  let engine = Engines.get("steam");
+  let metaWBO = new ServerWBO("global", {
+    syncID: Service.syncID,
+    storageVersion: STORAGE_VERSION,
+    engines: {steam: {syncID: engine.syncID,
+                      version: engine.version}}
+  });
+  let server = sync_httpd_setup({
+    "/1.0/johndoe/storage/meta/global": metaWBO.handler(),
+    "/1.0/johndoe/storage/crypto/steam": new ServerWBO("steam", {}).handler(),
+    "/1.0/johndoe/storage/steam": new ServerWBO("steam", {}).handler()
+  });
+  do_test_pending();
+  setUp();
+
+  try {
+    _("Engine is disabled.");
+    do_check_false(engine.enabled);
+
+    _("Sync.");
+    Weave.Service.login();
+    Weave.Service.sync();
+
+    _("Engine is enabled.");
+    do_check_true(engine.enabled);
+
+    _("Meta record still present.");
+    do_check_eq(metaWBO.data.engines.steam.syncID, engine.syncID);
+  } finally {
+    server.stop(do_test_finished);
+    Service.startOver();
+  }
+}
+
+function test_disabledRemotely() {
+  _("Test: Engine is enabled locally and disabled on a remote client");
+  Service.syncID = "abcdefghij";
+  let engine = Engines.get("steam");
+  let metaWBO = new ServerWBO("global", {syncID: Service.syncID,
+                                         storageVersion: STORAGE_VERSION,
+                                         engines: {}});
+  let server = sync_httpd_setup({
+    "/1.0/johndoe/storage/meta/global": metaWBO.handler(),
+    "/1.0/johndoe/storage/crypto/steam": new ServerWBO("steam", {}).handler(),
+    "/1.0/johndoe/storage/steam": new ServerWBO("steam", {}).handler()
+  });
+  do_test_pending();
+  setUp();
+
+  try {
+    _("Enable engine locally.");
+    Service._ignorePrefObserver = true;
+    engine.enabled = true;
+    Service._ignorePrefObserver = false;
+
+    _("Sync.");
+    Weave.Service.login();
+    Weave.Service.sync();
+
+    _("Engine is disabled.");
+    do_check_false(engine.enabled);
+
+    _("Meta record isn't uploaded.");
+    do_check_false(!!metaWBO.data.engines.steam);
+  } finally {
+    server.stop(do_test_finished);
+    Service.startOver();
+  }
+}
+
+function run_test() {
+  test_enabledLocally();
+  test_disabledLocally();
+  test_enabledRemotely();
+  test_disabledRemotely();
+}

services/sync/tests/unit/test_syncengine.js

@@ -112,10 +112,48 @@ function test_resetClient() {
   }
 }
 
+function test_wipeServer() {
+  _("SyncEngine.wipeServer deletes server data and resets the client.");
+  Svc.Prefs.set("clusterURL", "http://localhost:8080/");
+  let engine = makeSteamEngine();
+
+  const PAYLOAD = 42;
+  let steamCrypto = new ServerWBO("steam", PAYLOAD);
+  let steamCollection = new ServerWBO("steam", PAYLOAD);
+  let server = httpd_setup({
+    "/1.0/foo/storage/crypto/steam": steamCrypto.handler(),
+    "/1.0/foo/storage/steam": steamCollection.handler()
+  });
+  do_test_pending();
+
+  try {
+    // Some data to reset.
+    engine.toFetch = [Utils.makeGUID(), Utils.makeGUID(), Utils.makeGUID()];
+    engine.lastSync = 123.45;
+
+    _("Wipe server data and reset client.");
+    engine.wipeServer(true);
+    do_check_eq(steamCollection.payload, undefined);
+    do_check_eq(engine.lastSync, 0);
+    do_check_eq(engine.toFetch.length, 0);
+
+    _("We passed a truthy arg earlier in which case it doesn't wipe the crypto collection.");
+    do_check_eq(steamCrypto.payload, PAYLOAD);
+    engine.wipeServer();
+    do_check_eq(steamCrypto.payload, undefined);
+
+  } finally {
+    server.stop(do_test_finished);
+    syncTesting = new SyncTestingInfrastructure(makeSteamEngine);
+    Svc.Prefs.resetBranch("");
+  }
+}
+
 function run_test() {
   test_url_attributes();
   test_syncID();
   test_lastSync();
   test_toFetch();
   test_resetClient();
+  test_wipeServer();
 }

services/sync/tests/unit/test_syncengine_sync.js

@@ -119,27 +119,6 @@ function sync_httpd_setup(handlers) {
   return httpd_setup(handlers);
 }
 
-function createAndUploadKeypair() {
-  let storageURL = Svc.Prefs.get("clusterURL") + Svc.Prefs.get("storageAPI")
-                   + "/" + ID.get("WeaveID").username + "/storage/";
-
-  PubKeys.defaultKeyUri = storageURL + "keys/pubkey";
-  PrivKeys.defaultKeyUri = storageURL + "keys/privkey";
-  let keys = PubKeys.createKeypair(ID.get("WeaveCryptoID"),
-                                   PubKeys.defaultKeyUri,
-                                   PrivKeys.defaultKeyUri);
-  PubKeys.uploadKeypair(keys);
-}
-
-function createAndUploadSymKey(url) {
-  let symkey = Svc.Crypto.generateRandomKey();
-  let pubkey = PubKeys.getDefaultKey();
-  let meta = new CryptoMeta(url);
-  meta.addUnwrappedKey(pubkey, symkey);
-  let res = new Resource(meta.uri);
-  res.put(meta);
-}
-
 // Turn WBO cleartext into "encrypted" payload as it goes over the wire
 function encryptPayload(cleartext) {
   if (typeof cleartext == "object") {