diff --git a/src-tauri/src/matrix_crypto/verification.rs b/src-tauri/src/matrix_crypto/verification.rs index b909035f0..4cc29acda 100644 --- a/src-tauri/src/matrix_crypto/verification.rs +++ b/src-tauri/src/matrix_crypto/verification.rs @@ -591,4 +591,181 @@ mod tests { assert_eq!(method_from_code(3), Some(VerificationMethod::ReciprocateV1)); assert_eq!(method_from_code(4), None); } + + mod sas_flow { + use std::collections::BTreeMap; + + use matrix_sdk::ruma::api::client::keys::get_keys; + use matrix_sdk::ruma::api::client::sync::sync_events::DeviceLists; + use matrix_sdk::ruma::api::client::to_device::send_event_to_device; + use matrix_sdk::ruma::encryption::DeviceKeys; + use matrix_sdk::ruma::events::AnyToDeviceEvent; + use matrix_sdk::ruma::serde::Raw; + use matrix_sdk_crypto::types::requests::{ + AnyIncomingResponse, AnyOutgoingRequest, ToDeviceRequest, + }; + use matrix_sdk_crypto::{DecryptionSettings, EncryptionSyncChanges, TrustRequirement}; + + use super::*; + + const USER: &str = "@sable:example.org"; + + async fn machine(device: &str) -> OlmMachine { + let user = UserId::parse(USER).unwrap(); + OlmMachine::new(&user, device.into()).await + } + + async fn device_keys(of: &OlmMachine) -> Raw { + for request in of.outgoing_requests().await.unwrap() { + if let AnyOutgoingRequest::KeysUpload(upload) = request.request() { + if let Some(keys) = upload.device_keys.clone() { + return keys; + } + } + } + panic!("machine issued no keys upload"); + } + + async fn learn_device(learner: &OlmMachine, about: &OlmMachine) { + let user = UserId::parse(USER).unwrap(); + let keys = device_keys(about).await; + learner.update_tracked_users([user.as_ref()]).await.unwrap(); + + let mut response = get_keys::v3::Response::new(); + response.device_keys.insert( + user.clone(), + BTreeMap::from([(about.device_id().to_owned(), keys)]), + ); + + for request in learner.outgoing_requests().await.unwrap() { + if matches!(request.request(), AnyOutgoingRequest::KeysQuery(_)) { + learner + .mark_request_as_sent( + request.request_id(), + AnyIncomingResponse::KeysQuery(&response), + ) + .await + .unwrap(); + return; + } + } + panic!("machine issued no keys query"); + } + + fn as_events(sender: &UserId, request: &ToDeviceRequest) -> Vec> { + let mut events = Vec::new(); + for devices in request.messages.values() { + for content in devices.values() { + let event = json!({ + "sender": sender, + "type": request.event_type.to_string(), + "content": content, + }); + events.push(Raw::new(&event).unwrap().cast_unchecked()); + } + } + events + } + + async fn deliver(to: &OlmMachine, events: Vec>) { + let device_lists = DeviceLists::new(); + let counts = BTreeMap::new(); + let settings = DecryptionSettings { + sender_device_trust_requirement: TrustRequirement::Untrusted, + }; + to.receive_sync_changes( + EncryptionSyncChanges { + to_device_events: events, + changed_devices: &device_lists, + one_time_keys_counts: &counts, + unused_fallback_keys: None, + next_batch_token: None, + }, + &settings, + ) + .await + .unwrap(); + } + + async fn send_one( + from: &OlmMachine, + to: &OlmMachine, + request: OutgoingVerificationRequest, + ) { + match request { + OutgoingVerificationRequest::ToDevice(request) => { + deliver(to, as_events(from.user_id(), &request)).await; + } + OutgoingVerificationRequest::InRoom(_) => panic!("expected a to-device request"), + } + } + + async fn pump(from: &OlmMachine, to: &OlmMachine) { + for request in from.outgoing_requests().await.unwrap() { + if let AnyOutgoingRequest::ToDeviceRequest(to_device) = request.request() { + deliver(to, as_events(from.user_id(), to_device)).await; + from.mark_request_as_sent( + request.request_id(), + AnyIncomingResponse::ToDevice(&send_event_to_device::v3::Response::new()), + ) + .await + .unwrap(); + } + } + } + + fn emoji_of(machine: &OlmMachine, flow_id: &str) -> Value { + let user = UserId::parse(USER).unwrap(); + let request = machine.get_verification_request(&user, flow_id).unwrap(); + request_state(&request)["verification"]["emoji"].clone() + } + + #[tokio::test] + async fn emoji_appear_only_once_our_own_key_is_acked() { + let alice = machine("DEVICEA").await; + let bob = machine("DEVICEB").await; + learn_device(&alice, &bob).await; + learn_device(&bob, &alice).await; + + let user = UserId::parse(USER).unwrap(); + let bob_device = alice + .get_device(&user, "DEVICEB".into(), None) + .await + .unwrap() + .expect("alice should know bob's device"); + + let (alice_request, outgoing) = bob_device.request_verification(); + send_one(&alice, &bob, outgoing).await; + let flow_id = alice_request.flow_id().as_str().to_owned(); + + let bob_request = bob.get_verification_request(&user, &flow_id).unwrap(); + send_one(&bob, &alice, bob_request.accept().unwrap()).await; + + let (_, start) = alice_request.start_sas().await.unwrap().unwrap(); + send_one(&alice, &bob, start).await; + + let bob_sas = bob + .get_verification(&user, &flow_id) + .unwrap() + .sas_v1() + .unwrap(); + send_one(&bob, &alice, bob_sas.accept().unwrap()).await; + + pump(&alice, &bob).await; + assert!( + emoji_of(&bob, &flow_id).is_null(), + "the peer's key alone must not expose emoji" + ); + + pump(&bob, &alice).await; + assert!( + !emoji_of(&bob, &flow_id).is_null(), + "responder must have emoji once its own key is acked" + ); + assert!( + !emoji_of(&alice, &flow_id).is_null(), + "initiator must have emoji once the responder's key arrives" + ); + } + } } diff --git a/src/app/crypto/engineCrypto/EngineCrypto.ts b/src/app/crypto/engineCrypto/EngineCrypto.ts index 090670387..5fa2c571a 100644 --- a/src/app/crypto/engineCrypto/EngineCrypto.ts +++ b/src/app/crypto/engineCrypto/EngineCrypto.ts @@ -7,9 +7,13 @@ import { encodeRecoveryKey, EventType, ImportRoomKeyStage, + KnownMembership, + MatrixEventEvent, + MsgType, UserVerificationStatus, VerificationMethod, } from '$types/matrix-sdk'; +import { isVerificationEvent } from 'matrix-js-sdk/lib/rust-crypto/verification'; import { Device, DeviceVerification } from 'matrix-js-sdk/lib/models/device'; import { getHttpUriForMxc } from 'matrix-js-sdk/lib/content-repo'; import * as RustSdkCryptoJs from '@matrix-org/matrix-sdk-crypto-wasm'; @@ -65,6 +69,7 @@ import type { MatrixEvent, OwnDeviceKeys, Room, + RoomMember, SecretStorageStatus, StartDehydrationOpts, VerificationRequest, @@ -72,6 +77,8 @@ import type { const engineCryptoLog = createDebugLogger('engine-crypto'); +const DECRYPTION_WAIT_MS = 5 * 60 * 1000; + /** js-sdk keeps this union private to its own rust-crypto module; derived the same way. */ type CryptoEvents = (typeof CryptoEvent)[keyof typeof CryptoEvent]; @@ -244,6 +251,8 @@ export class EngineCrypto /** Live requests, keyed by flow id, so the synchronous CryptoApi getters can answer. */ readonly #verificationRequests = new Map(); + #flushing: Promise = Promise.resolve(); + constructor(mx: MatrixClient, identity: EngineIdentity) { super(); this.#mx = mx; @@ -349,6 +358,100 @@ export class EngineCrypto return request; } + async onLiveEventFromSync(event: MatrixEvent): Promise { + if (event.isState() || event.getUnsigned().transaction_id) return; + + const handle = async (candidate: MatrixEvent): Promise => { + if (isVerificationEvent(candidate)) await this.onKeyVerificationEvent(candidate); + }; + + if (event.isDecryptionFailure() || event.isEncrypted()) { + let timeoutId: ReturnType; + const onDecrypted = (decrypted: MatrixEvent, error?: Error) => { + if (error) return; + clearTimeout(timeoutId); + event.off(MatrixEventEvent.Decrypted, onDecrypted); + void handle(decrypted); + }; + timeoutId = setTimeout(() => { + event.off(MatrixEventEvent.Decrypted, onDecrypted); + }, DECRYPTION_WAIT_MS); + event.on(MatrixEventEvent.Decrypted, onDecrypted); + return; + } + + await handle(event); + } + + async onKeyVerificationEvent(event: MatrixEvent): Promise { + const roomId = event.getRoomId(); + const senderId = event.getSender(); + const eventId = event.getId(); + if (!roomId || !senderId || !eventId) return; + + const content = event.getContent(); + const isRequest = + event.getType() === EventType.RoomMessage && + content.msgtype === MsgType.KeyVerificationRequest; + + if (isRequest) { + await this.#sendTracked(await this.#call('queryKeysForUsers', { users: [senderId] })); + } + + await this.#call('receiveVerificationEvent', { + roomId, + event: JSON.stringify({ + event_id: eventId, + type: event.getType(), + sender: senderId, + state_key: event.getStateKey(), + content, + origin_server_ts: event.getTs(), + }), + }); + + if (isRequest) { + await this.onIncomingKeyVerificationRequest(senderId, eventId); + } else { + const flowId = (content['m.relates_to'] as { event_id?: string } | undefined)?.event_id; + if (flowId) await this.#verificationRequests.get(flowId)?.refresh(); + } + + await this.#flushOutgoingRequests(); + } + + onRoomStateEvent(event: MatrixEvent): void { + if (event.getType() !== EventType.RoomMember) return; + if ( + event.getStateKey() !== this.#identity.userId && + event.getContent().membership !== KnownMembership.Join + ) { + void this.forceDiscardSession(event.getRoomId() ?? ''); + } + } + + onRoomMembership(event: MatrixEvent, member: RoomMember, oldMembership?: string): void { + const roomId = event.getRoomId(); + if (!roomId) return; + if ( + oldMembership === KnownMembership.Join && + member.membership !== KnownMembership.Join && + member.userId === this.#identity.userId + ) { + void this.#call('clearRoomPendingKeyBundle', { roomId }); + } + } + + async #sendTracked(request: unknown): Promise { + if (!isOutgoingRequest(request)) return; + const response = await sendOutgoingRequest(this.#mx, request); + await this.#call('markRequestAsSent', { + requestId: request.id, + requestType: request.type, + response, + }); + } + async onIncomingKeyVerificationRequest(sender: string, transactionId: string): Promise { const state = (await this.#call('getVerificationRequest', { userId: sender, @@ -366,9 +469,18 @@ export class EngineCrypto this.emit(CryptoEvent.VerificationRequestReceived, request); } + #flushOutgoingRequests(): Promise { + this.#flushing = this.#flushing.then(() => + this.#drainOutgoingRequests().catch((error: unknown) => { + engineCryptoLog.error('general', 'Draining outgoing crypto requests failed', error); + }) + ); + return this.#flushing; + } + /** matrix-sdk-crypto only clears a request once told it was sent, so a failure here * leaves it queued for the next drain rather than losing it. */ - async #flushOutgoingRequests(): Promise { + async #drainOutgoingRequests(): Promise { if (this.#stopped) return; const requests = ((await this.#call('outgoingRequests')) ?? []) as OutgoingRequest[]; @@ -396,15 +508,30 @@ export class EngineCrypto const processed = await this.#receiveSyncChanges({ toDeviceEvents: events }); const received: ReceivedToDeviceMessage[] = []; - for (const event of processed) { - const message = JSON.parse(event.rawEvent) as IToDeviceEvent; + const messages = processed.map( + (event) => [event, JSON.parse(event.rawEvent) as IToDeviceEvent] as const + ); + + if ( + messages.some( + ([, message]) => + typeof message.type === 'string' && message.type.startsWith('m.key.verification.') + ) + ) { + await this.#flushOutgoingRequests(); + } + for (const [event, message] of messages) { if (typeof message.type === 'string' && message.type.startsWith('m.key.verification.')) { const transactionId = (message.content as { transaction_id?: string })?.transaction_id; if (transactionId && message.sender) { if (message.type === EventType.KeyVerificationRequest) { // eslint-disable-next-line no-await-in-loop await this.onIncomingKeyVerificationRequest(message.sender, transactionId); + } else if (message.type === EventType.KeyVerificationDone) { + // Rust removes completed requests while consuming the event, so no state snapshot + // exists to refresh. Keep the JS request alive long enough to expose Done. + this.#verificationRequests.get(transactionId)?.markDone(); } else { // Without this the verifier never learns the SAS digits arrived. // eslint-disable-next-line no-await-in-loop @@ -444,9 +571,22 @@ export class EngineCrypto } async onCryptoEvent(room: Room, event: MatrixEvent): Promise { - engineCryptoLog.debug('general', 'Room encryption configured', { + const config = event.getContent(); + if (config.algorithm !== 'm.megolm.v1.aes-sha2') { + engineCryptoLog.warn('general', 'Ignoring encryption event with invalid algorithm', { + roomId: room.roomId, + algorithm: config.algorithm, + }); + return; + } + + await this.#call('setRoomSettings', { roomId: room.roomId, - algorithm: event.getContent().algorithm, + settings: { + algorithm: config.algorithm, + sessionRotationPeriodMs: config.rotation_period_ms, + sessionRotationPeriodMessages: config.rotation_period_msgs, + }, }); } @@ -1032,7 +1172,7 @@ export class EngineCrypto const started = (await this.#call('userIdentity.requestVerificationDm', { userId, roomId, - eventId, + requestEventId: eventId, methods: SUPPORTED_VERIFICATION_METHOD_CODES, })) as { request: EngineVerificationState; outgoingRequest?: unknown }; @@ -1040,7 +1180,10 @@ export class EngineCrypto await sendOutgoingRequest(this.#mx, started.outgoingRequest); } await this.#flushOutgoingRequests(); - return new EngineVerificationRequest(this.#engineCall, started.request); + + const request = new EngineVerificationRequest(this.#engineCall, started.request); + this.#verificationRequests.set(started.request.flowId, request); + return request; } async requestOwnUserVerification(): Promise { diff --git a/src/app/crypto/engineCrypto/outgoingDispatch.test.ts b/src/app/crypto/engineCrypto/outgoingDispatch.test.ts index bf7875a97..72ab0e121 100644 --- a/src/app/crypto/engineCrypto/outgoingDispatch.test.ts +++ b/src/app/crypto/engineCrypto/outgoingDispatch.test.ts @@ -1,5 +1,5 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; -import type { MatrixClient } from '$types/matrix-sdk'; +import { VerificationPhase, EventType, type MatrixClient } from '$types/matrix-sdk'; import { engineInvoke } from '../olmMachine/engineInvoke'; import { EngineCrypto } from './EngineCrypto'; @@ -100,4 +100,122 @@ describe('verification outgoing requests', () => { ); expect(verificationCalls).toEqual([]); }); + + it('completes a request when the engine consumes the final done event', async () => { + const { mx } = clientSpy(); + mockInvoke.mockImplementation(async (_identity, method) => { + if (method === 'device.requestVerification') { + return { + request: { + flowId: '$f', + otherUserId: '@them:e.org', + phase: 3, + verification: { className: 'Sas' }, + }, + }; + } + if (method === 'receiveSyncChanges') { + return [ + { + type: 3, + rawEvent: JSON.stringify({ + type: EventType.KeyVerificationDone, + sender: '@them:e.org', + content: { transaction_id: '$f' }, + }), + }, + ]; + } + if (method === 'verificationRequest.state') { + throw new Error('verificationRequest.state: no verification request'); + } + return []; + }); + + const crypto = new EngineCrypto(mx, { userId: '@me:e.org', deviceId: 'D' }); + const request = await crypto.requestDeviceVerification('@them:e.org', 'THEIRS'); + + await expect( + crypto.preprocessToDeviceMessages([ + { + type: EventType.KeyVerificationDone, + sender: '@them:e.org', + content: { transaction_id: '$f' }, + } as never, + ]) + ).resolves.toEqual([]); + expect(request.phase).toBe(VerificationPhase.Done); + }); + + it('drains the outgoing queue before re-reading the SAS snapshot', async () => { + const authedRequest = vi.fn<(...args: never[]) => Promise>( + async () => + new Promise((resolve) => { + setTimeout(() => resolve('{}'), 5); + }) + ); + const mx = { http: { authedRequest } } as unknown as MatrixClient; + let peerKeyReceived = false; + let ourKeyAcked = false; + + mockInvoke.mockImplementation(async (_identity, method) => { + if (method === 'device.requestVerification') { + return { + request: { + flowId: '$f', + otherUserId: '@them:e.org', + phase: 3, + verification: { className: 'Sas' }, + }, + }; + } + if (method === 'receiveSyncChanges') { + peerKeyReceived = true; + return [ + { + type: 3, + rawEvent: JSON.stringify({ + type: 'm.key.verification.key', + sender: '@them:e.org', + content: { transaction_id: '$f' }, + }), + }, + ]; + } + if (method === 'outgoingRequests') { + return peerKeyReceived + ? [{ id: 'k', type: 3, body: '{}', event_type: 'm.key.verification.key', txn_id: 'k' }] + : []; + } + if (method === 'markRequestAsSent') { + ourKeyAcked = true; + return null; + } + if (method === 'verificationRequest.state') { + return { + flowId: '$f', + otherUserId: '@them:e.org', + phase: 3, + verification: { + className: 'Sas', + emoji: ourKeyAcked ? [{ symbol: '🌏', description: 'Globe' }] : null, + }, + }; + } + return []; + }); + + const crypto = new EngineCrypto(mx, { userId: '@me:e.org', deviceId: 'D' }); + const request = await crypto.requestDeviceVerification('@them:e.org', 'THEIRS'); + + await crypto.preprocessToDeviceMessages([ + { + type: 'm.key.verification.key', + sender: '@them:e.org', + content: { transaction_id: '$f' }, + } as never, + ]); + + expect(request.verifier?.getShowSasCallbacks()?.sas.emoji).toEqual([['🌏', 'Globe']]); + }); }); diff --git a/src/app/crypto/install.ts b/src/app/crypto/install.ts index d1f76faa9..b8f564699 100644 --- a/src/app/crypto/install.ts +++ b/src/app/crypto/install.ts @@ -1,7 +1,8 @@ import { CryptoEvent } from 'matrix-js-sdk/lib/crypto-api'; import { ReEmitter } from 'matrix-js-sdk/lib/ReEmitter'; import { isTauri } from '@tauri-apps/api/core'; -import type { MatrixClient } from '$types/matrix-sdk'; +import { ClientEvent, RoomMemberEvent, RoomStateEvent } from '$types/matrix-sdk'; +import type { MatrixClient, MatrixEvent, RoomMember } from '$types/matrix-sdk'; import { createDebugLogger } from '$utils/debugLogger'; import { engineOpen } from '$generated/tauri/commands'; import * as RustSdkCryptoJs from '@matrix-org/matrix-sdk-crypto-wasm'; @@ -72,6 +73,28 @@ export const reEmitCryptoEvents = (mx: MatrixClient, crypto: EngineCrypto): (() return () => reEmitter.stopReEmitting(crypto, REEMITTED_CRYPTO_EVENTS); }; +export const wireCryptoClientEvents = (mx: MatrixClient, crypto: EngineCrypto): (() => void) => { + const onLiveEvent = (event: MatrixEvent) => { + void crypto.onLiveEventFromSync(event); + }; + const onMembership = (event: MatrixEvent, member: RoomMember, oldMembership?: string) => { + crypto.onRoomMembership(event, member, oldMembership); + }; + const onStateEvent = (event: MatrixEvent) => { + crypto.onRoomStateEvent(event); + }; + + mx.on(ClientEvent.Event, onLiveEvent); + mx.on(RoomMemberEvent.Membership, onMembership); + mx.on(RoomStateEvent.Events, onStateEvent); + + return () => { + mx.removeListener(ClientEvent.Event, onLiveEvent); + mx.removeListener(RoomMemberEvent.Membership, onMembership); + mx.removeListener(RoomStateEvent.Events, onStateEvent); + }; +}; + export const installRustCrypto = async ( mx: MatrixClient, options: { storeDir?: string; passphrase?: string } = {} @@ -98,11 +121,13 @@ export const installRustCrypto = async ( // `MatrixClient.initRustCrypto` normally wires these events to the client. The native // engine is installed independently, so reproduce that SDK initialization step here. const stopReEmittingCryptoEvents = reEmitCryptoEvents(mx, engineCrypto); + const stopClientEvents = wireCryptoClientEvents(mx, engineCrypto); const stopEventBridge = await startCryptoEventBridge(engineCrypto, identity); const stopEngineCrypto = engineCrypto.stop.bind(engineCrypto); engineCrypto.stop = () => { stopReEmittingCryptoEvents(); + stopClientEvents(); stopEventBridge(); stopEngineCrypto(); }; diff --git a/src/app/crypto/verification/request.ts b/src/app/crypto/verification/request.ts index 66648cc7e..e9935804f 100644 --- a/src/app/crypto/verification/request.ts +++ b/src/app/crypto/verification/request.ts @@ -57,7 +57,8 @@ export class EngineVerificationRequest #syncVerifier(): void { const verification = this.#state.verification; if (!verification) { - this.#verifier = undefined; + if (this.#state.phase === EnginePhase.Done) this.#verifier?.settle(true); + else if (this.#state.phase === EnginePhase.Cancelled) this.#verifier?.settle(false); return; } @@ -105,6 +106,19 @@ export class EngineVerificationRequest this.emit(VerificationRequestEvent.Change); } + markDone(): void { + this.#state = { + ...this.#state, + phase: EnginePhase.Done, + isDone: true, + verification: this.#state.verification + ? { ...this.#state.verification, isDone: true } + : this.#state.verification, + }; + this.#syncVerifier(); + this.emit(VerificationRequestEvent.Change); + } + get transactionId(): string | undefined { return this.#state.flowId; } @@ -191,6 +205,7 @@ export class EngineVerificationRequest } finally { this.#accepting = false; } + this.emit(VerificationRequestEvent.Change); } async cancel(params?: { reason?: string; code?: string }): Promise { @@ -201,6 +216,7 @@ export class EngineVerificationRequest } finally { this.#declining = false; } + this.emit(VerificationRequestEvent.Change); } async startVerification(method: string): Promise { diff --git a/src/app/crypto/verification/state.ts b/src/app/crypto/verification/state.ts index 92a6dd673..4faa8e3df 100644 --- a/src/app/crypto/verification/state.ts +++ b/src/app/crypto/verification/state.ts @@ -47,7 +47,7 @@ export type EngineVerificationState = { theirSupportedMethods?: number[] | null; ourSupportedMethods?: number[] | null; cancelInfo?: EngineCancelInfo | null; - verification?: { className?: string } | null; + verification?: { className?: string; isDone?: boolean } | null; }; export const methodFromCode = (code: number): string | undefined => METHOD_BY_CODE[code]; diff --git a/src/app/crypto/verification/verifier.ts b/src/app/crypto/verification/verifier.ts index 535a9a720..fecdbb603 100644 --- a/src/app/crypto/verification/verifier.ts +++ b/src/app/crypto/verification/verifier.ts @@ -91,6 +91,15 @@ abstract class EngineVerifier this.#cancelled = true; } + settle(done: boolean): void { + if (done) { + this.completion.resolve(); + return; + } + this.markCancelled(); + this.completion.reject(new Error('Verification cancelled')); + } + abstract onChange(state: TState): void; abstract get verificationPhase(): VerificationPhase; diff --git a/src/client/initMatrix.test.ts b/src/client/initMatrix.test.ts index d819b0669..80dfa666e 100644 --- a/src/client/initMatrix.test.ts +++ b/src/client/initMatrix.test.ts @@ -27,6 +27,41 @@ import { } from './initMatrix'; describe('installSlidingSyncRequestPatch', () => { + it('invalidates device lists once per pos-less run, not once per request', async () => { + const markAllTrackedUsersAsDirty = vi.fn<() => Promise>(async () => undefined); + const crypto = { markAllTrackedUsersAsDirty }; + let pos: string | undefined; + const original = vi.fn<() => Promise>(async () => ({ pos }) as never); + const mx = { + slidingSync: original, + getRoom: () => undefined, + getCrypto: () => crypto, + } as unknown as MatrixClient; + const manager = { + isPaused: () => false, + getActiveRoomSubscriptionIds: () => new Set(), + trackSubscriptionRequest: () => () => undefined, + sanitizeOptimisticJoinResponse: () => undefined, + }; + + installSlidingSyncRequestPatch(mx, manager as never); + const sync = (requestPos?: string) => + mx.slidingSync({ extensions: {}, pos: requestPos } as never, '', undefined); + + await sync(undefined); + await crypto.markAllTrackedUsersAsDirty(); + await sync(undefined); + await crypto.markAllTrackedUsersAsDirty(); + expect(markAllTrackedUsersAsDirty).toHaveBeenCalledTimes(1); + + pos = 'p1'; + await sync(undefined); + pos = undefined; + await sync('p1'); + await crypto.markAllTrackedUsersAsDirty(); + expect(markAllTrackedUsersAsDirty).toHaveBeenCalledTimes(2); + }); + it('normalizes expanded timelines before returning the response to the SDK', async () => { const response = { rooms: { diff --git a/src/client/initMatrix.ts b/src/client/initMatrix.ts index 89f599e02..af59efef2 100644 --- a/src/client/initMatrix.ts +++ b/src/client/initMatrix.ts @@ -232,6 +232,41 @@ type SlidingSyncRequestWithConnId = MSC3575SlidingSyncRequest & { export const newSlidingSyncConnId = (): string => `sable-${globalThis.crypto?.randomUUID?.().slice(0, 8) ?? `${Date.now()}-${Math.random().toString(16).slice(2)}`}`; +type CryptoWithDeviceListInvalidation = { markAllTrackedUsersAsDirty: () => Promise }; + +const coalesceDeviceListInvalidation = (mx: MatrixClient) => { + let invalidatedThisRun = false; + let patched: CryptoWithDeviceListInvalidation | undefined; + let restore: (() => void) | undefined; + + const install = (invalidationAlreadyRan: boolean): void => { + const crypto = mx.getCrypto?.() as unknown as CryptoWithDeviceListInvalidation | undefined; + if (!crypto?.markAllTrackedUsersAsDirty || crypto === patched) return; + + patched = crypto; + invalidatedThisRun = invalidationAlreadyRan; + const original = crypto.markAllTrackedUsersAsDirty.bind(crypto); + restore = () => { + crypto.markAllTrackedUsersAsDirty = original; + }; + crypto.markAllTrackedUsersAsDirty = async () => { + if (invalidatedThisRun) return; + invalidatedThisRun = true; + await original(); + }; + }; + + install(false); + + return { + onRequest: (pos: string | undefined) => install(pos === undefined), + onResponse: (pos: string | undefined) => { + if (pos !== undefined) invalidatedThisRun = false; + }, + dispose: () => restore?.(), + }; +}; + export function installSlidingSyncRequestPatch( mx: MatrixClient, manager: SlidingSyncManager @@ -239,6 +274,7 @@ export function installSlidingSyncRequestPatch( slidingSyncRequestCleanupByClient.get(mx)?.(); const connId = newSlidingSyncConnId(); + const deviceListInvalidation = coalesceDeviceListInvalidation(mx); const mxWritable = mx as MatrixClientWithWritableSlidingSync; const original = mx.slidingSync.bind(mx) as SlidingSyncMethod; mxWritable.slidingSync = async (reqBody, baseUrl, abortSignal) => { @@ -260,8 +296,10 @@ export function installSlidingSyncRequestPatch( const roomIds = manager.getActiveRoomSubscriptionIds(); scopeTypingExtension(req.extensions, roomIds); + deviceListInvalidation.onRequest(req.pos); const response = await original(reqBody, baseUrl, abortSignal); + deviceListInvalidation.onResponse(response.pos); trackResponse(response); // Must run before the SDK processes the response. A throw would reach the SDK's // loop, which drops the response and retries the same `pos` forever. @@ -282,6 +320,7 @@ export function installSlidingSyncRequestPatch( slidingSyncRequestCleanupByClient.set(mx, () => { slidingSyncRequestCleanupByClient.delete(mx); + deviceListInvalidation.dispose(); mxWritable.slidingSync = original; }); } diff --git a/src/client/presenceSync.test.ts b/src/client/presenceSync.test.ts new file mode 100644 index 000000000..96b240495 --- /dev/null +++ b/src/client/presenceSync.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { MatrixClient } from '$types/matrix-sdk'; +import { PresenceSyncManager } from './presenceSync'; + +describe('PresenceSyncManager', () => { + it('does not consume to-device events already handled by sliding sync', async () => { + const preprocessToDeviceMessages = vi.fn<(...args: never[]) => Promise>(); + const authedRequest = vi + .fn<() => Promise<{ next_batch: string; to_device: { events: object[] } }>>() + .mockResolvedValue({ + next_batch: 'next', + to_device: { events: [{ type: 'm.key.verification.key' }] }, + }); + const mx = { + getCrypto: () => ({ preprocessToDeviceMessages }), + getUserId: () => '@me:example.org', + getOrCreateFilter: vi.fn<() => Promise>().mockResolvedValue('presence-filter'), + http: { authedRequest }, + } as unknown as MatrixClient; + const manager = new PresenceSyncManager(mx, 1, 60_000); + + manager.start(); + await vi.waitFor(() => expect(authedRequest).toHaveBeenCalledOnce()); + await new Promise((resolve) => setTimeout(resolve, 0)); + manager.dispose(); + + expect(preprocessToDeviceMessages).not.toHaveBeenCalled(); + }); +}); diff --git a/src/client/presenceSync.ts b/src/client/presenceSync.ts index 4d2b9a6d8..684f328ea 100644 --- a/src/client/presenceSync.ts +++ b/src/client/presenceSync.ts @@ -1,13 +1,5 @@ -import type { CryptoBackend, IDeviceLists, IToDeviceEvent, MatrixClient } from '$types/matrix-sdk'; -import { - ClientEvent, - EventType, - Filter, - Method, - processToDeviceMessages, - SetPresence, - User, -} from '$types/matrix-sdk'; +import type { MatrixClient } from '$types/matrix-sdk'; +import { ClientEvent, EventType, Filter, Method, SetPresence, User } from '$types/matrix-sdk'; import { createDebugLogger } from '$utils/debugLogger'; const debugLog = createDebugLogger('presenceSync'); @@ -15,11 +7,6 @@ const debugLog = createDebugLogger('presenceSync'); type PresenceSyncResponse = { next_batch?: string; presence?: { events?: unknown[] }; - to_device?: { events?: IToDeviceEvent[] }; - device_lists?: IDeviceLists; - device_one_time_keys_count?: Record; - device_unused_fallback_key_types?: string[]; - 'org.matrix.msc2732.device_unused_fallback_key_types'?: string[]; }; export class PresenceSyncManager { @@ -76,28 +63,6 @@ export class PresenceSyncManager { this.abortController?.abort(); } - private async processCrypto(response: PresenceSyncResponse): Promise { - const crypto = this.mx.getCrypto() as CryptoBackend | undefined; - if (!crypto) return; - - const toDeviceEvents = response.to_device?.events ?? []; - if (toDeviceEvents.length > 0) { - const processedEvents = await crypto.preprocessToDeviceMessages(toDeviceEvents); - processToDeviceMessages(processedEvents, this.mx); - } - - if (response.device_lists) { - await crypto.processDeviceLists(response.device_lists); - } - - await crypto.processKeyCounts( - response.device_one_time_keys_count, - response.device_unused_fallback_key_types ?? - response['org.matrix.msc2732.device_unused_fallback_key_types'] - ); - crypto.onSyncCompleted({ nextSyncToken: response.next_batch }); - } - private processPresence(response: PresenceSyncResponse): void { const events = response.presence?.events; if (!events || !Array.isArray(events)) return; @@ -155,14 +120,11 @@ export class PresenceSyncManager { { abortSignal: signal } ); - await this.processCrypto(response); this.processPresence(response); this.syncToken = response.next_batch; debugLog.info('sync', 'Presence sync response processed', { presenceEvents: response.presence?.events?.length ?? 0, - toDeviceEvents: response.to_device?.events?.length ?? 0, - hasKeyCounts: response.device_one_time_keys_count !== undefined, }); } catch (err) { if (!signal.aborted && !this.disposed) { diff --git a/src/client/slidingSync.test.ts b/src/client/slidingSync.test.ts index 691a05755..bfcee2a9c 100644 --- a/src/client/slidingSync.test.ts +++ b/src/client/slidingSync.test.ts @@ -832,6 +832,27 @@ describe('SlidingSyncManager room subscription coordination', () => { ); }); + it('keeps the space subscription when a space is first hydrated in the sidebar', async () => { + const manager = makeManager(makeMockMx()); + const roomId = '!space:example.com'; + const internals = manager as unknown as { + listsFullyLoaded: boolean; + initialListHydrationCompleted: boolean; + }; + internals.listsFullyLoaded = true; + internals.initialListHydrationCompleted = true; + manager.attach(); + + manager.setSpaceSubscriptions([roomId]); + fireRoomData(roomId, { initial: true }); + await Promise.resolve(); + + expect(mocks.slidingSyncInstance.useCustomSubscription).toHaveBeenLastCalledWith( + roomId, + 'space' + ); + }); + it('hydrates sidebar state once for rooms first seen after startup', async () => { const manager = makeManager(makeMockMx()); const roomId = '!new:example.com'; diff --git a/src/client/slidingSync.ts b/src/client/slidingSync.ts index 98836cc7d..58ebef3a3 100644 --- a/src/client/slidingSync.ts +++ b/src/client/slidingSync.ts @@ -1836,8 +1836,6 @@ export class SlidingSyncManager { this.slidingSync.useCustomSubscription(roomId, CALL_ROOM_SUBSCRIPTION_KEY); } else if (this.activeRoomSubscriptions.has(roomId)) { this.slidingSync.useCustomSubscription(roomId, ACTIVE_ROOM_SUBSCRIPTION_KEY); - } else if (this.sidebarRoomSubscriptions.has(roomId)) { - this.slidingSync.useCustomSubscription(roomId, SIDEBAR_ROOM_SUBSCRIPTION_KEY); } else if (this.spaceSubscriptions.has(roomId)) { this.slidingSync.useCustomSubscription( roomId, @@ -1845,6 +1843,10 @@ export class SlidingSyncManager { ? SPACE_IMAGE_PACK_SUBSCRIPTION_KEY : SPACE_SUBSCRIPTION_KEY ); + } else if (this.sidebarRoomSubscriptions.has(roomId)) { + // Spaces need their child state even when their initial list data also + // placed them in the lightweight sidebar subscription. + this.slidingSync.useCustomSubscription(roomId, SIDEBAR_ROOM_SUBSCRIPTION_KEY); } else { this.slidingSync.useCustomSubscription(roomId, IMAGE_PACK_SUBSCRIPTION_KEY); }