RustCrypto | Implement keybackup loop (#3652)

* Implement `CryptoApi.checkKeyBackup`

* Deprecate `MatrixClient.enableKeyBackup`.

* fix integ test

* more tests

* Implement keybackup loop

* cleaning

* update matrix-sdk-crypto-wasm to 1.2.1

* fix lint

* avoid real timer stuff

* Simplify test

* post merge lint fix

* revert change on yarn.lock

* code review

* Generate test data for exported keys

* code review cleaning

* cleanup legacy backup loop

* Update spec/test-utils/test-data/generate-test-data.py

Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

* Update spec/test-utils/test-data/generate-test-data.py

Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

* update yarn.lock for new wasm bindings

---------

Co-authored-by: Richard van der Hoff <richard@matrix.org>
Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

package.json

@@ -55,7 +55,7 @@
     ],
     "dependencies": {
         "@babel/runtime": "^7.12.5",
-        "@matrix-org/matrix-sdk-crypto-wasm": "^1.2.0",
+        "@matrix-org/matrix-sdk-crypto-wasm": "^1.2.1",
         "another-json": "^0.2.0",
         "bs58": "^5.0.0",
         "content-type": "^1.0.4",

spec/integ/crypto/megolm-backup.spec.ts

@@ -19,7 +19,7 @@ import "fake-indexeddb/auto";
 import { IDBFactory } from "fake-indexeddb";
 
 import { IKeyBackupSession } from "../../../src/crypto/keybackup";
-import { createClient, CryptoEvent, ICreateClientOpts, IEvent, MatrixClient } from "../../../src";
+import { createClient, CryptoEvent, ICreateClientOpts, IEvent, MatrixClient, TypedEventEmitter } from "../../../src";
 import { SyncResponder } from "../../test-utils/SyncResponder";
 import { E2EKeyReceiver } from "../../test-utils/E2EKeyReceiver";
 import { E2EKeyResponder } from "../../test-utils/E2EKeyResponder";
@@ -27,6 +27,7 @@ import { mockInitialApiRequests } from "../../test-utils/mockEndpoints";
 import { awaitDecryption, CRYPTO_BACKENDS, InitCrypto, syncPromise } from "../../test-utils/test-utils";
 import * as testData from "../../test-utils/test-data";
 import { KeyBackupInfo } from "../../../src/crypto-api/keybackup";
+import { IKeyBackup } from "../../../src/crypto/backup";
 
 const ROOM_ID = "!ROOM:ID";
 
@@ -83,6 +84,58 @@ afterEach(() => {
     indexedDB = new IDBFactory();
 });
 
+enum MockKeyUploadEvent {
+    KeyUploaded = "KeyUploaded",
+}
+
+type MockKeyUploadEventHandlerMap = {
+    [MockKeyUploadEvent.KeyUploaded]: (roomId: string, sessionId: string, backupVersion: string) => void;
+};
+
+/*
+ * Test helper. Returns an event emitter that will emit an event every time fetchmock sees a request to backup a key.
+ */
+function mockUploadEmitter(
+    expectedVersion: string,
+): TypedEventEmitter<MockKeyUploadEvent, MockKeyUploadEventHandlerMap> {
+    const emitter = new TypedEventEmitter();
+    fetchMock.put(
+        "path:/_matrix/client/v3/room_keys/keys",
+        (url, request) => {
+            const version = new URLSearchParams(new URL(url).search).get("version");
+            if (version != expectedVersion) {
+                return {
+                    status: 403,
+                    body: {
+                        current_version: expectedVersion,
+                        errcode: "M_WRONG_ROOM_KEYS_VERSION",
+                        error: "Wrong backup version.",
+                    },
+                };
+            }
+            const uploadPayload: IKeyBackup = JSON.parse(request.body?.toString() ?? "{}");
+            let count = 0;
+            for (const [roomId, value] of Object.entries(uploadPayload.rooms)) {
+                for (const sessionId of Object.keys(value.sessions)) {
+                    emitter.emit(MockKeyUploadEvent.KeyUploaded, roomId, sessionId, version);
+                    count++;
+                }
+            }
+            return {
+                status: 200,
+                body: {
+                    count: count,
+                    etag: "abcdefg",
+                },
+            };
+        },
+        {
+            overwriteRoutes: true,
+        },
+    );
+    return emitter;
+}
+
 describe.each(Object.entries(CRYPTO_BACKENDS))("megolm-keys backup (%s)", (backend: string, initCrypto: InitCrypto) => {
     // oldBackendOnly is an alternative to `it` or `test` which will skip the test if we are running against the
     // Rust backend. Once we have full support in the rust sdk, it will go away.
@@ -176,6 +229,255 @@ describe.each(Object.entries(CRYPTO_BACKENDS))("megolm-keys backup (%s)", (backe
         expect(event.getContent()).toEqual("testytest");
     });
 
+    describe("backupLoop", () => {
+        it("Alice should upload known keys when backup is enabled", async function () {
+            // 404 means that there is no active backup
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", 404);
+
+            aliceClient = await initTestClient();
+            const aliceCrypto = aliceClient.getCrypto()!;
+            await aliceClient.startClient();
+
+            // tell Alice to trust the dummy device that signed the backup
+            await waitForDeviceList();
+            await aliceCrypto.setDeviceVerified(testData.TEST_USER_ID, testData.TEST_DEVICE_ID);
+
+            // check that signalling is working
+            const remainingZeroPromise = new Promise<void>((resolve, reject) => {
+                aliceClient.on(CryptoEvent.KeyBackupSessionsRemaining, (remaining) => {
+                    if (remaining == 0) {
+                        resolve();
+                    }
+                });
+            });
+
+            const someRoomKeys = testData.MEGOLM_SESSION_DATA_ARRAY;
+
+            const uploadMockEmitter = mockUploadEmitter(testData.SIGNED_BACKUP_DATA.version!);
+
+            const uploadPromises = someRoomKeys.map((data) => {
+                new Promise<void>((resolve) => {
+                    uploadMockEmitter.on(MockKeyUploadEvent.KeyUploaded, (roomId, sessionId, version) => {
+                        if (
+                            data.room_id == roomId &&
+                            data.session_id == sessionId &&
+                            version == testData.SIGNED_BACKUP_DATA.version
+                        ) {
+                            resolve();
+                        }
+                    });
+                });
+            });
+
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", testData.SIGNED_BACKUP_DATA, {
+                overwriteRoutes: true,
+            });
+
+            const result = await aliceCrypto.checkKeyBackupAndEnable();
+            expect(result).toBeTruthy();
+
+            await aliceCrypto.importRoomKeys(someRoomKeys);
+
+            // The backup loop is waiting a random amount of time to avoid different clients firing at the same time.
+            jest.runAllTimers();
+
+            await Promise.all(uploadPromises);
+
+            // Wait until all keys are backed up to ensure that when a new key is received the loop is restarted
+            await remainingZeroPromise;
+
+            // A new key import should trigger a new upload.
+            const newKey = testData.MEGOLM_SESSION_DATA;
+
+            const newKeyUploadPromise = new Promise<void>((resolve) => {
+                uploadMockEmitter.on(MockKeyUploadEvent.KeyUploaded, (roomId, sessionId, version) => {
+                    if (
+                        newKey.room_id == roomId &&
+                        newKey.session_id == sessionId &&
+                        version == testData.SIGNED_BACKUP_DATA.version
+                    ) {
+                        resolve();
+                    }
+                });
+            });
+
+            await aliceCrypto.importRoomKeys([newKey]);
+
+            jest.runAllTimers();
+            await newKeyUploadPromise;
+        });
+
+        it("Alice should re-upload all keys if a new trusted backup is available", async function () {
+            aliceClient = await initTestClient();
+            const aliceCrypto = aliceClient.getCrypto()!;
+            await aliceClient.startClient();
+
+            // tell Alice to trust the dummy device that signed the backup
+            await waitForDeviceList();
+            await aliceCrypto.setDeviceVerified(testData.TEST_USER_ID, testData.TEST_DEVICE_ID);
+
+            // check that signalling is working
+            const remainingZeroPromise = new Promise<void>((resolve) => {
+                aliceClient.on(CryptoEvent.KeyBackupSessionsRemaining, (remaining) => {
+                    if (remaining == 0) {
+                        resolve();
+                    }
+                });
+            });
+
+            const someRoomKeys = testData.MEGOLM_SESSION_DATA_ARRAY;
+
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", testData.SIGNED_BACKUP_DATA, {
+                overwriteRoutes: true,
+            });
+
+            const result = await aliceCrypto.checkKeyBackupAndEnable();
+            expect(result).toBeTruthy();
+
+            mockUploadEmitter(testData.SIGNED_BACKUP_DATA.version!);
+            await aliceCrypto.importRoomKeys(someRoomKeys);
+
+            // The backup loop is waiting a random amount of time to avoid different clients firing at the same time.
+            jest.runAllTimers();
+
+            // wait for all keys to be backed up
+            await remainingZeroPromise;
+
+            const newBackupVersion = "2";
+            const uploadMockEmitter = mockUploadEmitter(newBackupVersion);
+            const newBackup = JSON.parse(JSON.stringify(testData.SIGNED_BACKUP_DATA));
+            newBackup.version = newBackupVersion;
+
+            // Let's simulate that a new backup is available by returning error code on key upload
+
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", newBackup, {
+                overwriteRoutes: true,
+            });
+
+            // If we import a new key the loop will try to upload to old version, it will
+            // fail then check the current version and switch if trusted
+            const uploadPromises = someRoomKeys.map((data) => {
+                new Promise<void>((resolve) => {
+                    uploadMockEmitter.on(MockKeyUploadEvent.KeyUploaded, (roomId, sessionId, version) => {
+                        if (data.room_id == roomId && data.session_id == sessionId && version == newBackupVersion) {
+                            resolve();
+                        }
+                    });
+                });
+            });
+
+            const disableOldBackup = new Promise<void>((resolve) => {
+                aliceClient.on(CryptoEvent.KeyBackupFailed, (errCode) => {
+                    if (errCode == "M_WRONG_ROOM_KEYS_VERSION") {
+                        resolve();
+                    }
+                });
+            });
+
+            const enableNewBackup = new Promise<void>((resolve) => {
+                aliceClient.on(CryptoEvent.KeyBackupStatus, (enabled) => {
+                    if (enabled) {
+                        resolve();
+                    }
+                });
+            });
+
+            // A new key import should trigger a new upload.
+            const newKey = testData.MEGOLM_SESSION_DATA;
+
+            const newKeyUploadPromise = new Promise<void>((resolve) => {
+                uploadMockEmitter.on(MockKeyUploadEvent.KeyUploaded, (roomId, sessionId, version) => {
+                    if (newKey.room_id == roomId && newKey.session_id == sessionId && version == newBackupVersion) {
+                        resolve();
+                    }
+                });
+            });
+
+            await aliceCrypto.importRoomKeys([newKey]);
+
+            jest.runAllTimers();
+
+            await disableOldBackup;
+            await enableNewBackup;
+
+            jest.runAllTimers();
+
+            await Promise.all(uploadPromises);
+            await newKeyUploadPromise;
+        });
+
+        it("Backup loop should be resistant to network failures", async function () {
+            aliceClient = await initTestClient();
+            const aliceCrypto = aliceClient.getCrypto()!;
+            await aliceClient.startClient();
+
+            // tell Alice to trust the dummy device that signed the backup
+            await waitForDeviceList();
+            await aliceCrypto.setDeviceVerified(testData.TEST_USER_ID, testData.TEST_DEVICE_ID);
+
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", testData.SIGNED_BACKUP_DATA, {
+                overwriteRoutes: true,
+            });
+
+            // on the first key upload attempt, simulate a network failure
+            const failurePromise = new Promise((resolve) => {
+                fetchMock.put(
+                    "path:/_matrix/client/v3/room_keys/keys",
+                    () => {
+                        resolve(undefined);
+                        throw new TypeError(`Failed to fetch`);
+                    },
+                    {
+                        overwriteRoutes: true,
+                    },
+                );
+            });
+
+            // kick the import loop off and wait for the failed request
+            const someRoomKeys = testData.MEGOLM_SESSION_DATA_ARRAY;
+            await aliceCrypto.importRoomKeys(someRoomKeys);
+
+            const result = await aliceCrypto.checkKeyBackupAndEnable();
+            expect(result).toBeTruthy();
+            jest.runAllTimers();
+            await failurePromise;
+
+            // Fix the endpoint to do successful uploads
+            const successPromise = new Promise((resolve) => {
+                fetchMock.put(
+                    "path:/_matrix/client/v3/room_keys/keys",
+                    () => {
+                        resolve(undefined);
+                        return {
+                            status: 200,
+                            body: {
+                                count: 2,
+                                etag: "abcdefg",
+                            },
+                        };
+                    },
+                    {
+                        overwriteRoutes: true,
+                    },
+                );
+            });
+
+            // check that a `KeyBackupSessionsRemaining` event is emitted with `remaining == 0`
+            const allKeysUploadedPromise = new Promise((resolve) => {
+                aliceClient.on(CryptoEvent.KeyBackupSessionsRemaining, (remaining) => {
+                    if (remaining == 0) {
+                        resolve(undefined);
+                    }
+                });
+            });
+
+            // run the timers, which will make the backup loop redo the request
+            await jest.runAllTimersAsync();
+            await successPromise;
+            await allKeysUploadedPromise;
+        });
+    });
+
     it("getActiveSessionBackupVersion() should give correct result", async function () {
         // 404 means that there is no active backup
         fetchMock.get("express:/_matrix/client/v3/room_keys/version", 404);
@@ -363,10 +665,10 @@ describe.each(Object.entries(CRYPTO_BACKENDS))("megolm-keys backup (%s)", (backe
             expect(await aliceCrypto.getActiveSessionBackupVersion()).toEqual(testData.SIGNED_BACKUP_DATA.version);
 
             const newBackupVersion = "2";
-            const unsignedBackup = JSON.parse(JSON.stringify(testData.SIGNED_BACKUP_DATA));
+            const newBackup = JSON.parse(JSON.stringify(testData.SIGNED_BACKUP_DATA));
-            unsignedBackup.version = newBackupVersion;
+            newBackup.version = newBackupVersion;
 
-            fetchMock.get("path:/_matrix/client/v3/room_keys/version", unsignedBackup, {
+            fetchMock.get("path:/_matrix/client/v3/room_keys/version", newBackup, {
                 overwriteRoutes: true,
             });
 

spec/test-utils/test-data/generate-test-data.py

@@ -30,6 +30,7 @@ import json
 from canonicaljson import encode_canonical_json
 from cryptography.hazmat.primitives.asymmetric import ed25519, x25519
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+from random import randbytes, seed
 
 # input data
 TEST_USER_ID = "@alice:localhost"
@@ -116,6 +117,10 @@ def main() -> None:
         TEST_USER_ID: {f"ed25519:{TEST_DEVICE_ID}": sig}
     }
 
+    set_of_exported_room_keys = [build_exported_megolm_key(), build_exported_megolm_key()]
+
+    additional_exported_room_key = build_exported_megolm_key()
+
     print(
         f"""\
 /* Test data for cryptography tests
@@ -123,7 +128,7 @@ def main() -> None:
  * Do not edit by hand! This file is generated by `./generate-test-data.py`
  */
 
-import {{ IDeviceKeys }} from "../../../src/@types/crypto";
+import {{ IDeviceKeys, IMegolmSessionData }} from "../../../src/@types/crypto";
 import {{ IDownloadKeyResult }} from "../../../src";
 import {{ KeyBackupInfo }} from "../../../src/crypto-api";
 
@@ -167,11 +172,24 @@ export const BACKUP_DECRYPTION_KEY_BASE64 = "{ B64_BACKUP_DECRYPTION_KEY }";
 
 /** Signed backup data, suitable for return from `GET /_matrix/client/v3/room_keys/keys/{{roomId}}/{{sessionId}}` */
 export const SIGNED_BACKUP_DATA: KeyBackupInfo = { json.dumps(backup_data, indent=4) };
+
+/** A set of megolm keys that can be imported via CryptoAPI#importRoomKeys */
+export const MEGOLM_SESSION_DATA_ARRAY: IMegolmSessionData[] = {
+    json.dumps(set_of_exported_room_keys, indent=4)
+};
+
+/** An exported megolm session */
+export const MEGOLM_SESSION_DATA: IMegolmSessionData = { 
+        json.dumps(additional_exported_room_key, indent=4)
+};
 """,
         end="",
     )
 
 
+# Use static seed to have stable random test data upon new generation
+seed(10)
+
 def build_cross_signing_keys_data() -> dict:
     """Build the signed cross-signing-keys data for return from /keys/query"""
     master_private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
@@ -265,6 +283,39 @@ def sign_json(json_object: dict, private_key: ed25519.Ed25519PrivateKey) -> str:
 
     return signature_base64
 
+def build_exported_megolm_key() -> dict:
+    """
+    Creates an exported megolm room key, as per https://gitlab.matrix.org/matrix-org/olm/blob/master/docs/megolm.md#session-export-format
+    that can be imported via importRoomKeys API.
+    """
+    index = 0
+    private_key = ed25519.Ed25519PrivateKey.from_private_bytes(randbytes(32))
+    # Just use radom bytes for the ratchet parts
+    ratchet = randbytes(32 * 4)
+    # exported key, start with version byte
+    exported_key = bytearray(b'\x01')
+    exported_key += index.to_bytes(4, 'big')
+    exported_key += ratchet
+    # KPub
+    exported_key += private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
+
+
+    megolm_export = {
+        "algorithm": "m.megolm.v1.aes-sha2",
+        "room_id": "!roomA:example.org",
+        "sender_key": "/Bu9e34hUClhddpf4E5gu5qEAdMY31+1A9HbiAeeQgo",
+        "session_id": encode_base64(
+            private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
+        ),
+        "session_key": encode_base64(exported_key),
+        "sender_claimed_keys": {
+            "ed25519": encode_base64(ed25519.Ed25519PrivateKey.from_private_bytes(randbytes(32)).public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)),
+        },
+        "forwarding_curve25519_key_chain": [],
+    }
+
+    return megolm_export
+
 
 if __name__ == "__main__":
     main()

spec/test-utils/test-data/index.ts

@@ -3,7 +3,7 @@
  * Do not edit by hand! This file is generated by `./generate-test-data.py`
  */
 
-import { IDeviceKeys } from "../../../src/@types/crypto";
+import { IDeviceKeys, IMegolmSessionData } from "../../../src/@types/crypto";
 import { IDownloadKeyResult } from "../../../src";
 import { KeyBackupInfo } from "../../../src/crypto-api";
 
@@ -116,3 +116,42 @@ export const SIGNED_BACKUP_DATA: KeyBackupInfo = {
         }
     }
 };
+
+/** A set of megolm keys that can be imported via CryptoAPI#importRoomKeys */
+export const MEGOLM_SESSION_DATA_ARRAY: IMegolmSessionData[] = [
+    {
+        "algorithm": "m.megolm.v1.aes-sha2",
+        "room_id": "!roomA:example.org",
+        "sender_key": "/Bu9e34hUClhddpf4E5gu5qEAdMY31+1A9HbiAeeQgo",
+        "session_id": "FYOoKQSwe4d9jhTZ/LQCZFJINjPEqZ7Or4Z08reP92M",
+        "session_key": "AQAAAABZ0jXQOprFfXe41tIFmAtHxflJp4O2hM/vzQQpOazOCFeWSoW5P3Z9Q+voU3eXehMwyP8/hm/Q8xLP6/PmJdy+71se/17kdFwcDGgLxBWfa4ODM9zlI4EjKbNqmiii5loJ7rBhA/XXaw80m0hfU6zTDX/KrO55J0Pt4vJ0LDa3LBWDqCkEsHuHfY4U2fy0AmRSSDYzxKmezq+GdPK3j/dj",
+        "sender_claimed_keys": {
+            "ed25519": "QdgHgdpDgihgovpPzUiThXur1fbErTFh7paFvNKSgN0"
+        },
+        "forwarding_curve25519_key_chain": []
+    },
+    {
+        "algorithm": "m.megolm.v1.aes-sha2",
+        "room_id": "!roomA:example.org",
+        "sender_key": "/Bu9e34hUClhddpf4E5gu5qEAdMY31+1A9HbiAeeQgo",
+        "session_id": "mPYSGA2l1tOQiipEDEVYhDSdTSFh2lDW1qpGKYZRxTc",
+        "session_key": "AQAAAAAHwgkB49BTPAEGTCK6degxUIbl8GPG2ugPRYhNtOpNic63u11+baXFfjDw5fmVfD1gJXpQQjGsqrIYioxrB1xzl7mfb942UHhYdaMQZowpp1fSpJVsxR5TddUU2EWifYD9EQsoz8mY1zqoazm4vUP4v9yxaTcUBj2c6HMJCY0gCJj2EhgNpdbTkIoqRAxFWIQ0nU0hYdpQ1taqRimGUcU3",
+        "sender_claimed_keys": {
+            "ed25519": "IrkbT6H+0urDf6wKDSyVC1fh1t84Vz6T62snni86Cog"
+        },
+        "forwarding_curve25519_key_chain": []
+    }
+];
+
+/** An exported megolm session */
+export const MEGOLM_SESSION_DATA: IMegolmSessionData = {
+    "algorithm": "m.megolm.v1.aes-sha2",
+    "room_id": "!roomA:example.org",
+    "sender_key": "/Bu9e34hUClhddpf4E5gu5qEAdMY31+1A9HbiAeeQgo",
+    "session_id": "ipdI6Zs/7DzFTEhiA2iGaMDfHkIYCleqXT6L+5e1/co",
+    "session_key": "AQAAAABXGO+Z9jlQJhIL6ByhXrv2BwCIxkhh7MXpKLsYmXkJcWrQlirmXmD79ga1zo+I4DCtEZzyGSpDWXBC6G7ez3H4gDMBam1RE3Jm5tc+oTlIri32UkYgSL0kBkcEnttqmIXBlK8tAfJo3cJnlh7F4ltEOAqrdME6dU0zXTkqXmURqYqXSOmbP+w8xUxIYgNohmjA3x5CGApXql0+i/uXtf3K",
+    "sender_claimed_keys": {
+        "ed25519": "Bhbpt6hqMZlSH4sJV7xiEEEiPVeTWz4Vkujl1EMdIPI"
+    },
+    "forwarding_curve25519_key_chain": []
+};

spec/unit/rust-crypto/rust-crypto.spec.ts

@@ -61,32 +61,7 @@ describe("RustCrypto", () => {
         );
 
         it("should import and export keys", async () => {
-            const someRoomKeys = [
+            const someRoomKeys = testData.MEGOLM_SESSION_DATA_ARRAY;
-                {
-                    algorithm: "m.megolm.v1.aes-sha2",
-                    room_id: "!cLDYAnjpiQXIrSwngM:localhost:8480",
-                    sender_key: "C9FMqTD20C0VaGWE/aSImkimuE6HDa/RyYj5gRUg3gY",
-                    session_id: "iGQG5GaP1/B3dSH6zCQDQqrNuotrtQjVC7w1OsUDwbg",
-                    session_key:
-                        "AQAAAADaCbP2gdOy8jrhikjploKgSBaFSJ5rvHcziaADbwNEzeCSrfuAUlXvCvxik8kU+MfCHIi5arN2M7UM5rGKdzkHnkReoIByFkeMdbjKWk5SFpVQexcM74eDhBGj+ICkQqOgApfnEbSswrmreB0+MhHHyLStwW5fy5f8A9QW1sbPuohkBuRmj9fwd3Uh+swkA0KqzbqLa7UI1Qu8NTrFA8G4",
-                    sender_claimed_keys: {
-                        ed25519: "RSq0Xw0RR0DeqlJ/j3qrF5qbN0D96fKk8lz9kZJlG9k",
-                    },
-                    forwarding_curve25519_key_chain: [],
-                },
-                {
-                    algorithm: "m.megolm.v1.aes-sha2",
-                    room_id: "!cLDYAnjpiQXIrSwngM:localhost:8480",
-                    sender_key: "C9FMqTD20C0VaGWE/aSImkimuE6HDa/RyYj5gRUg3gY",
-                    session_id: "P/Jy9Tog4CMtLseeS4Fe2AEXZov3k6cibcop/uyhr78",
-                    session_key:
-                        "AQAAAAATyAVm0c9c9DW9Od72MxvfSDYoysBw3C6yMJ3bYuTmssHN7yNGm59KCtKeFp2Y5qO7lvUmwOfSTvTASUb7HViE7Lt+Bvp5WiMTJ2Pv6m+N12ihyowV5lgtKFWI18Wxd0AugMTVQRwjBK6aMobf86NXWD2hiKm3N6kWbC0PXmqV7T/ycvU6IOAjLS7HnkuBXtgBF2aL95OnIm3KKf7soa+/",
-                    sender_claimed_keys: {
-                        ed25519: "RSq0Xw0RR0DeqlJ/j3qrF5qbN0D96fKk8lz9kZJlG9k",
-                    },
-                    forwarding_curve25519_key_chain: [],
-                },
-            ];
             let importTotal = 0;
             const opt: ImportRoomKeysOpts = {
                 progressCallback: (stage) => {
@@ -95,11 +70,11 @@ describe("RustCrypto", () => {
             };
             await rustCrypto.importRoomKeys(someRoomKeys, opt);
 
-            expect(importTotal).toBe(2);
+            expect(importTotal).toBe(someRoomKeys.length);
 
             const keys = await rustCrypto.exportRoomKeys();
             expect(Array.isArray(keys)).toBeTruthy();
-            expect(keys.length).toBe(2);
+            expect(keys.length).toBe(someRoomKeys.length);
 
             const aSession = someRoomKeys[0];
 

src/client.ts

@@ -2252,6 +2252,8 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
             CryptoEvent.VerificationRequestReceived,
             CryptoEvent.UserTrustStatusChanged,
             CryptoEvent.KeyBackupStatus,
+            CryptoEvent.KeyBackupSessionsRemaining,
+            CryptoEvent.KeyBackupFailed,
         ]);
     }
 

src/crypto/backup.ts

@@ -440,6 +440,7 @@ export class BackupManager {
      * @param maxDelay - Maximum delay to wait in ms. 0 means no delay.
      */
     public async scheduleKeyBackupSend(maxDelay = 10000): Promise<void> {
+        logger.debug(`Key backup: scheduleKeyBackupSend currentSending:${this.sendingBackups} delay:${maxDelay}`);
         if (this.sendingBackups) return;
 
         this.sendingBackups = true;
@@ -452,6 +453,7 @@ export class BackupManager {
             await sleep(delay);
             if (!this.clientRunning) {
                 logger.debug("Key backup send aborted, client stopped");
+                this.sendingBackups = false;
                 return;
             }
             let numFailures = 0; // number of consecutive failures
@@ -463,24 +465,26 @@ export class BackupManager {
                     const numBackedUp = await this.backupPendingKeys(KEY_BACKUP_KEYS_PER_REQUEST);
                     if (numBackedUp === 0) {
                         // no sessions left needing backup: we're done
+                        this.sendingBackups = false;
                         return;
                     }
                     numFailures = 0;
                 } catch (err) {
                     numFailures++;
                     logger.log("Key backup request failed", err);
-                    if ((<MatrixError>err).data) {
+                    if (err instanceof MatrixError) {
-                        if (
+                        const errCode = err.data.errcode;
-                            (<MatrixError>err).data.errcode == "M_NOT_FOUND" ||
+                        if (errCode == "M_NOT_FOUND" || errCode == "M_WRONG_ROOM_KEYS_VERSION") {
-                            (<MatrixError>err).data.errcode == "M_WRONG_ROOM_KEYS_VERSION"
+                            // Set to false now as `checkKeyBackup` might schedule a backupsend before this one ends.
-                        ) {
+                            this.sendingBackups = false;
-                            // Re-check key backup status on error, so we can be
-                            // sure to present the current situation when asked.
-                            await this.checkKeyBackup();
                             // Backup version has changed or this backup version
                             // has been deleted
-                            this.baseApis.crypto!.emit(CryptoEvent.KeyBackupFailed, (<MatrixError>err).data.errcode!);
+                            this.baseApis.crypto!.emit(CryptoEvent.KeyBackupFailed, errCode);
-                            throw err;
+                            // Re-check key backup status on error, so we can be
+                            // sure to present the current situation when asked.
+                            // This call might restart the backup loop if new backup version is trusted
+                            await this.checkKeyBackup();
+                            return;
                         }
                     }
                 }
@@ -491,10 +495,14 @@ export class BackupManager {
 
                 if (!this.clientRunning) {
                     logger.debug("Key backup send loop aborted, client stopped");
+                    this.sendingBackups = false;
                     return;
                 }
             }
-        } finally {
+        } catch (err) {
+            // No one actually checks errors on this promise, it's spawned internally.
+            // Just log, apps/client should use events to check status
+            logger.log(`Backup loop failed ${err}`);
             this.sendingBackups = false;
         }
     }

src/rust-crypto/OutgoingRequestProcessor.ts

@@ -75,7 +75,12 @@ export class OutgoingRequestProcessor {
         } else if (msg instanceof SignatureUploadRequest) {
             resp = await this.rawJsonRequest(Method.Post, "/_matrix/client/v3/keys/signatures/upload", {}, msg.body);
         } else if (msg instanceof KeysBackupRequest) {
-            resp = await this.rawJsonRequest(Method.Put, "/_matrix/client/v3/room_keys/keys", {}, msg.body);
+            resp = await this.rawJsonRequest(
+                Method.Put,
+                "/_matrix/client/v3/room_keys/keys",
+                { version: msg.version },
+                msg.body,
+            );
         } else if (msg instanceof ToDeviceRequest) {
             const path =
                 `/_matrix/client/v3/sendToDevice/${encodeURIComponent(msg.event_type)}/` +

src/rust-crypto/backup.ts

@@ -22,6 +22,8 @@ import { logger } from "../logger";
 import { ClientPrefix, IHttpOpts, MatrixError, MatrixHttpApi, Method } from "../http-api";
 import { CryptoEvent } from "../crypto";
 import { TypedEventEmitter } from "../models/typed-event-emitter";
+import { OutgoingRequestProcessor } from "./OutgoingRequestProcessor";
+import { sleep } from "../utils";
 
 /**
  * @internal
@@ -30,14 +32,28 @@ export class RustBackupManager extends TypedEventEmitter<RustBackupCryptoEvents,
     /** Have we checked if there is a backup on the server which we can use */
     private checkedForBackup = false;
     private activeBackupVersion: string | null = null;
+    private stopped = false;
+
+    /** whether {@link backupKeysLoop} is currently running */
+    private backupKeysLoopRunning = false;
 
     public constructor(
         private readonly olmMachine: OlmMachine,
         private readonly http: MatrixHttpApi<IHttpOpts & { onlyData: true }>,
+        private readonly outgoingRequestProcessor: OutgoingRequestProcessor,
     ) {
         super();
     }
 
+    /**
+     * Tells the RustBackupManager to stop.
+     * The RustBackupManager is scheduling background uploads of keys to the backup, this
+     * call allows to cancel the process when the client is stoppped.
+     */
+    public stop(): void {
+        this.stopped = true;
+    }
+
     /**
      * Get the backup version we are currently backing up to, if any
      */
@@ -129,14 +145,10 @@ export class RustBackupManager extends TypedEventEmitter<RustBackupCryptoEvents,
                 await this.enableKeyBackup(backupInfo);
             } else if (activeVersion !== backupInfo.version) {
                 logger.log(`On backup version ${activeVersion} but found version ${backupInfo.version}: switching.`);
+                // This will remove any pending backup request, remove the backup key and reset the backup state of each room key we have.
                 await this.disableKeyBackup();
+                // Enabling will now trigger re-upload of all the keys
                 await this.enableKeyBackup(backupInfo);
-                // We're now using a new backup, so schedule all the keys we have to be
-                // uploaded to the new backup. This is a bit of a workaround to upload
-                // keys to a new backup in *most* cases, but it won't cover all cases
-                // because we don't remember what backup version we uploaded keys to:
-                // see https://github.com/vector-im/element-web/issues/14833
-                await this.scheduleAllGroupSessionsForBackup();
             } else {
                 logger.log(`Backup version ${backupInfo.version} still current`);
             }
@@ -157,7 +169,18 @@ export class RustBackupManager extends TypedEventEmitter<RustBackupCryptoEvents,
 
         this.emit(CryptoEvent.KeyBackupStatus, true);
 
-        // TODO: kick off an upload loop
+        this.backupKeysLoop();
+    }
+
+    /**
+     * Restart the backup key loop if there is an active trusted backup.
+     * Doesn't try to check the backup server side. To be called when a new
+     * megolm key is known locally.
+     */
+    public async maybeUploadKey(): Promise<void> {
+        if (this.activeBackupVersion != null) {
+            this.backupKeysLoop();
+        }
     }
 
     private async disableKeyBackup(): Promise<void> {
@@ -166,9 +189,73 @@ export class RustBackupManager extends TypedEventEmitter<RustBackupCryptoEvents,
         this.emit(CryptoEvent.KeyBackupStatus, false);
     }
 
-    private async scheduleAllGroupSessionsForBackup(): Promise<void> {
+    private async backupKeysLoop(maxDelay = 10000): Promise<void> {
-        // TODO stub
+        if (this.backupKeysLoopRunning) {
+            logger.log(`Backup loop already running`);
+            return;
+        }
+        this.backupKeysLoopRunning = true;
+
+        logger.log(`Starting loop for ${this.activeBackupVersion}.`);
+
+        // wait between 0 and `maxDelay` seconds, to avoid backup
+        // requests from different clients hitting the server all at
+        // the same time when a new key is sent
+        const delay = Math.random() * maxDelay;
+        await sleep(delay);
+
+        try {
+            let numFailures = 0; // number of consecutive network failures for exponential backoff
+
+            while (!this.stopped) {
+                // Get a batch of room keys to upload
+                const request: RustSdkCryptoJs.KeysBackupRequest | null = await this.olmMachine.backupRoomKeys();
+
+                if (!request || this.stopped || !this.activeBackupVersion) {
+                    logger.log(`Ending loop for ${this.activeBackupVersion}.`);
+                    return;
+                }
+
+                try {
+                    await this.outgoingRequestProcessor.makeOutgoingRequest(request);
+                    numFailures = 0;
+
+                    const keyCount: RustSdkCryptoJs.RoomKeyCounts = await this.olmMachine.roomKeyCounts();
+                    const remaining = keyCount.total - keyCount.backedUp;
+                    this.emit(CryptoEvent.KeyBackupSessionsRemaining, remaining);
+                } catch (err) {
+                    numFailures++;
+                    logger.error("Error processing backup request for rust crypto-sdk", err);
+                    if (err instanceof MatrixError) {
+                        const errCode = err.data.errcode;
+                        if (errCode == "M_NOT_FOUND" || errCode == "M_WRONG_ROOM_KEYS_VERSION") {
+                            await this.disableKeyBackup();
+                            this.emit(CryptoEvent.KeyBackupFailed, err.data.errcode!);
+                            // There was an active backup and we are out of sync with the server
+                            // force a check server side
+                            this.backupKeysLoopRunning = false;
+                            this.checkKeyBackupAndEnable(true);
+                            return;
+                        } else if (errCode == "M_LIMIT_EXCEEDED") {
+                            // wait for that and then continue?
+                            const waitTime = err.data.retry_after_ms;
+                            if (waitTime > 0) {
+                                sleep(waitTime);
+                                continue;
+                            } // else go to the normal backoff
+                        }
+                    }
+
+                    // Some other errors (mx, network, or CORS or invalid urls?) anyhow backoff
+                    // exponential backoff if we have failures
+                    await sleep(1000 * Math.pow(2, Math.min(numFailures - 1, 4)));
+                }
+            }
+        } finally {
+            this.backupKeysLoopRunning = false;
+        }
     }
+
     /**
      * Get information about the current key backup from the server
      *
@@ -195,8 +282,13 @@ export class RustBackupManager extends TypedEventEmitter<RustBackupCryptoEvents,
     }
 }
 
-export type RustBackupCryptoEvents = CryptoEvent.KeyBackupStatus;
+export type RustBackupCryptoEvents =
+    | CryptoEvent.KeyBackupStatus
+    | CryptoEvent.KeyBackupSessionsRemaining
+    | CryptoEvent.KeyBackupFailed;
 
 export type RustBackupCryptoEventMap = {
     [CryptoEvent.KeyBackupStatus]: (enabled: boolean) => void;
+    [CryptoEvent.KeyBackupSessionsRemaining]: (remaining: number) => void;
+    [CryptoEvent.KeyBackupFailed]: (errCode: string) => void;
 };

src/rust-crypto/rust-crypto.ts

@@ -118,8 +118,12 @@ export class RustCrypto extends TypedEventEmitter<RustCryptoEvents, RustCryptoEv
         this.keyClaimManager = new KeyClaimManager(olmMachine, this.outgoingRequestProcessor);
         this.eventDecryptor = new EventDecryptor(olmMachine);
 
-        this.backupManager = new RustBackupManager(olmMachine, http);
+        this.backupManager = new RustBackupManager(olmMachine, http, this.outgoingRequestProcessor);
-        this.reemitter.reEmit(this.backupManager, [CryptoEvent.KeyBackupStatus]);
+        this.reemitter.reEmit(this.backupManager, [
+            CryptoEvent.KeyBackupStatus,
+            CryptoEvent.KeyBackupSessionsRemaining,
+            CryptoEvent.KeyBackupFailed,
+        ]);
 
         // Fire if the cross signing keys are imported from the secret storage
         const onCrossSigningKeysImport = (): void => {
@@ -148,6 +152,7 @@ export class RustCrypto extends TypedEventEmitter<RustCryptoEvents, RustCryptoEv
         this.stopped = true;
 
         this.keyClaimManager.stop();
+        this.backupManager.stop();
 
         // make sure we close() the OlmMachine; doing so means that all the Rust objects will be
         // cleaned up; in particular, the indexeddb connections will be closed, which means they
@@ -1022,9 +1027,11 @@ export class RustCrypto extends TypedEventEmitter<RustCryptoEvents, RustCryptoEv
         for (const key of keys) {
             this.onRoomKeyUpdated(key);
         }
+        this.backupManager.maybeUploadKey();
     }
 
     private onRoomKeyUpdated(key: RustSdkCryptoJs.RoomKeyInfo): void {
+        if (this.stopped) return;
         logger.debug(`Got update for session ${key.senderKey.toBase64()}|${key.sessionId} in ${key.roomId.toString()}`);
         const pendingList = this.eventDecryptor.getEventsPendingRoomKey(key);
         if (pendingList.length === 0) return;

yarn.lock

@@ -1463,14 +1463,13 @@
     "@jridgewell/resolve-uri" "^3.1.0"
     "@jridgewell/sourcemap-codec" "^1.4.14"
 
-"@matrix-org/matrix-sdk-crypto-wasm@^1.2.0":
+"@matrix-org/matrix-sdk-crypto-wasm@^1.2.1":
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/@matrix-org/matrix-sdk-crypto-wasm/-/matrix-sdk-crypto-wasm-1.2.1.tgz#5b546c8a0e53b614f10b77b3b649818aed9d0db1"
   integrity sha512-DCb7Q83PCQK0uav5vB3KNV/hJPlxAhT/ddar+VHz2kC39hMLKGzWYVhprpLYVcavaE/6OX+Q/xFkAoV/3QtUHQ==
 
 "@matrix-org/olm@https://gitlab.matrix.org/api/v4/projects/27/packages/npm/@matrix-org/olm/-/@matrix-org/olm-3.2.14.tgz":
   version "3.2.14"
-  uid acd96c00a881d0f462e1f97a56c73742c8dbc984
   resolved "https://gitlab.matrix.org/api/v4/projects/27/packages/npm/@matrix-org/olm/-/@matrix-org/olm-3.2.14.tgz#acd96c00a881d0f462e1f97a56c73742c8dbc984"
 
 "@microsoft/tsdoc-config@0.16.2":