Skip to content

Commit a74284a

Browse files
richvdhrichvdh
committed
Implement outgoing requests from Rust SDK
1 parent 0ebea8b commit a74284a

File tree

2 files changed

+271
-11
lines changed

2 files changed

+271
-11
lines changed

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

Lines changed: 164 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,21 @@ limitations under the License.
1616

1717
import "fake-indexeddb/auto";
1818
import { IDBFactory } from "fake-indexeddb";
19+
import * as RustSdkCryptoJs from "@matrix-org/matrix-sdk-crypto-js";
20+
import {
21+
KeysBackupRequest,
22+
KeysClaimRequest,
23+
KeysQueryRequest,
24+
KeysUploadRequest,
25+
SignatureUploadRequest,
26+
} from "@matrix-org/matrix-sdk-crypto-js";
27+
import { Mocked } from "jest-mock";
28+
import MockHttpBackend from "matrix-mock-request";
1929

2030
import { RustCrypto } from "../../src/rust-crypto/rust-crypto";
2131
import { initRustCrypto } from "../../src/rust-crypto";
22-
import { IHttpOpts, MatrixHttpApi } from "../../src";
32+
import { HttpApiEvent, HttpApiEventHandlerMap, IHttpOpts, MatrixHttpApi } from "../../src";
33+
import { TypedEventEmitter } from "../../src/models/typed-event-emitter";
2334

2435
afterEach(() => {
2536
// reset fake-indexeddb after each test, to make sure we don't leak connections
@@ -32,17 +43,163 @@ describe("RustCrypto", () => {
3243
const TEST_USER = "@alice:example.com";
3344
const TEST_DEVICE_ID = "TEST_DEVICE";
3445

35-
let rustCrypto: RustCrypto;
46+
describe(".exportRoomKeys", () => {
47+
let rustCrypto: RustCrypto;
3648

37-
beforeEach(async () => {
38-
const mockHttpApi = {} as MatrixHttpApi<IHttpOpts>;
39-
rustCrypto = (await initRustCrypto(mockHttpApi, TEST_USER, TEST_DEVICE_ID)) as RustCrypto;
40-
});
49+
beforeEach(async () => {
50+
const mockHttpApi = {} as MatrixHttpApi<IHttpOpts>;
51+
rustCrypto = (await initRustCrypto(mockHttpApi, TEST_USER, TEST_DEVICE_ID)) as RustCrypto;
52+
});
4153

42-
describe(".exportRoomKeys", () => {
4354
it("should return a list", async () => {
4455
const keys = await rustCrypto.exportRoomKeys();
4556
expect(Array.isArray(keys)).toBeTruthy();
4657
});
4758
});
59+
60+
describe("outgoing requests", () => {
61+
/** the RustCrypto implementation under test */
62+
let rustCrypto: RustCrypto;
63+
64+
/** A mock http backend which rustCrypto is connected to */
65+
let httpBackend: MockHttpBackend;
66+
67+
/** a mocked-up OlmMachine which rustCrypto is connected to */
68+
let olmMachine: Mocked<RustSdkCryptoJs.OlmMachine>;
69+
70+
/** A list of results to be returned from olmMachine.outgoingRequest. Each call will shift a result off
71+
* the front of the queue, until it is empty. */
72+
let outgoingRequestQueue: Array<Array<any>>;
73+
74+
/** wait for a call to olmMachine.markRequestAsSent */
75+
function awaitCallToMarkAsSent(): Promise<void> {
76+
return new Promise((resolve, _reject) => {
77+
olmMachine.markRequestAsSent.mockImplementationOnce(async () => {
78+
resolve(undefined);
79+
});
80+
});
81+
}
82+
83+
beforeEach(async () => {
84+
httpBackend = new MockHttpBackend();
85+
86+
await RustSdkCryptoJs.initAsync();
87+
88+
const dummyEventEmitter = new TypedEventEmitter<HttpApiEvent, HttpApiEventHandlerMap>();
89+
const httpApi = new MatrixHttpApi(dummyEventEmitter, {
90+
baseUrl: "https://example.com",
91+
prefix: "/_matrix",
92+
fetchFn: httpBackend.fetchFn as typeof global.fetch,
93+
});
94+
95+
// for these tests we use a mock OlmMachine, with an implementation of outgoingRequests that
96+
// returns objects from outgoingRequestQueue
97+
outgoingRequestQueue = [];
98+
olmMachine = {
99+
outgoingRequests: jest.fn().mockImplementation(() => {
100+
return Promise.resolve(outgoingRequestQueue.shift() ?? []);
101+
}),
102+
markRequestAsSent: jest.fn(),
103+
close: jest.fn(),
104+
} as unknown as Mocked<RustSdkCryptoJs.OlmMachine>;
105+
106+
rustCrypto = new RustCrypto(olmMachine, httpApi, TEST_USER, TEST_DEVICE_ID);
107+
});
108+
109+
it("should poll for outgoing messages", () => {
110+
rustCrypto.onSyncCompleted({});
111+
expect(olmMachine.outgoingRequests).toHaveBeenCalled();
112+
});
113+
114+
/* simple requests that map directly to the request body */
115+
const tests: Array<[any, "POST" | "PUT", string]> = [
116+
[KeysUploadRequest, "POST", "https://example.com/_matrix/client/v3/keys/upload"],
117+
[KeysQueryRequest, "POST", "https://example.com/_matrix/client/v3/keys/query"],
118+
[KeysClaimRequest, "POST", "https://example.com/_matrix/client/v3/keys/claim"],
119+
[SignatureUploadRequest, "POST", "https://example.com/_matrix/client/v3/keys/signatures/upload"],
120+
[KeysBackupRequest, "PUT", "https://example.com/_matrix/client/v3/room_keys/keys"],
121+
];
122+
123+
for (const [RequestClass, expectedMethod, expectedPath] of tests) {
124+
it(`should handle ${RequestClass.name}s`, async () => {
125+
const testBody = '{ "foo": "bar" }';
126+
const outgoingRequest = new RequestClass("1234", testBody);
127+
outgoingRequestQueue.push([outgoingRequest]);
128+
129+
const testResponse = '{ "result": 1 }';
130+
httpBackend
131+
.when(expectedMethod, "/_matrix")
132+
.check((req) => {
133+
expect(req.path).toEqual(expectedPath);
134+
expect(req.rawData).toEqual(testBody);
135+
expect(req.headers["Accept"]).toEqual("application/json");
136+
expect(req.headers["Content-Type"]).toEqual("application/json");
137+
})
138+
.respond(200, testResponse, true);
139+
140+
rustCrypto.onSyncCompleted({});
141+
142+
expect(olmMachine.outgoingRequests).toHaveBeenCalledTimes(1);
143+
144+
const markSentCallPromise = awaitCallToMarkAsSent();
145+
await httpBackend.flushAllExpected();
146+
147+
await markSentCallPromise;
148+
expect(olmMachine.markRequestAsSent).toHaveBeenCalledWith("1234", outgoingRequest.type, testResponse);
149+
httpBackend.verifyNoOutstandingRequests();
150+
});
151+
}
152+
153+
it("does not explode with unknown requests", async () => {
154+
const outgoingRequest = { id: "5678", type: 987 };
155+
outgoingRequestQueue.push([outgoingRequest]);
156+
157+
rustCrypto.onSyncCompleted({});
158+
159+
await awaitCallToMarkAsSent();
160+
expect(olmMachine.markRequestAsSent).toHaveBeenCalledWith("5678", 987, "");
161+
});
162+
163+
it("stops looping when stop() is called", async () => {
164+
const testResponse = '{ "result": 1 }';
165+
166+
for (let i = 0; i < 5; i++) {
167+
outgoingRequestQueue.push([new KeysQueryRequest("1234", "{}")]);
168+
httpBackend.when("POST", "/_matrix").respond(200, testResponse, true);
169+
}
170+
171+
rustCrypto.onSyncCompleted({});
172+
173+
expect(rustCrypto["outgoingRequestLoopRunning"]).toBeTruthy();
174+
175+
// go a couple of times round the loop
176+
await httpBackend.flush("/_matrix", 1);
177+
await awaitCallToMarkAsSent();
178+
179+
await httpBackend.flush("/_matrix", 1);
180+
await awaitCallToMarkAsSent();
181+
182+
// a second sync while this is going on shouldn't make any difference
183+
rustCrypto.onSyncCompleted({});
184+
185+
await httpBackend.flush("/_matrix", 1);
186+
await awaitCallToMarkAsSent();
187+
188+
// now stop...
189+
rustCrypto.stop();
190+
191+
// which should (eventually) cause the loop to stop with no further calls to outgoingRequests
192+
olmMachine.outgoingRequests.mockReset();
193+
194+
await new Promise((resolve) => {
195+
setTimeout(resolve, 100);
196+
});
197+
expect(rustCrypto["outgoingRequestLoopRunning"]).toBeFalsy();
198+
httpBackend.verifyNoOutstandingRequests();
199+
expect(olmMachine.outgoingRequests).not.toHaveBeenCalled();
200+
201+
// we sent three, so there should be 2 left
202+
expect(outgoingRequestQueue.length).toEqual(2);
203+
});
204+
});
48205
});

‎src/rust-crypto/rust-crypto.ts

Lines changed: 107 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,29 @@ limitations under the License.
1515
*/
1616

1717
import * as RustSdkCryptoJs from "@matrix-org/matrix-sdk-crypto-js";
18+
import {
19+
DecryptedRoomEvent,
20+
KeysBackupRequest,
21+
KeysClaimRequest,
22+
KeysQueryRequest,
23+
KeysUploadRequest,
24+
SignatureUploadRequest,
25+
} from "@matrix-org/matrix-sdk-crypto-js";
1826

1927
import type { IEventDecryptionResult, IMegolmSessionData } from "../@types/crypto";
2028
import { MatrixEvent } from "../models/event";
2129
import { CryptoBackend, OnSyncCompletedData } from "../common-crypto/CryptoBackend";
22-
import { IHttpOpts, MatrixHttpApi } from "../http-api";
30+
import { logger } from "../logger";
31+
import { IHttpOpts, IRequestOpts, MatrixHttpApi, Method } from "../http-api";
32+
import { QueryDict } from "../utils";
2333

24-
// import { logger } from "../logger";
34+
/**
35+
* Common interface for all the request types returned by `OlmMachine.outgoingRequests`.
36+
*/
37+
interface OutgoingRequest {
38+
readonly id: string | undefined;
39+
readonly type: number;
40+
}
2541

2642
/**
2743
* An implementation of {@link CryptoBackend} using the Rust matrix-sdk-crypto.
@@ -30,9 +46,12 @@ export class RustCrypto implements CryptoBackend {
3046
public globalBlacklistUnverifiedDevices = false;
3147
public globalErrorOnUnknownDevices = false;
3248

33-
/** whether stop() has been called */
49+
/** whether {@link stop} has been called */
3450
private stopped = false;
3551

52+
/** whether {@link outgoingRequestLoop} is currently running */
53+
private outgoingRequestLoopRunning = false;
54+
3655
public constructor(
3756
private readonly olmMachine: RustSdkCryptoJs.OlmMachine,
3857
private readonly http: MatrixHttpApi<IHttpOpts>,
@@ -69,11 +88,95 @@ export class RustCrypto implements CryptoBackend {
6988
return [];
7089
}
7190

91+
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
92+
//
93+
// SyncCryptoCallbacks implementation
94+
//
95+
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
96+
7297
/** called by the sync loop after processing each sync.
7398
*
7499
* TODO: figure out something equivalent for sliding sync.
75100
*
76101
* @param syncState - information on the completed sync.
77102
*/
78-
public onSyncCompleted(syncState: OnSyncCompletedData): void {}
103+
public onSyncCompleted(syncState: OnSyncCompletedData): void {
104+
// Processing the /sync may have produced new outgoing requests which need sending, so kick off the outgoing
105+
// request loop, if it's not already running.
106+
this.outgoingRequestLoop();
107+
}
108+
109+
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
110+
//
111+
// Outgoing requests
112+
//
113+
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
114+
115+
private async outgoingRequestLoop(): Promise<void> {
116+
if (this.outgoingRequestLoopRunning) {
117+
return;
118+
}
119+
this.outgoingRequestLoopRunning = true;
120+
try {
121+
while (!this.stopped) {
122+
const outgoingRequests: Object[] = await this.olmMachine.outgoingRequests();
123+
if (outgoingRequests.length == 0 || this.stopped) {
124+
// no more messages to send (or we have been told to stop): exit the loop
125+
return;
126+
}
127+
for (const msg of outgoingRequests) {
128+
await this.doOutgoingRequest(msg as OutgoingRequest);
129+
}
130+
}
131+
} catch (e) {
132+
logger.error("Error processing outgoing-message requests from rust crypto-sdk", e);
133+
} finally {
134+
this.outgoingRequestLoopRunning = false;
135+
}
136+
}
137+
138+
private async doOutgoingRequest(msg: OutgoingRequest): Promise<void> {
139+
let resp: string;
140+
141+
/* refer https://docs.rs/matrix-sdk-crypto/0.6.0/matrix_sdk_crypto/requests/enum.OutgoingRequests.html
142+
* for the complete list of request types
143+
*/
144+
if (msg instanceof KeysUploadRequest) {
145+
resp = await this.rawJsonRequest(Method.Post, "/_matrix/client/v3/keys/upload", {}, msg.body);
146+
} else if (msg instanceof KeysQueryRequest) {
147+
resp = await this.rawJsonRequest(Method.Post, "/_matrix/client/v3/keys/query", {}, msg.body);
148+
} else if (msg instanceof KeysClaimRequest) {
149+
resp = await this.rawJsonRequest(Method.Post, "/_matrix/client/v3/keys/claim", {}, msg.body);
150+
} else if (msg instanceof SignatureUploadRequest) {
151+
resp = await this.rawJsonRequest(Method.Post, "/_matrix/client/v3/keys/signatures/upload", {}, msg.body);
152+
} else if (msg instanceof KeysBackupRequest) {
153+
resp = await this.rawJsonRequest(Method.Put, "/_matrix/client/v3/room_keys/keys", {}, msg.body);
154+
} else {
155+
// TODO: ToDeviceRequest, RoomMessageRequest
156+
logger.warn("Unsupported outgoing message", Object.getPrototypeOf(msg));
157+
resp = "";
158+
}
159+
160+
if (msg.id) {
161+
await this.olmMachine.markRequestAsSent(msg.id, msg.type, resp);
162+
}
163+
}
164+
165+
private async rawJsonRequest(
166+
method: Method,
167+
path: string,
168+
queryParams: QueryDict,
169+
body: string,
170+
opts: IRequestOpts = {},
171+
): Promise<string> {
172+
// unbeknownst to HttpApi, we are sending JSON
173+
opts.headers ??= {};
174+
opts.headers["Content-Type"] = "application/json";
175+
176+
// we use the full prefix
177+
opts.prefix ??= "";
178+
179+
const resp = await this.http.authedRequest(method, path, queryParams, body, opts);
180+
return await resp.text();
181+
}
79182
}

0 commit comments

Comments
 (0)