tuwunel_service/oauth/sessions/
adopt.rs

1use std::str::from_utf8;
2
3use futures::StreamExt;
4use tuwunel_core::{Result, err, implement, utils::random_string, warn};
5use tuwunel_database::Cbor;
6
7use super::{SESSION_ID_LENGTH, Session, Sessions};
8use crate::{
9	migrations::local_user_id,
10	oauth::{Provider, UserInfo, unique_id_sub},
11};
12
13/// Results from adopting provider subjects stored by another database.
14///
15/// The counters distinguish new writes from safe skips. A missing source
16/// column is reported separately from rows that could not be interpreted.
17#[derive(Clone, Copy, Debug, Default)]
18pub struct Counts {
19	/// New durable associations written.
20	pub adopted: usize,
21
22	/// Associations that already resolve to the intended user.
23	pub already_bound: usize,
24
25	/// Associations left untouched because the identity key is occupied.
26	pub collision: usize,
27
28	/// Rows whose local account is absent or unusable.
29	pub absent: usize,
30
31	/// Rows whose subject or localpart is not valid UTF-8.
32	pub invalid: usize,
33
34	/// Whether the foreign identity column exists.
35	pub foreign_column: bool,
36
37	unreadable: usize,
38}
39
40#[derive(Clone, Copy)]
41enum Adoption {
42	Adopted,
43	AlreadyBound,
44	Collision,
45	Absent,
46	Invalid,
47}
48
49impl Counts {
50	fn tally(&mut self, result: Result<Adoption>) {
51		match result {
52			| Ok(Adoption::Adopted) => self.adopted = self.adopted.saturating_add(1),
53			| Ok(Adoption::Absent) => self.absent = self.absent.saturating_add(1),
54			| Ok(Adoption::Invalid) => self.invalid = self.invalid.saturating_add(1),
55			| Ok(Adoption::Collision) => self.collision = self.collision.saturating_add(1),
56			| Ok(Adoption::AlreadyBound) => {
57				self.already_bound = self.already_bound.saturating_add(1);
58			},
59			| Err(e) => {
60				warn!(error = %e, "a provider subject could not be read");
61				self.unreadable = self.unreadable.saturating_add(1);
62			},
63		}
64	}
65}
66
67#[implement(Sessions)]
68/// Adopts foreign provider subjects as durable, one-time session bridges.
69///
70/// Existing identity keys are never replaced. Each new association commits
71/// independently, so a read error leaves earlier writes available to an
72/// idempotent retry.
73#[tracing::instrument(level = "debug", skip(self, provider))]
74pub async fn adopt_foreign_subjects(&self, provider: &Provider) -> Result<Counts> {
75	unique_id_sub((provider, ""))?;
76
77	let Some(subjects) = self
78		.db
79		.database
80		.open_cf("openidsubject_localpart")?
81	else {
82		return Ok(Counts::default());
83	};
84
85	let server_name = self.services.globals.server_name();
86	let cork = self.db.database.cork_and_sync();
87	let counts = Counts {
88		foreign_column: true,
89		..Default::default()
90	};
91
92	let counts = subjects
93		.raw_stream()
94		.fold(counts, async |mut counts, row| {
95			let result = match row {
96				| Err(e) => Err(e),
97				| Ok((sub, localpart)) => match (from_utf8(sub), from_utf8(localpart)) {
98					| (Ok(sub), Ok(localpart)) =>
99						self.adopt_foreign_subject(provider, sub, localpart, server_name)
100							.await,
101
102					| _ => Ok(Adoption::Invalid),
103				},
104			};
105
106			counts.tally(result);
107			counts
108		})
109		.await;
110
111	drop(cork);
112
113	let unreadable = counts.unreadable;
114
115	unreadable
116		.eq(&0)
117		.then_some(counts)
118		.ok_or_else(|| err!(Database("{unreadable} provider subjects could not be read")))
119}
120
121#[implement(Sessions)]
122async fn adopt_foreign_subject(
123	&self,
124	provider: &Provider,
125	sub: &str,
126	localpart: &str,
127	server_name: &ruma::ServerName,
128) -> Result<Adoption> {
129	let Some(user_id) = local_user_id(localpart, server_name) else {
130		return Ok(Adoption::Absent);
131	};
132
133	if user_id == self.services.globals.server_user || !self.services.users.exists(&user_id).await
134	{
135		return Ok(Adoption::Absent);
136	}
137
138	let unique_id = unique_id_sub((provider, sub))?;
139	let _write_guard = self.write_locks.lock(&unique_id).await;
140	match self.get_sess_id_by_unique_id(&unique_id).await {
141		| Err(e) if e.is_not_found() => (),
142		| Err(e) => return Err(e),
143		| Ok(sess_id) => {
144			match self.get(&sess_id).await {
145				| Err(e) if !e.is_not_found() => return Err(e),
146				| Ok(session) if session.user_id.as_deref() == Some(&user_id) => {
147					return Ok(Adoption::AlreadyBound);
148				},
149				| Ok(_) | Err(_) => (),
150			}
151
152			warn!(%user_id, %unique_id, "provider subject association collides");
153			return Ok(Adoption::Collision);
154		},
155	}
156
157	let session = Session {
158		idp_id: Some(provider.id().to_owned()),
159		sess_id: Some(random_string(SESSION_ID_LENGTH)),
160		user_id: Some(user_id),
161		user_info: Some(UserInfo {
162			sub: sub.to_owned(),
163			..Default::default()
164		}),
165		..Default::default()
166	};
167
168	let sess_id = session
169		.sess_id
170		.as_deref()
171		.expect("adopted session id was just initialized");
172
173	let mut txn = self.db.database.txn();
174
175	txn.raw_put(&self.db.oauthid_session, sess_id, Cbor(&session));
176	txn.insert_raw(&self.db.oauthuniqid_oauthid, &unique_id, sess_id);
177
178	txn.execute();
179
180	Ok(Adoption::Adopted)
181}
tuwunel_service/oauth/sessions/adopt.rs

tuwunel_service/oauth/sessions/
adopt.rs
