macltok

public

Star0

Watch0

23 commits 3 branches 0 tags 0 releases3.0 MiB

use std::{future::Future, panic::AssertUnwindSafe};

use futures_util::FutureExt;
use matrix_sdk::encryption::verification::VerificationRequest;
use matrix_sdk::{
    Client,
    notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
    ruma::{
        OwnedDeviceId, OwnedEventId, RoomId, TransactionId,
        api::client::receipt::create_receipt::v3::ReceiptType, events::receipt::ReceiptThread,
    },
    send_queue::{LocalEchoContent, SendHandle},
};
use tokio::sync::mpsc;

use crate::{
    EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
    NotificationCommand, RoomNotificationMode, TimelineId,
    encryption::VerificationController,
    transport::{EventSender, Responder},
};

pub(super) struct Job<T> {
    pub(super) command: T,
    pub(super) reply: Responder,
}

/// a panicking command answers with an error instead of taking its worker down with it
pub(super) async fn guarded<T>(
    label: &'static str,
    work: impl Future<Output = Result<T, Error>>,
) -> Result<T, Error> {
    AssertUnwindSafe(work)
        .catch_unwind()
        .await
        .unwrap_or_else(|_| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))
}

pub(super) async fn run_notifications(
    client: Client,
    mut jobs: mpsc::Receiver<Job<NotificationCommand>>,
) {
    while let Some(Job { command, reply }) = jobs.recv().await {
        reply.done(guarded("notification command", notification(&client, command)).await);
    }
}

pub(super) async fn run_encryption(
    client: Client,
    events: EventSender,
    mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
    mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,
) {
    let mut verification = VerificationController::default();
    loop {
        tokio::select! {
            job = jobs.recv() => {
                let Some(Job { command, reply }) = job else { return };
                let work = encryption(&client, &events, &mut verification, command);
                reply.done(guarded("encryption command", work).await);
            }
            Some(request) = incoming.recv() => {
                verification.adopt_incoming(events.clone(), request).await;
            }
        }
    }
}

async fn encryption(
    client: &Client,
    events: &EventSender,
    verification: &mut VerificationController,
    command: EncryptionCommand,
) -> Result<(), Error> {
    let encryption = client.encryption();
    match command {
        EncryptionCommand::Recover(key) => encryption
            .recovery()
            .recover(&key)
            .await
            .map_err(|source| Error::from_recovery("account recovery failed", source)),
        EncryptionCommand::RecoverKeyBackup(key) => encryption
            .recovery()
            .recover_and_fix_backup(&key)
            .await
            .map_err(|source| Error::from_recovery("recovering key backup failed", source)),
        EncryptionCommand::CreateKeyBackup => encryption
            .backups()
            .create()
            .await
            .map_err(|source| Error::from_sdk("creating key backup failed", source)),
        EncryptionCommand::EnableKeyBackup => encryption
            .recovery()
            .enable_backup()
            .await
            .map_err(|source| Error::from_recovery("enabling key backup failed", source)),
        EncryptionCommand::DisableKeyBackup { delete_from_server } => {
            let backups = encryption.backups();
            if delete_from_server {
                backups.disable_and_delete().await
            } else {
                backups.disable().await
            }
            .map_err(|source| Error::from_sdk("disabling key backup failed", source))
        }
        EncryptionCommand::StartVerification(target) => {
            verification.start(client, events.clone(), target).await
        }
        EncryptionCommand::Verify(action) => verification.apply(events.clone(), action).await,
    }
}

async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {
    let settings = client.notification_settings().await;
    match command {
        NotificationCommand::SetRoomMode { room_id, mode } => {
            settings
                .set_room_notification_mode(&room_id, sdk_mode(mode))
                .await
        }
        NotificationCommand::ResetRoomMode(room_id) => {
            settings.delete_user_defined_room_rules(&room_id).await
        }
        NotificationCommand::SetDefaultMode {
            encrypted,
            one_to_one,
            mode,
        } => {
            settings
                .set_default_room_notification_mode(
                    IsEncrypted::from(encrypted),
                    IsOneToOne::from(one_to_one),
                    sdk_mode(mode),
                )
                .await
        }
        NotificationCommand::AddKeyword(keyword) => settings.add_keyword(keyword).await,
        NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
    }
    .map_err(|source| {
        Error::with_source(
            ErrorKind::Server,
            "updating notification settings failed",
            source,
        )
    })
}

const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {
    match mode {
        RoomNotificationMode::AllMessages => SdkMode::AllMessages,
        RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,
        RoomNotificationMode::Mute => SdkMode::Mute,
    }
}

fn room(client: &Client, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {
    client.get_room(room_id).ok_or_else(Error::room_unknown)
}

/// the closed-room path. without a timeline nothing stops the marker going backwards
pub(super) async fn mark_read(
    client: &Client,
    timeline_id: &TimelineId,
    event_id: OwnedEventId,
) -> Result<(), Error> {
    let room = room(client, timeline_id.room_id())?;
    let receipt_failed = |source| Error::from_sdk("sending read receipt failed", source);
    let thread = match timeline_id.thread_root() {
        None => ReceiptThread::Main,
        Some(root) => ReceiptThread::Thread(root.clone()),
    };
    room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())
        .await
        .map_err(receipt_failed)?;
    if timeline_id.thread_root().is_none() {
        room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)
            .await
            .map_err(receipt_failed)?;
    }
    Ok(())
}

pub(super) async fn set_typing(
    client: &Client,
    room_id: &RoomId,
    typing: bool,
) -> Result<(), Error> {
    room(client, room_id)?
        .typing_notice(typing)
        .await
        .map_err(|source| Error::from_sdk("typing update failed", source))
}

async fn send_handle(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<SendHandle, Error> {
    let (echoes, _) = room(client, room_id)?
        .send_queue()
        .subscribe()
        .await
        .map_err(|source| {
            Error::with_source(ErrorKind::Store, "reading send queue failed", source)
        })?;
    echoes
        .into_iter()
        .find_map(|echo| match echo.content {
            LocalEchoContent::Event { send_handle, .. }
                if echo.transaction_id.as_str() == transaction_id.as_str() =>
            {
                Some(send_handle)
            }
            _ => None,
        })
        .ok_or_else(|| {
            Error::new(
                ErrorKind::Invalid,
                "send is no longer pending or is not a message",
            )
        })
}

pub(super) async fn cancel_send(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<(), Error> {
    match send_handle(client, room_id, transaction_id)
        .await?
        .abort()
        .await
    {
        Ok(true) => Ok(()),
        Ok(false) => Err(Error::new(
            ErrorKind::Invalid,
            "send could not be cancelled because it has already been sent",
        )),
        Err(source) => Err(Error::with_source(
            ErrorKind::Store,
            "cancelling send failed",
            source,
        )),
    }
}

pub(super) async fn retry_send(
    client: &Client,
    room_id: &RoomId,
    transaction_id: &TransactionId,
) -> Result<(), Error> {
    send_handle(client, room_id, transaction_id)
        .await?
        .unwedge()
        .await
        .map_err(|source| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;
    room(client, room_id)?.send_queue().set_enabled(true);
    Ok(())
}

pub(super) async fn upload_avatar(
    client: &Client,
    mimetype: &str,
    bytes: Vec<u8>,
    max_bytes: usize,
) -> Result<(), Error> {
    if bytes.len() > max_bytes {
        return Err(Error::new(
            ErrorKind::MediaTooLarge,
            "avatar exceeds the configured byte limit",
        ));
    }
    let mimetype: mime::Mime = mimetype
        .parse()
        .map_err(|_| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;
    if mimetype.type_() != mime::IMAGE {
        return Err(Error::new(
            ErrorKind::Invalid,
            "an avatar has to be an image",
        ));
    }
    client
        .account()
        .upload_avatar(&mimetype, bytes)
        .await
        .map(drop)
        .map_err(|source| Error::from_sdk("uploading avatar failed", source))
}

pub(super) async fn rename_device(
    client: &Client,
    device_id: &OwnedDeviceId,
    display_name: &str,
) -> Result<(), Error> {
    client
        .rename_device(device_id, display_name)
        .await
        .map(drop)
        .map_err(|source| Error::from_http("renaming device failed", source))
}

pub(super) async fn delete_devices(
    client: &Client,
    device_ids: &[OwnedDeviceId],
    auth: Option<InteractiveAuth>,
) -> Result<(), Error> {
    let user_id = client
        .user_id()
        .ok_or_else(|| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;
    let auth = auth
        .map(|auth| auth.into_sdk(user_id))
        .transpose()
        .map_err(|message| {
            Error::new(
                ErrorKind::Invalid,
                format!("interactive authentication data is invalid, {message}"),
            )
        })?;
    match client.delete_devices(device_ids, auth).await {
        Ok(_) => Ok(()),
        Err(source) => match source.as_uiaa_response() {
            Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(
                info,
            ))),
            None => Err(Error::from_http("deleting devices failed", source)),
        },
    }
}

1	use std::{future::Future, panic::AssertUnwindSafe};
2
3	use futures_util::FutureExt;
4	use matrix_sdk::encryption::verification::VerificationRequest;
5	use matrix_sdk::{
6	Client,
7	notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
8	ruma::{
9	OwnedDeviceId, OwnedEventId, RoomId, TransactionId,
10	api::client::receipt::create_receipt::v3::ReceiptType, events::receipt::ReceiptThread,
11	},
12	send_queue::{LocalEchoContent, SendHandle},
13	};
14	use tokio::sync::mpsc;
15
16	use crate::{
17	EncryptionCommand, Error, ErrorKind, InteractiveAuth, InteractiveAuthChallenge,
18	NotificationCommand, RoomNotificationMode, TimelineId,
19	encryption::VerificationController,
20	transport::{EventSender, Responder},
21	};
22
23	pub(super) struct Job<T> {
24	pub(super) command: T,
25	pub(super) reply: Responder,
26	}
27
28	/// a panicking command answers with an error instead of taking its worker down with it
29	pub(super) async fn guarded<T>(
30	label: &'static str,
31	work: impl Future<Output = Result<T, Error>>,
32	) -> Result<T, Error> {
33	AssertUnwindSafe(work)
34	.catch_unwind()
35	.await
36	.unwrap_or_else(\|_\| Err(Error::new(ErrorKind::Other, format!("{label} panicked"))))
37	}
38
39	pub(super) async fn run_notifications(
40	client: Client,
41	mut jobs: mpsc::Receiver<Job<NotificationCommand>>,
42	) {
43	while let Some(Job { command, reply }) = jobs.recv().await {
44	reply.done(guarded("notification command", notification(&client, command)).await);
45	}
46	}
47
48	pub(super) async fn run_encryption(
49	client: Client,
50	events: EventSender,
51	mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
52	mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,
53	) {
54	let mut verification = VerificationController::default();
55	loop {
56	tokio::select! {
57	job = jobs.recv() => {
58	let Some(Job { command, reply }) = job else { return };
59	let work = encryption(&client, &events, &mut verification, command);
60	reply.done(guarded("encryption command", work).await);
61	}
62	Some(request) = incoming.recv() => {
63	verification.adopt_incoming(events.clone(), request).await;
64	}
65	}
66	}
67	}
68
69	async fn encryption(
70	client: &Client,
71	events: &EventSender,
72	verification: &mut VerificationController,
73	command: EncryptionCommand,
74	) -> Result<(), Error> {
75	let encryption = client.encryption();
76	match command {
77	EncryptionCommand::Recover(key) => encryption
78	.recovery()
79	.recover(&key)
80	.await
81	.map_err(\|source\| Error::from_recovery("account recovery failed", source)),
82	EncryptionCommand::RecoverKeyBackup(key) => encryption
83	.recovery()
84	.recover_and_fix_backup(&key)
85	.await
86	.map_err(\|source\| Error::from_recovery("recovering key backup failed", source)),
87	EncryptionCommand::CreateKeyBackup => encryption
88	.backups()
89	.create()
90	.await
91	.map_err(\|source\| Error::from_sdk("creating key backup failed", source)),
92	EncryptionCommand::EnableKeyBackup => encryption
93	.recovery()
94	.enable_backup()
95	.await
96	.map_err(\|source\| Error::from_recovery("enabling key backup failed", source)),
97	EncryptionCommand::DisableKeyBackup { delete_from_server } => {
98	let backups = encryption.backups();
99	if delete_from_server {
100	backups.disable_and_delete().await
101	} else {
102	backups.disable().await
103	}
104	.map_err(\|source\| Error::from_sdk("disabling key backup failed", source))
105	}
106	EncryptionCommand::StartVerification(target) => {
107	verification.start(client, events.clone(), target).await
108	}
109	EncryptionCommand::Verify(action) => verification.apply(events.clone(), action).await,
110	}
111	}
112
113	async fn notification(client: &Client, command: NotificationCommand) -> Result<(), Error> {
114	let settings = client.notification_settings().await;
115	match command {
116	NotificationCommand::SetRoomMode { room_id, mode } => {
117	settings
118	.set_room_notification_mode(&room_id, sdk_mode(mode))
119	.await
120	}
121	NotificationCommand::ResetRoomMode(room_id) => {
122	settings.delete_user_defined_room_rules(&room_id).await
123	}
124	NotificationCommand::SetDefaultMode {
125	encrypted,
126	one_to_one,
127	mode,
128	} => {
129	settings
130	.set_default_room_notification_mode(
131	IsEncrypted::from(encrypted),
132	IsOneToOne::from(one_to_one),
133	sdk_mode(mode),
134	)
135	.await
136	}
137	NotificationCommand::AddKeyword(keyword) => settings.add_keyword(keyword).await,
138	NotificationCommand::RemoveKeyword(keyword) => settings.remove_keyword(&keyword).await,
139	}
140	.map_err(\|source\| {
141	Error::with_source(
142	ErrorKind::Server,
143	"updating notification settings failed",
144	source,
145	)
146	})
147	}
148
149	const fn sdk_mode(mode: RoomNotificationMode) -> SdkMode {
150	match mode {
151	RoomNotificationMode::AllMessages => SdkMode::AllMessages,
152	RoomNotificationMode::MentionsAndKeywordsOnly => SdkMode::MentionsAndKeywordsOnly,
153	RoomNotificationMode::Mute => SdkMode::Mute,
154	}
155	}
156
157	fn room(client: &Client, room_id: &RoomId) -> Result<matrix_sdk::Room, Error> {
158	client.get_room(room_id).ok_or_else(Error::room_unknown)
159	}
160
161	/// the closed-room path. without a timeline nothing stops the marker going backwards
162	pub(super) async fn mark_read(
163	client: &Client,
164	timeline_id: &TimelineId,
165	event_id: OwnedEventId,
166	) -> Result<(), Error> {
167	let room = room(client, timeline_id.room_id())?;
168	let receipt_failed = \|source\| Error::from_sdk("sending read receipt failed", source);
169	let thread = match timeline_id.thread_root() {
170	None => ReceiptThread::Main,
171	Some(root) => ReceiptThread::Thread(root.clone()),
172	};
173	room.send_single_receipt(ReceiptType::Read, thread, event_id.clone())
174	.await
175	.map_err(receipt_failed)?;
176	if timeline_id.thread_root().is_none() {
177	room.send_single_receipt(ReceiptType::FullyRead, ReceiptThread::Unthreaded, event_id)
178	.await
179	.map_err(receipt_failed)?;
180	}
181	Ok(())
182	}
183
184	pub(super) async fn set_typing(
185	client: &Client,
186	room_id: &RoomId,
187	typing: bool,
188	) -> Result<(), Error> {
189	room(client, room_id)?
190	.typing_notice(typing)
191	.await
192	.map_err(\|source\| Error::from_sdk("typing update failed", source))
193	}
194
195	async fn send_handle(
196	client: &Client,
197	room_id: &RoomId,
198	transaction_id: &TransactionId,
199	) -> Result<SendHandle, Error> {
200	let (echoes, _) = room(client, room_id)?
201	.send_queue()
202	.subscribe()
203	.await
204	.map_err(\|source\| {
205	Error::with_source(ErrorKind::Store, "reading send queue failed", source)
206	})?;
207	echoes
208	.into_iter()
209	.find_map(\|echo\| match echo.content {
210	LocalEchoContent::Event { send_handle, .. }
211	if echo.transaction_id.as_str() == transaction_id.as_str() =>
212	{
213	Some(send_handle)
214	}
215	_ => None,
216	})
217	.ok_or_else(\|\| {
218	Error::new(
219	ErrorKind::Invalid,
220	"send is no longer pending or is not a message",
221	)
222	})
223	}
224
225	pub(super) async fn cancel_send(
226	client: &Client,
227	room_id: &RoomId,
228	transaction_id: &TransactionId,
229	) -> Result<(), Error> {
230	match send_handle(client, room_id, transaction_id)
231	.await?
232	.abort()
233	.await
234	{
235	Ok(true) => Ok(()),
236	Ok(false) => Err(Error::new(
237	ErrorKind::Invalid,
238	"send could not be cancelled because it has already been sent",
239	)),
240	Err(source) => Err(Error::with_source(
241	ErrorKind::Store,
242	"cancelling send failed",
243	source,
244	)),
245	}
246	}
247
248	pub(super) async fn retry_send(
249	client: &Client,
250	room_id: &RoomId,
251	transaction_id: &TransactionId,
252	) -> Result<(), Error> {
253	send_handle(client, room_id, transaction_id)
254	.await?
255	.unwedge()
256	.await
257	.map_err(\|source\| Error::with_source(ErrorKind::Store, "retrying send failed", source))?;
258	room(client, room_id)?.send_queue().set_enabled(true);
259	Ok(())
260	}
261
262	pub(super) async fn upload_avatar(
263	client: &Client,
264	mimetype: &str,
265	bytes: Vec<u8>,
266	max_bytes: usize,
267	) -> Result<(), Error> {
268	if bytes.len() > max_bytes {
269	return Err(Error::new(
270	ErrorKind::MediaTooLarge,
271	"avatar exceeds the configured byte limit",
272	));
273	}
274	let mimetype: mime::Mime = mimetype
275	.parse()
276	.map_err(\|_\| Error::new(ErrorKind::Invalid, "avatar MIME type is invalid"))?;
277	if mimetype.type_() != mime::IMAGE {
278	return Err(Error::new(
279	ErrorKind::Invalid,
280	"an avatar has to be an image",
281	));
282	}
283	client
284	.account()
285	.upload_avatar(&mimetype, bytes)
286	.await
287	.map(drop)
288	.map_err(\|source\| Error::from_sdk("uploading avatar failed", source))
289	}
290
291	pub(super) async fn rename_device(
292	client: &Client,
293	device_id: &OwnedDeviceId,
294	display_name: &str,
295	) -> Result<(), Error> {
296	client
297	.rename_device(device_id, display_name)
298	.await
299	.map(drop)
300	.map_err(\|source\| Error::from_http("renaming device failed", source))
301	}
302
303	pub(super) async fn delete_devices(
304	client: &Client,
305	device_ids: &[OwnedDeviceId],
306	auth: Option<InteractiveAuth>,
307	) -> Result<(), Error> {
308	let user_id = client
309	.user_id()
310	.ok_or_else(\|\| Error::new(ErrorKind::Unauthenticated, "the client is not logged in"))?;
311	let auth = auth
312	.map(\|auth\| auth.into_sdk(user_id))
313	.transpose()
314	.map_err(\|message\| {
315	Error::new(
316	ErrorKind::Invalid,
317	format!("interactive authentication data is invalid, {message}"),
318	)
319	})?;
320	match client.delete_devices(device_ids, auth).await {
321	Ok(_) => Ok(()),
322	Err(source) => match source.as_uiaa_response() {
323	Some(info) => Err(Error::interactive_auth(InteractiveAuthChallenge::from(
324	info,
325	))),
326	None => Err(Error::from_http("deleting devices failed", source)),
327	},
328	}
329	}