tuwunel_service/migrations/
email_bindings.rs

1use futures::StreamExt;
2use ruma::{MilliSecondsSinceUnixEpoch, ServerName, thirdparty::Medium};
3use tuwunel_core::{Result, debug_warn, err, info, warn};
4
5use super::local_user_id;
6use crate::{Services, threepid::canonicalize_email};
7
8/// Adopts the email addresses a foreign database binds in a column of its own.
9///
10/// Some databases hold one address per localpart in their own column rather
11/// than in the third-party identifier store this server reads, so the binding
12/// is invisible here and the address looks unclaimed. Adopting it keeps the
13/// address answering for its owner, which is what an account authenticated by
14/// an identity provider carries instead of a password.
15///
16/// Addresses are taken one at a time because two foreign addresses can fold
17/// onto one canonical key, and a concurrent pass would clear both through the
18/// in-use check before either wrote the reverse row.
19pub(super) async fn migrate_email_bindings(services: &Services) -> Result {
20	let Some(localpart_email) = services.db.open_cf("localpart_email")? else {
21		return Ok(());
22	};
23
24	let server_name = services.globals.server_name();
25	let bound_at = MilliSecondsSinceUnixEpoch::now();
26	let cork = services.db.cork_and_sync();
27
28	let (adopted, skipped, unreadable) = localpart_email
29		.stream()
30		.fold((0_usize, 0_usize, 0_usize), async |acc, binding: Result<(&str, &str)>| {
31			let adopted = match binding {
32				| Err(e) => Err(e),
33				| Ok((localpart, address)) =>
34					adopt_one(services, server_name, bound_at, localpart, address).await,
35			};
36
37			tally_adoption(acc, adopted)
38		})
39		.await;
40
41	drop(cork);
42
43	match skipped {
44		| 0 if adopted > 0 => info!(%adopted, "Adopted email bindings from a foreign database"),
45		| 0 => (),
46		| _ => warn!(
47			%adopted,
48			%skipped,
49			"Adopted email bindings from a foreign database; some addresses were left behind"
50		),
51	}
52
53	// Leaving the marker unstamped is what makes a read failure recoverable: the
54	// pass is idempotent, so the next boot retries it whole.
55	unreadable
56		.eq(&0)
57		.then_some(())
58		.ok_or_else(|| err!(Database("{unreadable} email bindings could not be read")))
59}
60
61/// Binds one foreign row, reporting whether it produced a binding.
62///
63/// A `false` return is a row with nothing to bind: an unusable localpart or
64/// address, an account absent here, the server's own, or an address already
65/// held by a different account, which this store cannot represent twice. An
66/// error is a read that failed and must not be mistaken for any of them.
67async fn adopt_one(
68	services: &Services,
69	server_name: &ServerName,
70	bound_at: MilliSecondsSinceUnixEpoch,
71	localpart: &str,
72	address: &str,
73) -> Result<bool> {
74	let Some(user_id) = local_user_id(localpart, server_name) else {
75		debug_warn!(%localpart, "skipping an unusable localpart");
76		return Ok(false);
77	};
78
79	// A deactivated account still reserves its address, since neither server
80	// unhooks a binding on deactivation.
81	match services.db["userid_password"].get(&user_id).await {
82		| Ok(_) if user_id != services.globals.server_user => (),
83		| Ok(_) => return Ok(false),
84		| Err(e) if e.is_not_found() => return Ok(false),
85		| Err(e) => return Err(e),
86	}
87
88	let Ok(email_canon) = canonicalize_email(address) else {
89		debug_warn!(%localpart, "skipping an unusable address");
90		return Ok(false);
91	};
92
93	if services
94		.threepid
95		.bound_elsewhere(&user_id, &email_canon)
96		.await?
97	{
98		return Ok(false);
99	}
100
101	services
102		.threepid
103		.put_binding(&user_id, &email_canon, Medium::Email, bound_at, bound_at)
104		.await;
105
106	Ok(true)
107}
108
109fn tally_adoption(
110	(adopted, skipped, unreadable): (usize, usize, usize),
111	result: Result<bool>,
112) -> (usize, usize, usize) {
113	match result {
114		| Ok(true) => (adopted.saturating_add(1), skipped, unreadable),
115		| Ok(false) => (adopted, skipped.saturating_add(1), unreadable),
116		| Err(e) => {
117			warn!(error = %e, "an email binding could not be read");
118
119			(adopted, skipped, unreadable.saturating_add(1))
120		},
121	}
122}
tuwunel_service/migrations/email_bindings.rs

tuwunel_service/migrations/
email_bindings.rs
