1use super::{SqlxStore, entities::sso_provider::Model};
2use crate::{
3 model::{self, ActiveRow, SqlxModel},
4 schema::AuthSchema,
5};
6use alibi_core::{
7 AuthResult,
8 store::{SsoProvider, SsoProviderStore},
9};
10use async_trait::async_trait;
11#[async_trait]
12impl<S: AuthSchema> SsoProviderStore for SqlxStore<S>
13where
14 S::Account: crate::schema::SqlxAccountModel,
15{
16 async fn mark_sso_domain_verified(&self, id: &str, domain: &str) -> AuthResult<bool> {
17 let mut sql = crate::sql::Sql::with(self.exec().engine(), "UPDATE ");
18 sql.ident(Model::TABLE);
19 sql.push(" SET ");
20 sql.assign("domainVerified", true);
21 sql.push(" WHERE ");
22 sql.compare(Model::TABLE, "id", " = ", id);
23 sql.push(" AND ");
24 let field = self
25 .config()
26 .advanced
27 .database
28 .sso_provider
29 .as_ref()
30 .map_or("domain", |mapping| mapping.predicate_field("domain"));
31 sql.compare(Model::TABLE, field, " = ", domain);
32 self.exec().execute(sql).await.map(|changed| changed > 0)
33 }
34 async fn update_sso_provider(
35 &self,
36 id: &str,
37 provider_id: &str,
38 prepare: &alibi_core::store::SsoProviderUpdateWork,
39 ) -> AuthResult<Option<SsoProvider>> {
40 let tx = self.begin(true).await?;
41 let exec = crate::pool::Exec::tx(&tx);
42 let outcome = async {
43 let Some(row) = self.lock_sso_provider_with(exec, id, provider_id).await? else {
44 return Ok(None);
45 };
46 self.update_sso_provider_with(exec, prepare(&row)?).await
47 }
48 .await;
49 match outcome {
50 Ok(row) => {
51 tx.commit().await?;
52 Ok(row)
53 }
54 Err(error) => {
55 tx.rollback().await?;
56 Err(error)
57 }
58 }
59 }
60 async fn delete_sso_provider_and_accounts(
61 &self,
62 id: &str,
63 provider_id: &str,
64 ) -> AuthResult<bool> {
65 let tx = self.begin(true).await?;
66 let exec = crate::pool::Exec::tx(&tx);
67 let outcome = self.delete_sso_provider_with(exec, id, provider_id).await;
68 match outcome {
69 Ok(value) => {
70 tx.commit().await?;
71 Ok(value)
72 }
73 Err(error) => {
74 tx.rollback().await?;
75 Err(error)
76 }
77 }
78 }
79 async fn list_sso_providers(&self) -> AuthResult<Vec<SsoProvider>> {
80 let mut sql = provider_select(self.exec());
81 sql.push(" LIMIT ");
82 sql.bind(self.find_many_limit());
83 let rows = self.exec().fetch_all::<ProviderRow>(sql).await?;
84 project_rows(rows, self.exec()).await
85 }
86 async fn create_sso_provider(&self, row: SsoProvider) -> AuthResult<SsoProvider> {
87 let extra = prepare_fields(self.exec(), &row.additional_fields, true).await?;
88 let mut active = ActiveRow::new();
89 active.set("id", row.id);
90 active.set("issuer", row.issuer);
91 active.set("domain", row.domain);
92 active.set("providerId", row.provider_id);
93 active.set("userId", row.user_id);
94 active.set("organizationId", row.organization_id);
95 active.set("oidcConfig", row.oidc_config);
96 active.set("samlConfig", row.saml_config);
97 active.set("domainVerified", row.domain_verified);
98 insert_provider(self.exec(), &active, &extra).await
99 }
100 async fn get_sso_provider(&self, provider_id: &str) -> AuthResult<Option<SsoProvider>> {
101 let mut sql = provider_select(self.exec());
102 sql.push(" WHERE ");
103 sql.compare(Model::TABLE, "providerId", " = ", provider_id);
104 let row = self.exec().fetch_optional::<ProviderRow>(sql).await?;
105 project_row(row, self.exec()).await
106 }
107 async fn list_sso_providers_by_user(&self, user_id: &str) -> AuthResult<Vec<SsoProvider>> {
108 let mut sql = provider_select(self.exec());
109 sql.push(" WHERE ");
110 sql.compare(Model::TABLE, "userId", " = ", user_id);
111 sql.push(" LIMIT ");
112 sql.bind(self.find_many_limit());
113 let rows = self.exec().fetch_all::<ProviderRow>(sql).await?;
114 project_rows(rows, self.exec()).await
115 }
116}
117
118impl<S: AuthSchema> SqlxStore<S>
119where
120 S::Account: crate::schema::SqlxAccountModel,
121{
122 pub(super) async fn lock_sso_provider_with(
123 &self,
124 exec: crate::pool::Exec<'_>,
125 id: &str,
126 provider_id: &str,
127 ) -> AuthResult<Option<SsoProvider>> {
128 let mut lock = crate::sql::Sql::with(exec.engine(), "UPDATE ");
129 lock.ident(Model::TABLE);
130 lock.push(" SET ");
131 lock.assign("providerId", provider_id);
132 lock.push(" WHERE ");
133 lock.compare(Model::TABLE, "id", " = ", id);
134 lock.push(" AND ");
135 lock.compare(Model::TABLE, "providerId", " = ", provider_id);
136 if exec.execute(lock).await? == 0 {
137 return Ok(None);
138 }
139 let mut read = provider_select(exec);
140 read.push(" WHERE ");
141 read.compare(Model::TABLE, "id", " = ", id);
142 let row: SsoProvider = exec
143 .fetch_optional::<ProviderRow>(read)
144 .await?
145 .ok_or_else(|| alibi_core::AuthError::not_found("Provider not found"))?
146 .project(exec)
147 .await?;
148 Ok(Some(row))
149 }
150 pub(super) async fn update_sso_provider_with(
151 &self,
152 exec: crate::pool::Exec<'_>,
153 update: alibi_core::store::SsoProviderUpdate,
154 ) -> AuthResult<Option<SsoProvider>> {
155 use crate::schema::SqlxAccountModel;
156 let id = update.provider.id.clone();
157 let id = id.as_str();
158 let provider_id = update.provider.provider_id.clone();
159 let provider_id = provider_id.as_str();
160 if update.identity_changed {
161 let mut accounts = model::select_model::<S::Account>(exec);
162 accounts.push(" WHERE ");
163 accounts.compare_model::<S::Account>(
164 <S::Account as SqlxModel>::TABLE,
165 S::Account::provider_id_column(),
166 " = ",
167 provider_id,
168 );
169 accounts.push(" LIMIT 1");
170 if exec.fetch_optional::<S::Account>(accounts).await?.is_some() {
171 return Err(alibi_core::AuthError::Api {
172 status: 409,
173 code: None,
174 message:
175 "Cannot change SSO provider identity fields while linked accounts exist"
176 .into(),
177 });
178 }
179 }
180 let extra = prepare_fields(exec, &update.additional_field_changes, false).await?;
181 let mut active = ActiveRow::new();
182 active.set("id", id);
183 active.set("issuer", update.provider.issuer);
184 active.set("domain", update.provider.domain);
185 active.set("oidcConfig", update.provider.oidc_config);
186 active.set("samlConfig", update.provider.saml_config);
187 active.set("domainVerified", update.provider.domain_verified);
188 update_provider(exec, &active, &extra).await
189 }
190 pub(super) async fn delete_sso_provider_with(
191 &self,
192 exec: crate::pool::Exec<'_>,
193 id: &str,
194 provider_id: &str,
195 ) -> AuthResult<bool> {
196 use crate::schema::SqlxAccountModel;
197 if self
198 .lock_sso_provider_with(exec, id, provider_id)
199 .await?
200 .is_none()
201 {
202 return Ok(false);
203 }
204 let mut accounts = crate::sql::Sql::with(exec.engine(), "DELETE FROM ");
205 accounts.ident(<S::Account as SqlxModel>::TABLE);
206 accounts.push(" WHERE ");
207 accounts.compare_model::<S::Account>(
208 <S::Account as SqlxModel>::TABLE,
209 S::Account::provider_id_column(),
210 " = ",
211 provider_id,
212 );
213 _ = exec.execute(accounts).await?;
214 let mut provider = crate::sql::Sql::with(exec.engine(), "DELETE FROM ");
215 provider.ident(Model::TABLE);
216 provider.push(" WHERE ");
217 provider.compare(Model::TABLE, "id", " = ", id);
218 provider.push(" AND ");
219 provider.compare(Model::TABLE, "providerId", " = ", provider_id);
220 _ = exec.execute(provider).await?;
221 Ok(true)
222 }
223}
224
225#[derive(sqlx::FromRow)]
226struct ProviderRow {
227 #[sqlx(flatten)]
228 model: Model,
229 __alibi_sso_fields: String,
230}
231impl ProviderRow {
232 async fn project(self, exec: crate::pool::Exec<'_>) -> AuthResult<SsoProvider> {
233 let provider = self.into_provider()?;
234 match exec.sso_fields() {
235 Some(fields) => alibi_core::store::project_sso_provider(fields, provider).await,
236 None => Ok(provider),
237 }
238 }
239 fn into_provider(self) -> AuthResult<SsoProvider> {
240 let mut provider: SsoProvider = self.model.into();
241 provider.additional_fields = serde_json::from_str(&self.__alibi_sso_fields)?;
242 Ok(provider)
243 }
244}
245fn provider_columns(
246 sql: &mut crate::sql::Sql,
247 qualified: bool,
248 fields: Option<&alibi_core::field_policy::SessionFields>,
249) {
250 use alibi_core::store::{SsoFieldType, sso_field_column};
251 for (index, column) in Model::COLUMN_NAMES.iter().enumerate() {
252 if index > 0 {
253 sql.push(", ");
254 }
255 if qualified {
256 sql.column(Model::TABLE, column);
257 } else {
258 sql.ident(column);
259 }
260 sql.push(" AS ");
261 sql.ident(column);
262 }
263 sql.push(", CAST(");
264 let postgres = sql.engine() == crate::pool::Engine::Postgres;
265 sql.push(if postgres {
266 "jsonb_build_object("
267 } else {
268 "json_object("
269 });
270 if let Some(fields) = fields {
271 for (index, (name, field)) in fields.0.iter().enumerate() {
272 if index > 0 {
273 sql.push(", ");
274 }
275 sql.bind(name);
276 sql.push(", ");
277 let kind = SsoFieldType::from_config(field);
278 let emit = |sql: &mut crate::sql::Sql| {
279 let column = sso_field_column(name, field);
280 if qualified {
281 sql.column(Model::TABLE, column);
282 } else {
283 sql.ident(column);
284 }
285 };
286 if !postgres && kind == SsoFieldType::Boolean {
287 sql.push("json(CASE WHEN ");
288 emit(sql);
289 sql.push(" IS NULL THEN 'null' WHEN ");
290 emit(sql);
291 sql.push(" THEN 'true' ELSE 'false' END)");
292 } else if !postgres && kind == SsoFieldType::Json {
293 sql.push("json(");
294 emit(sql);
295 sql.push(")");
296 } else {
297 emit(sql);
298 }
299 }
300 }
301 sql.push(") AS TEXT) AS ");
302 sql.ident("__alibi_sso_fields");
303}
304fn provider_select(exec: crate::pool::Exec<'_>) -> crate::sql::Sql {
305 let mut sql = crate::sql::Sql::with(exec.engine(), "SELECT ");
306 provider_columns(&mut sql, true, exec.sso_fields());
307 sql.push(" FROM ");
308 sql.ident(Model::TABLE);
309 sql
310}
311async fn extra_bindings(
312 exec: crate::pool::Exec<'_>,
313 extra: &serde_json::Map<String, serde_json::Value>,
314) -> AuthResult<Vec<(String, crate::value::SqlValue)>> {
315 use crate::value::ColumnKind;
316 use alibi_core::store::{SsoFieldType, sso_field_column};
317 let mut values = Vec::new();
318 if let Some(fields) = exec.sso_fields() {
319 for (name, field) in &fields.0 {
320 if let Some(value) = extra.get(name) {
321 if SsoFieldType::from_config(field) == SsoFieldType::Date {
322 let date = value
323 .as_str()
324 .map(|value| {
325 chrono::DateTime::parse_from_rfc3339(value)
326 .map(|date| date.with_timezone(&chrono::Utc))
327 })
328 .transpose()
329 .map_err(|error| alibi_core::AuthError::internal(error.to_string()))?;
330 values.push((
331 sso_field_column(name, field).to_owned(),
332 crate::value::SqlValue::Timestamp(date),
333 ));
334 continue;
335 }
336 let kind = match SsoFieldType::from_config(field) {
337 SsoFieldType::String => ColumnKind::Text,
338 SsoFieldType::Boolean => ColumnKind::Boolean,
339 SsoFieldType::Number => ColumnKind::Number,
340 SsoFieldType::Json => ColumnKind::Json,
341 SsoFieldType::Date => ColumnKind::Other,
342 };
343 let raw = crate::additional_fields::raw_value(
344 &alibi_core::utils::json::JsValue::from(value.clone()),
345 )?;
346 values.push((
347 sso_field_column(name, field).to_owned(),
348 crate::additional_fields::prepare_value(exec, kind, raw).await?,
349 ));
350 }
351 }
352 }
353 Ok(values)
354}
355async fn insert_provider(
356 exec: crate::pool::Exec<'_>,
357 active: &ActiveRow,
358 extra: &serde_json::Map<String, serde_json::Value>,
359) -> AuthResult<SsoProvider> {
360 let mut sql = crate::sql::Sql::with(exec.engine(), "INSERT INTO ");
361 sql.ident(Model::TABLE);
362 sql.push(" (");
363 let mut fields = active
364 .present()
365 .map(|(column, value)| (column.to_owned(), value.clone()))
366 .collect::<Vec<_>>();
367 fields.extend(extra_bindings(exec, extra).await?);
368 sql.column_list(
369 &fields
370 .iter()
371 .map(|(column, _)| column.as_str())
372 .collect::<Vec<_>>(),
373 );
374 sql.push(") VALUES (");
375 for (index, (column, value)) in fields.into_iter().enumerate() {
376 if index > 0 {
377 sql.push(", ");
378 }
379 if Model::COLUMN_NAMES.contains(&column.as_str()) {
380 model::bind_model_value::<Model>(&mut sql, &column, value);
381 } else {
382 sql.bind(value);
383 }
384 }
385 sql.push(") RETURNING ");
386 provider_columns(&mut sql, false, exec.sso_fields());
387 exec.fetch_optional::<ProviderRow>(sql)
388 .await?
389 .ok_or_else(|| alibi_core::AuthError::internal("Provider was not inserted"))?
390 .project(exec)
391 .await
392}
393async fn update_provider(
394 exec: crate::pool::Exec<'_>,
395 active: &ActiveRow,
396 extra: &serde_json::Map<String, serde_json::Value>,
397) -> AuthResult<Option<SsoProvider>> {
398 let id = active
399 .get("id")
400 .and_then(crate::model::ActiveValue::value)
401 .cloned()
402 .ok_or_else(|| {
403 alibi_core::AuthError::internal("Provider update requires its primary key")
404 })?;
405 let mut sql = crate::sql::Sql::with(exec.engine(), "UPDATE ");
406 sql.ident(Model::TABLE);
407 sql.push(" SET ");
408 let mut fields = active
409 .changed()
410 .map(|(column, value)| (column.to_owned(), value.clone()))
411 .collect::<Vec<_>>();
412 fields.extend(extra_bindings(exec, extra).await?);
413 for (index, (column, value)) in fields.into_iter().enumerate() {
414 if index > 0 {
415 sql.push(", ");
416 }
417 sql.ident(&column);
418 sql.push(" = ");
419 if Model::COLUMN_NAMES.contains(&column.as_str()) {
420 model::bind_model_value::<Model>(&mut sql, &column, value);
421 } else {
422 sql.bind(value);
423 }
424 }
425 sql.push(" WHERE ");
426 sql.compare(Model::TABLE, "id", " = ", id);
427 sql.push(" RETURNING ");
428 provider_columns(&mut sql, false, exec.sso_fields());
429 let row = exec.fetch_optional::<ProviderRow>(sql).await?;
430 project_row(row, exec).await
431}
432
433async fn project_row(
434 row: Option<ProviderRow>,
435 exec: crate::pool::Exec<'_>,
436) -> AuthResult<Option<SsoProvider>> {
437 match row {
438 Some(row) => Ok(Some(row.project(exec).await?)),
439 None => Ok(None),
440 }
441}
442async fn project_rows(
443 rows: Vec<ProviderRow>,
444 exec: crate::pool::Exec<'_>,
445) -> AuthResult<Vec<SsoProvider>> {
446 let mut providers = Vec::with_capacity(rows.len());
447 for row in rows {
448 providers.push(row.project(exec).await?);
449 }
450 Ok(providers)
451}
452async fn prepare_fields(
453 exec: crate::pool::Exec<'_>,
454 fields: &serde_json::Map<String, serde_json::Value>,
455 create: bool,
456) -> AuthResult<serde_json::Map<String, serde_json::Value>> {
457 match exec.sso_fields() {
458 Some(policy) => alibi_core::store::prepare_sso_fields(policy, fields, create).await,
459 None => Ok(fields.clone()),
460 }
461}